datasync / yoho-datasync-producer · Commits

Go to a project

Authored by liangyi.chen@yoho.cn
Commit 6deb47cd05edc7be502e344a6b9b1c4ba771949e 1 parent 6d31e963

update

Inline Side-by-side
Showing 12 changed files with 15 additions and 558 deletions
... ... @@ -2,11 +2,11 @@ package com.yoho.datasync.producer.canal.common;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.yoho.datasync.producer.common.constant.DatasyncConstant;
import com.yoho.datasync.producer.common.entity.MqMessageEntity;
import com.yoho.datasync.producer.common.helper.MessageHelper;
import com.yoho.datasync.producer.common.mqcomponent.MqBeansResgister;
import com.yoho.datasync.producer.common.utils.CharUtils;
import com.yoho.datasync.core.DatasyncConstant;
import com.yoho.datasync.message.CharUtils;
import com.yoho.datasync.message.MessageHelper;
import com.yoho.datasync.message.MqBeansResgister;
import com.yoho.datasync.message.MqMessageEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
... ...
package com.yoho.datasync.producer.common.config;
import com.yoho.datasync.producer.common.entity.TableConfig;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Component
@ConfigurationProperties
public class TableConfigLoader {
@Setter@Getter
private List<TableConfig> tableConfigs;
private Map<String, TableConfig> tableConfigMap = new HashMap<>();
public Map<String, TableConfig> getTableConfigMap() {
return tableConfigMap;
}
@PostConstruct
private void init() {
if(!CollectionUtils.isEmpty(tableConfigs)) {
tableConfigs.forEach(tableConfig ->
tableConfigMap.put(this.getTableConfigKey(tableConfig.getDbName(), tableConfig.getTableName()),
tableConfig));
}
}
public TableConfig getTableConfigByKey(String dbName, String tableName) {
String tabelConfigKey = getTableConfigKey(dbName, tableName);
return this.tableConfigMap.get(tabelConfigKey);
}
private String getTableConfigKey(String dbName, String tableName) {
if (StringUtils.isBlank(dbName)) {
return tableName;
}
return dbName + "." + tableName;
}
}
package com.yoho.datasync.producer.common.constant;
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.producer.common.entity;
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(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.producer.common.entity;
import lombok.Data;
@Data
public class TableConfig {
private String dbName;
private String tableName;
private String[] primaryKeys;
private int queueSize;
}
package com.yoho.datasync.producer.common.helper;
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.producer.common.helper;
import com.yoho.datasync.producer.common.config.TableConfigLoader;
import com.yoho.datasync.producer.common.entity.MqMessageEntity;
import com.yoho.datasync.producer.common.entity.TableConfig;
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();
}
private 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.getTableConfigByKey(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.getTableConfigByKey(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);
}
}
}
... ... @@ -13,6 +13,8 @@ public class GrassArticlePraise {
private Integer createTime;
private Integer updateTime;
private Integer status;
}
... ...
package com.yoho.datasync.producer.common.mqcomponent;
import com.yoho.datasync.producer.common.config.TableConfigLoader;
import com.yoho.datasync.producer.common.entity.TableConfig;
import com.yoho.datasync.producer.common.helper.FastJsonMessageConverter;
import com.yoho.datasync.producer.common.helper.MessageHelper;
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.getTableConfigMap().values();
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.producer.common.utils;
import org.apache.commons.lang3.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 卫衣"));
}
}
... ... @@ -24,6 +24,7 @@
<project-version>1.0.0-SNAPSHOT</project-version>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<datasync.core.version>1.0.0-SNAPSHOT</datasync.core.version>
</properties>
<dependencies>
... ... @@ -61,6 +62,12 @@
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
<groupId>com.yoho.datasync.core</groupId>
<artifactId>yoho-datasync-core-base</artifactId>
<version>${datasync.core.version}</version>
</dependency>
</dependencies>
<dependencyManagement>
... ...
... ... @@ -5,7 +5,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
@SpringBootApplication
@ComponentScan(value = "com.yoho.datasync.producer")
@ComponentScan(value = "com.yoho.datasync")
public class YohoDatasyncProducerApplication {
public static void main(String[] args) {
... ...