datasync / yoho-datasync-core · Commits

Go to a project

Authored by liangyi.chen@yoho.cn
Commit 289715eb0210b18570edbd17fe9efbcbaff22094 0 parents

Create

Inline Side-by-side
Showing 23 changed files with 1084 additions and 0 deletions
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="CompilerConfiguration">
<annotationProcessing>
<profile name="Maven default annotation processors profile" enabled="true">
<sourceOutputDir name="target/generated-sources/annotations" />
<sourceTestOutputDir name="target/generated-test-sources/test-annotations" />
<outputRelativeToContentRoot value="true" />
<module name="yoho-datasync-core-base" />
</profile>
</annotationProcessing>
<bytecodeTargetLevel>
<module name="core" target="1.8" />
</bytecodeTargetLevel>
</component>
</project>
\ No newline at end of file
... ...
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ExternalStorageConfigurationManager" enabled="true" />
<component name="MavenProjectsManager">
<option name="originalFiles">
<list>
<option value="$PROJECT_DIR$/pom.xml" />
</list>
</option>
</component>
<component name="ProjectRootManager" version="2" languageLevel="JDK_1_6" default="false" project-jdk-name="1.8" project-jdk-type="JavaSDK">
<output url="file://$PROJECT_DIR$/out" />
</component>
</project>
\ No newline at end of file
... ...
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Palette2">
<group name="Swing">
<item class="com.intellij.uiDesigner.HSpacer" tooltip-text="Horizontal Spacer" icon="/com/intellij/uiDesigner/icons/hspacer.png" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="1" hsize-policy="6" anchor="0" fill="1" />
</item>
<item class="com.intellij.uiDesigner.VSpacer" tooltip-text="Vertical Spacer" icon="/com/intellij/uiDesigner/icons/vspacer.png" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="1" anchor="0" fill="2" />
</item>
<item class="javax.swing.JPanel" icon="/com/intellij/uiDesigner/icons/panel.png" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="3" hsize-policy="3" anchor="0" fill="3" />
</item>
<item class="javax.swing.JScrollPane" icon="/com/intellij/uiDesigner/icons/scrollPane.png" removable="false" auto-create-binding="false" can-attach-label="true">
<default-constraints vsize-policy="7" hsize-policy="7" anchor="0" fill="3" />
</item>
<item class="javax.swing.JButton" icon="/com/intellij/uiDesigner/icons/button.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="3" anchor="0" fill="1" />
<initial-values>
<property name="text" value="Button" />
</initial-values>
</item>
<item class="javax.swing.JRadioButton" icon="/com/intellij/uiDesigner/icons/radioButton.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="3" anchor="8" fill="0" />
<initial-values>
<property name="text" value="RadioButton" />
</initial-values>
</item>
<item class="javax.swing.JCheckBox" icon="/com/intellij/uiDesigner/icons/checkBox.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="3" anchor="8" fill="0" />
<initial-values>
<property name="text" value="CheckBox" />
</initial-values>
</item>
<item class="javax.swing.JLabel" icon="/com/intellij/uiDesigner/icons/label.png" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="0" anchor="8" fill="0" />
<initial-values>
<property name="text" value="Label" />
</initial-values>
</item>
<item class="javax.swing.JTextField" icon="/com/intellij/uiDesigner/icons/textField.png" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1">
<preferred-size width="150" height="-1" />
</default-constraints>
</item>
<item class="javax.swing.JPasswordField" icon="/com/intellij/uiDesigner/icons/passwordField.png" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1">
<preferred-size width="150" height="-1" />
</default-constraints>
</item>
<item class="javax.swing.JFormattedTextField" icon="/com/intellij/uiDesigner/icons/formattedTextField.png" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1">
<preferred-size width="150" height="-1" />
</default-constraints>
</item>
<item class="javax.swing.JTextArea" icon="/com/intellij/uiDesigner/icons/textArea.png" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JTextPane" icon="/com/intellij/uiDesigner/icons/textPane.png" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JEditorPane" icon="/com/intellij/uiDesigner/icons/editorPane.png" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JComboBox" icon="/com/intellij/uiDesigner/icons/comboBox.png" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="2" anchor="8" fill="1" />
</item>
<item class="javax.swing.JTable" icon="/com/intellij/uiDesigner/icons/table.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JList" icon="/com/intellij/uiDesigner/icons/list.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="2" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JTree" icon="/com/intellij/uiDesigner/icons/tree.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3">
<preferred-size width="150" height="50" />
</default-constraints>
</item>
<item class="javax.swing.JTabbedPane" icon="/com/intellij/uiDesigner/icons/tabbedPane.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="3" hsize-policy="3" anchor="0" fill="3">
<preferred-size width="200" height="200" />
</default-constraints>
</item>
<item class="javax.swing.JSplitPane" icon="/com/intellij/uiDesigner/icons/splitPane.png" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="3" hsize-policy="3" anchor="0" fill="3">
<preferred-size width="200" height="200" />
</default-constraints>
</item>
<item class="javax.swing.JSpinner" icon="/com/intellij/uiDesigner/icons/spinner.png" removable="false" auto-create-binding="true" can-attach-label="true">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1" />
</item>
<item class="javax.swing.JSlider" icon="/com/intellij/uiDesigner/icons/slider.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="8" fill="1" />
</item>
<item class="javax.swing.JSeparator" icon="/com/intellij/uiDesigner/icons/separator.png" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="6" anchor="0" fill="3" />
</item>
<item class="javax.swing.JProgressBar" icon="/com/intellij/uiDesigner/icons/progressbar.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="0" fill="1" />
</item>
<item class="javax.swing.JToolBar" icon="/com/intellij/uiDesigner/icons/toolbar.png" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="6" anchor="0" fill="1">
<preferred-size width="-1" height="20" />
</default-constraints>
</item>
<item class="javax.swing.JToolBar$Separator" icon="/com/intellij/uiDesigner/icons/toolbarSeparator.png" removable="false" auto-create-binding="false" can-attach-label="false">
<default-constraints vsize-policy="0" hsize-policy="0" anchor="0" fill="1" />
</item>
<item class="javax.swing.JScrollBar" icon="/com/intellij/uiDesigner/icons/scrollbar.png" removable="false" auto-create-binding="true" can-attach-label="false">
<default-constraints vsize-policy="6" hsize-policy="0" anchor="0" fill="2" />
</item>
</group>
</component>
</project>
\ No newline at end of file
... ...
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>
\ No newline at end of file
... ...
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>yoho-datasync-core</artifactId>
<groupId>com.yoho.datasync.core</groupId>
<version>1.0.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<packaging>jar</packaging>
<artifactId>yoho-datasync-core-base</artifactId>
<properties>
<java.version>1.8</java.version>
<project-version>1.0.0-SNAPSHOT</project-version>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
<build>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
\ No newline at end of file
... ...
package com.yoho.datasync.core;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface ClassTransfer {
public Class<?> targetClass();
}
... ...
package com.yoho.datasync.core;
public class DatasyncConstant {
public static final String ACTION_UPDATE = "0";
public static final String ACTION_DELETE = "1";
public static final String ACTION_INSERT = "2";
}
... ...
package com.yoho.datasync.core;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Target(ElementType.FIELD)
@Retention(RetentionPolicy.RUNTIME)
public @interface FieldTransfer {
public Class<?> targetClass() default ClassTransfer.class;
public String targetFieldName() default "";
}
... ...
package com.yoho.datasync.core;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface MqConsumerListerner {
String dbName();
String tableName();
}
... ...
package com.yoho.datasync.message;
import org.springframework.util.StringUtils;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
public class CharUtils {
/**
* 将字符转换成大写，如a->A
*
* @param c
* @return
*/
public static char toUpperCase(char c) {
if (c < 97 || c > 122)
return c;
else
return (char) (c - 32);
}
/**
* 将字符转换成小写，如A->a
*
* @param c
* @return
*/
public static char toLowerCase(char c) {
if (c < 65 || c > 90)
return c;
else
return (char) (c + 32);
}
public static boolean isLowerCase(char c) {
return (c >= 'a' && c <= 'z');
}
public static boolean isUpperCase(char c) {
return (c >= 'A' && c <= 'Z');
}
public static String getLowerCaseNames(String names) {
StringBuilder sb = new StringBuilder();
char[] cc = names.toCharArray();
sb.append(toLowerCase(cc[0]));
for (int i = 1; i < cc.length; i++) {
if (isLowerCase(cc[i - 1]) && isUpperCase(cc[i])) {
sb.append("_" + toLowerCase(cc[i]));
} else {
sb.append(toLowerCase(cc[i]));
}
}
;
return sb.toString();
}
/**
* 将驼峰风格替换为下划线风格
*
* @param str
* @return
*/
public static String CamelhumpToUnderline(String str) {
Matcher matcher = Pattern.compile("[A-Z]").matcher(str);
StringBuilder builder = new StringBuilder(str);
for (int i = 0; matcher.find(); i++) {
builder.replace(matcher.start() + i, matcher.end() + i, "_" + matcher.group().toLowerCase());
}
if (builder.charAt(0) == '_') {
builder.deleteCharAt(0);
}
return builder.toString();
}
/**
* 将下划线风格替换为驼峰风格
*
* @param str
* @return
*/
public static String underlineToCamelhump(String str) {
Matcher matcher = Pattern.compile("_[a-z]").matcher(str);
StringBuilder builder = new StringBuilder(str);
for (int i = 0; matcher.find(); i++) {
builder.replace(matcher.start() - i, matcher.end() - i, matcher.group().substring(1).toUpperCase());
}
if (Character.isUpperCase(builder.charAt(0))) {
builder.replace(0, 1, String.valueOf(Character.toLowerCase(builder.charAt(0))));
}
return builder.toString();
}
public static String underlineToCamelhumpWithNumber(String str) {
Matcher matcher = Pattern.compile("_[a-z0-9]").matcher(str);
StringBuilder builder = new StringBuilder(str);
for (int i = 0; matcher.find(); i++) {
builder.replace(matcher.start() - i, matcher.end() - i, matcher.group().substring(1).toUpperCase());
}
if (Character.isUpperCase(builder.charAt(0))) {
builder.replace(0, 1, String.valueOf(Character.toLowerCase(builder.charAt(0))));
}
return builder.toString();
}
public static String getBeanName(String str) {
StringBuilder builder = new StringBuilder(str);
builder.replace(0, 1, String.valueOf(Character.toUpperCase(builder.charAt(0))));
return builder.toString();
}
public static boolean isNumeric(String str) {
Pattern pattern = Pattern.compile("[0-9]*");
Matcher isNum = pattern.matcher(str);
if (!isNum.matches()) {
return false;
}
return true;
}
public static String standardized(String str){
if(StringUtils.isEmpty(str)){
return str;
}
return str.trim().toLowerCase().replaceAll(" ", "");
}
public static void main(String[] args) {
System.out.println(isLowerCase('C'));
String s1 = "size_attribute";
System.out.println(underlineToCamelhump(s1));
String s2 = "hserInfo";
System.out.println(getLowerCaseNames(s2));
String s3 = "brand";
System.out.println(getBeanName(s3));
String s4 = "skn_7_days_view";
System.out.println(underlineToCamelhumpWithNumber(s4));
System.out.println(standardized("Vans 卫衣"));
}
}
... ...
package com.yoho.datasync.message;
import com.alibaba.fastjson.JSONObject;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.AbstractMessageConverter;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.stereotype.Component;
import java.io.UnsupportedEncodingException;
@Component
public class FastJsonMessageConverter extends AbstractMessageConverter {
public static final String DEFAULT_CHARSET = "UTF-8";
private volatile String defaultCharset = DEFAULT_CHARSET;
public FastJsonMessageConverter() {
super();
}
@Override
public Object fromMessage(Message message) throws MessageConversionException {
return message;
}
@Override
protected Message createMessage(Object objectToConvert, MessageProperties messageProperties) throws MessageConversionException {
byte[] bytes = null;
try {
String jsonString = JSONObject.toJSONString(objectToConvert);
bytes = jsonString.getBytes(this.defaultCharset);
} catch (UnsupportedEncodingException e) {
throw new MessageConversionException("Failed to convert Message content", e);
}
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
messageProperties.setContentEncoding(this.defaultCharset);
if (bytes != null) {
messageProperties.setContentLength(bytes.length);
}
return new Message(bytes, messageProperties);
}
}
\ No newline at end of file
... ...
package com.yoho.datasync.message;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@Component
public class MessageHelper {
private static final Logger logger = LoggerFactory.getLogger(MessageHelper.class);
private static final String QUEUE_PERIFX = "yohodatasync.";
@Resource
private TableConfigLoader tableConfigLoader;
private String getQueuePerfix(String dbName, String tableName) {
StringBuilder queuePerfix = new StringBuilder(QUEUE_PERIFX);
if (StringUtils.isNotBlank(dbName)) {
queuePerfix.append(dbName);
queuePerfix.append(".");
}
if (StringUtils.isNotBlank(tableName)) {
queuePerfix.append(tableName);
}
return queuePerfix.toString();
}
/**
* 获取更新时的真实的队列名
* yohodatasync.shops.user_account
*/
public String getRealQueueName(String dbName, String tableName) {
StringBuilder queuePerfix = new StringBuilder(QUEUE_PERIFX);
if (StringUtils.isNotBlank(dbName)) {
queuePerfix.append(dbName);
queuePerfix.append(".");
}
if (StringUtils.isNotBlank(tableName)) {
queuePerfix.append(tableName);
}
return queuePerfix.toString();
}
public String getPrimaryKeysValue(TableConfig tableConfig, Map<String, Object> dataMap) {
String[] primaryKeys = tableConfig.getPrimaryKeys();
StringBuilder primaryKeysValue = new StringBuilder();
for (String primaryKey : primaryKeys) {
primaryKeysValue.append(dataMap.get(primaryKey));
}
return primaryKeysValue.toString();
}
/**
* 当前表对应的队列是否是单队列
*
* @param tableConfig
* @return
*/
private boolean isSingleQueue(TableConfig tableConfig) {
if (tableConfig == null) {
return true;
}
int queueSize = tableConfig.getQueueSize();
if (queueSize <= 1) {
return true;
}
String[] primaryKeys = tableConfig.getPrimaryKeys();
return primaryKeys == null;
}
private String genRealQueueName(String queuePerfix, int index) {
return queuePerfix + "." + index;
}
/**
* 根据表名获取队列名
*
* @param tableName
* @return
*/
public List<String> getTableQueueNames(String dbName, String tableName) {
String queuePerfix = this.getQueuePerfix(dbName, tableName);
TableConfig tableConfig = tableConfigLoader.getTableConfig(dbName, tableName);
if (this.isSingleQueue(tableConfig)) {
return Arrays.asList(this.genRealQueueName(queuePerfix, 0));
}
List<String> results = new ArrayList<>();
for (int i = 0; i < tableConfig.getQueueSize(); i++) {
results.add(this.genRealQueueName(queuePerfix, i));
}
return results;
}
/**
* 获取更新时的真实的队列名
*
* @param mqMessage
* @param tableName
* @param mqMessage
* @return
*/
public String getRealQueueName(String dbName, String tableName, MqMessageEntity mqMessage) {
String queuePerfix = this.getQueuePerfix(dbName, tableName);
try {
TableConfig tableConfig = tableConfigLoader.getTableConfig(dbName, tableName);
Map<String, Object> dataMap = mqMessage.getData();
if (this.isSingleQueue(tableConfig) || dataMap==null) {
return this.genRealQueueName(queuePerfix, 0);
}
String primaryKeysValue = getPrimaryKeysValue(tableConfig, dataMap);
if (StringUtils.isBlank(primaryKeysValue)) {
return this.genRealQueueName(queuePerfix, 0);
}
int queueIndex = Math.abs(primaryKeysValue.hashCode()) % tableConfig.getQueueSize();
return this.genRealQueueName(queuePerfix, queueIndex);
} catch (Exception e) {
logger.error(e.getMessage(), e);
return this.genRealQueueName(queuePerfix, 0);
}
}
}
... ...
package com.yoho.datasync.message;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@Component
public class MqBeansResgister implements InitializingBean {
private static final int requestedHeartBeat = 10;
private static final int channelCacheSize = 60;
@Value("${rabbitmq.username}")
private String userName;
@Value("${rabbitmq.password}")
private String passWord;
@Value("${rabbitmq.server}")
private String server;
@Value("${rabbitmq.port}")
private String port;
@Value("${rabbitmq.server}:${rabbitmq.port}")
private String addresses;
@Value("${rabbitmq.virtualHost}")
private String virtualHost;
@Value("${rabbitmq.exchange}")
private String exchange;
@Resource
private FastJsonMessageConverter fastJsonMessageConverter;
@Resource
private TableConfigLoader tableConfigLoader;
@Resource
private MessageHelper messageHelper;
private List<String> queueNames = new ArrayList<>();
private ConnectionFactory connectionFactory;
private RabbitAdmin rabbitAdmin;
private DirectExchange directExchange;
private RabbitTemplate rabbitTemplate;
//优先级在PostConstruct之后，tableConfigLoader加载完
@Override
public void afterPropertiesSet() {
//定义connectionFactory
this.connectionFactory = createConnectionFactory();
//定义rabbitAdmin
this.rabbitAdmin = new RabbitAdmin(connectionFactory);
//定义Exchange
this.directExchange = new DirectExchange(exchange);
//定义队列
Collection<TableConfig> configs = tableConfigLoader.getTableConfigs();
for (TableConfig tableConfig: configs){
queueNames.addAll(messageHelper.getTableQueueNames(tableConfig.getDbName(),tableConfig.getTableName()));
}
for (String queueName:queueNames ){
this.declareAndGetQueueName(queueName);
}
//定义rabbitTemplate
this.rabbitTemplate = createRabbitTemplate();
}
private ConnectionFactory createConnectionFactory() {
CachingConnectionFactory cf = new CachingConnectionFactory(server);
cf.setUsername(userName);
cf.setPassword(passWord);
cf.setAddresses(addresses);
cf.setVirtualHost(virtualHost);
cf.setChannelCacheSize(channelCacheSize);
cf.setRequestedHeartBeat(requestedHeartBeat);
return cf ;
}
private String declareAndGetQueueName(String queueName){
Queue queue = new Queue(queueName);
// declareQueue
this.rabbitAdmin.declareQueue(queue);
//binding
Binding binding = BindingBuilder.bind(queue).to(this.directExchange).with(queueName);
this.rabbitAdmin.declareBinding(binding);
return queueName;
}
private RabbitTemplate createRabbitTemplate() {
RabbitTemplate amqpTemplate = new RabbitTemplate(this.connectionFactory );
amqpTemplate.setMessageConverter(fastJsonMessageConverter);
amqpTemplate.setExchange(exchange);
return amqpTemplate;
}
public RabbitTemplate getRabbitTemplate() {
return rabbitTemplate;
}
public ConnectionFactory getConnectionFactory() {
return connectionFactory;
}
}
... ...
package com.yoho.datasync.message;
import com.yoho.datasync.core.MqConsumerListerner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
@Component
public class MqConsumerFactory implements ApplicationContextAware {
private static final Logger logger = LoggerFactory.getLogger(MqConsumerFactory.class);
@Autowired
private MessageHelper messageHelper;
@Autowired
private FastJsonMessageConverter fastJsonMessageConverter;
@Autowired
private MqBeansResgister searchMqBeansResgister;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Map<String, Object> consumerBeans = applicationContext.getBeansWithAnnotation(MqConsumerListerner.class);
if(consumerBeans==null || consumerBeans.isEmpty()){
logger.info("can not load any rabbit consumer...");
return;
}
this.addMessageListener(consumerBeans);
}
private void addMessageListener(Map<String, Object> consumerBeans) {
for (Object consumer : consumerBeans.values()) {
MqConsumerListerner searchRmqListerner = consumer.getClass().getAnnotation(MqConsumerListerner.class);
String dbName = searchRmqListerner.dbName();
String tableName = searchRmqListerner.tableName();
List<String> queueNames = messageHelper.getTableQueueNames(dbName,tableName);
for (String queueName : queueNames) {
try {
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
listenerContainer.setConnectionFactory(searchMqBeansResgister.getConnectionFactory());
listenerContainer.setConcurrentConsumers(1);
listenerContainer.setPrefetchCount(1);
//linstener
MessageListenerAdapter adapter = new MessageListenerAdapter(consumer);
adapter.setMessageConverter(fastJsonMessageConverter);
listenerContainer.setMessageListener(adapter);
listenerContainer.setQueueNames(queueName);
listenerContainer.start();
logger.info("starting rabbitmq container {} success, concurrentConsumers {}.", queueName,1);
} catch (Exception e) {
logger.error("starting rabbitmq container failed.", e);
}
}
}
}
}
... ...
package com.yoho.datasync.message;
import lombok.Data;
import java.util.Map;
@Data
public class MqMessageEntity {
/**
* binlog操作类型，新增，删除等
*/
private String action;
private String dbName;
private String tableName;
/**
* 版本号
*/
private long version;
private Map<String, Object> data;
public MqMessageEntity() {
}
public MqMessageEntity(String action, String dbName, String tableName, long version, Map<String, Object> data) {
this.action = action;
this.dbName = dbName;
this.tableName = tableName;
this.version = version;
this.data = data;
}
}
... ...
package com.yoho.datasync.message;
import lombok.Data;
@Data
public class TableConfig {
private String dbName;
private String tableName;
private String[] primaryKeys;
private int queueSize;
}
... ...
package com.yoho.datasync.message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import javax.annotation.PostConstruct;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
@Component
public class TableConfigLoader {
private static final Logger logger = LoggerFactory.getLogger(TableConfigLoader.class);
private Map<String,String> loadTableProperties(){
try {
Resource resource = new ClassPathResource("table.properties");
Properties props = PropertiesLoaderUtils.loadProperties(resource);
Map<String,String> param = new HashMap<String,String>((Map) props);
return param;
}catch (Exception e){
logger.error(e.getMessage(),e);
return new HashMap<>();
}
}
private Map<String, TableConfig> tableConfigMap = new HashMap<>();
@PostConstruct
private void init() {
Map<String,String> props = loadTableProperties();
for (Map.Entry<String, String> entry : props.entrySet()) {
TableConfig tableConfig = new TableConfig();
//配置库名和表名
String key = entry.getKey();
if (key.contains(".")) {
tableConfig.setDbName(key.split("\\.")[0]);
tableConfig.setTableName(key.split("\\.")[1]);
} else {
tableConfig.setDbName("");
tableConfig.setTableName(key);
}
String configValues = entry.getValue();
String[] configValueArray = configValues.split("\\|");
String[] originPrimaryKeys = configValueArray[0].split(",");
String[] primaryKeys = new String[originPrimaryKeys.length];
for (int i = 0; i < originPrimaryKeys.length; i++) {
if (originPrimaryKeys[i].indexOf("_") >= 0) {
primaryKeys[i] = CharUtils.underlineToCamelhump(originPrimaryKeys[i]);
} else {
primaryKeys[i] = originPrimaryKeys[i];
}
}
tableConfig.setPrimaryKeys(primaryKeys);
tableConfig.setQueueSize(Integer.valueOf(configValueArray[1]));
tableConfigMap.put(this.getTableConfigKey(tableConfig.getDbName(), tableConfig.getTableName()), tableConfig);
}
}
public Collection<TableConfig> getTableConfigs() {
return this.tableConfigMap.values();
}
public TableConfig getTableConfig(String dbName, String tableName) {
String tabelConfigKey = getTableConfigKey(dbName, tableName);
return this.tableConfigMap.get(tabelConfigKey);
}
private String getTableConfigKey(String dbName, String tableName) {
if (StringUtils.isEmpty(dbName)) {
return tableName;
}
return new StringBuilder(dbName).append(".").append(tableName).toString();
}
}
... ...
package com.yoho.datasync.model.yh_grass;
import com.yoho.datasync.core.ClassTransfer;
import com.yoho.datasync.core.FieldTransfer;
import com.yoho.datasync.model.yh_pcms.PublicUserPraise;
@ClassTransfer(targetClass = PublicUserPraise.class)
public class GrassArticlePraise {
private Integer id;
@FieldTransfer(targetFieldName = "targetId")
private Integer articleId;
@FieldTransfer
private Integer uid;
@FieldTransfer
private Integer createTime;
@FieldTransfer
private Integer updateTime;
@FieldTransfer
private Integer status;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public Integer getArticleId() {
return articleId;
}
public void setArticleId(Integer articleId) {
this.articleId = articleId;
}
public Integer getUid() {
return uid;
}
public void setUid(Integer uid) {
this.uid = uid;
}
public Integer getCreateTime() {
return createTime;
}
public void setCreateTime(Integer createTime) {
this.createTime = createTime;
}
public Integer getStatus() {
return status;
}
public void setStatus(Integer status) {
this.status = status;
}
public Integer getUpdateTime() {
return updateTime;
}
public void setUpdateTime(Integer updateTime) {
this.updateTime = updateTime;
}
}
... ...
package com.yoho.datasync.model.yh_pcms;
import lombok.Data;
import javax.persistence.*;
@Data
@MappedSuperclass
public class BaseEntity {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Column(name = "id", nullable = false, unique = true)
protected Integer id;
}
... ...
package com.yoho.datasync.model.yh_pcms;
import lombok.Data;
import lombok.EqualsAndHashCode;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Table;
import java.io.Serializable;
@Data
@Entity
@Table(name = "public_user_praise")
@EqualsAndHashCode(callSuper = true)
public class PublicUserPraise extends BaseEntity implements Serializable {
@Column(name = "target_id")
private Integer targetId;
private Integer uid;
@Column(name = "praise_type")
private Integer praiseType;
private Integer status;
@Column(name = "src_channel")
private Integer srcChannel;
@Column(name = "create_time")
private Long createTime;
@Column(name = "update_time")
private Long updateTime;
}
... ...
####from yh_grass####
yh_grass.grass_article_praise=id|1
... ...
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.yoho.datasync.core</groupId>
<artifactId>yoho-datasync-core</artifactId>
<packaging>pom</packaging>
<version>1.0.0-SNAPSHOT</version>
<modules>
<module>base</module>
</modules>
<dependencies>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.6</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>javax.persistence</groupId>
<artifactId>javax.persistence-api</artifactId>
<version>2.2</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.8.1</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>2.1.4.RELEASE</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.13.sec01</version>
</dependency>
</dependencies>
</project>
\ No newline at end of file
... ...
<?xml version="1.0" encoding="UTF-8"?>
<module type="JAVA_MODULE" version="4" />
\ No newline at end of file
... ...