chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

format

Browse Source
This commit is contained in:
agapple
2019-10-15 21:35:57 +08:00
parent 65b5325ca6
commit dd7554374f
5 changed files with 19 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java
View File

@@ -234,11 +234,12 @@ public class MQMessageUtils {
* *
* @param partitionsNum 分区数 * @param partitionsNum 分区数
* @param pkHashConfigs 分区库表主键正则表达式 * @param pkHashConfigs 分区库表主键正则表达式
* @param databaseHash 是否取消根据database进行hash * @param databaseHash 是否取消根据database进行hash
* @return 分区message数组 * @return 分区message数组
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public static Message[] messagePartition(EntryRowData[] datas, long id, Integer partitionsNum, String pkHashConfigs,boolean databaseHash) { public static Message[] messagePartition(EntryRowData[] datas, long id, Integer partitionsNum,
String pkHashConfigs, boolean databaseHash) {
if (partitionsNum == null) { if (partitionsNum == null) {
partitionsNum = 1; partitionsNum = 1;
} }
@@ -281,7 +282,7 @@ public class MQMessageUtils {
for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
int hashCode = 0; int hashCode = 0;
if(databaseHash){ if (databaseHash) {
hashCode = database.hashCode(); hashCode = database.hashCode();
} }
CanalEntry.EventType eventType = rowChange.getEventType(); CanalEntry.EventType eventType = rowChange.getEventType();
@@ -448,10 +449,11 @@ public class MQMessageUtils {
* @param flatMessage flatMessage * @param flatMessage flatMessage
* @param partitionsNum 分区数量 * @param partitionsNum 分区数量
* @param pkHashConfigs hash映射 * @param pkHashConfigs hash映射
* @param databaseHash 是否取消根据database进行hash * @param databaseHash 是否取消根据database进行hash
* @return 拆分后的flatMessage数组 * @return 拆分后的flatMessage数组
*/ */
public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs,boolean databaseHash) { public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs,
boolean databaseHash) {
if (partitionsNum == null) { if (partitionsNum == null) {
partitionsNum = 1; partitionsNum = 1;
} }
@@ -482,7 +484,7 @@ public class MQMessageUtils {
int idx = 0; int idx = 0;
for (Map<String, String> row : flatMessage.getData()) { for (Map<String, String> row : flatMessage.getData()) {
int hashCode = 0; int hashCode = 0;
if(databaseHash){ if (databaseHash) {
hashCode = database.hashCode(); hashCode = database.hashCode();
} }
if (pkNames != null) { if (pkNames != null) {

8
server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
View File

@@ -48,10 +48,9 @@ public class MQProperties {
private String exchange = ""; private String exchange = "";
// 消息发送的并行度 // 消息发送的并行度
private int parallelThreadSize = 8; private int parallelThreadSize = 8;
//是否取消根据database进行hash // 是否取消根据database进行hash
private boolean databaseHash = true; private boolean databaseHash = true;
public static class CanalDestination { public static class CanalDestination {
private String canalDestination; private String canalDestination;
@@ -349,6 +348,7 @@ public class MQProperties {
public void setDatabaseHash(boolean databaseHash) { public void setDatabaseHash(boolean databaseHash) {
this.databaseHash = databaseHash; this.databaseHash = databaseHash;
} }
@Override @Override
public String toString() { public String toString() {
return "MQProperties [servers=" + servers + ", retries=" + retries + ", batchSize=" + batchSize + ", lingerMs=" return "MQProperties [servers=" + servers + ", retries=" + retries + ", batchSize=" + batchSize + ", lingerMs="
@@ -362,9 +362,7 @@ public class MQProperties {
+ kerberosEnable + ", kerberosKrb5FilePath=" + kerberosKrb5FilePath + ", kerberosJaasFilePath=" + kerberosEnable + ", kerberosKrb5FilePath=" + kerberosKrb5FilePath + ", kerberosJaasFilePath="
+ kerberosJaasFilePath + ", username=" + username + ", password=" + password + ", vhost=" + vhost + kerberosJaasFilePath + ", username=" + username + ", password=" + password + ", vhost=" + vhost
+ ", aliyunUID=" + aliyunUID + ", exchange=" + exchange + ", parallelThreadSize=" + parallelThreadSize + ", aliyunUID=" + aliyunUID + ", exchange=" + exchange + ", parallelThreadSize=" + parallelThreadSize
+ ",databaseHash=" + databaseHash +"]"; + ",databaseHash=" + databaseHash + "]";
} }
} }

2
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -42,7 +42,7 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
private Producer<String, byte[]> producer; private Producer<String, byte[]> producer;
private MQProperties kafkaProperties; private MQProperties kafkaProperties;
private boolean databaseHash; private boolean databaseHash;
@Override @Override
public void init(MQProperties kafkaProperties) { public void init(MQProperties kafkaProperties) {

9
server/src/main/java/com/alibaba/otter/canal/rabbitmq/AliyunCredentialsProvider.java
View File

@@ -10,10 +10,8 @@ import java.security.NoSuchAlgorithmException;
import com.alibaba.mq.amqp.utils.UserUtils; import com.alibaba.mq.amqp.utils.UserUtils;
import com.rabbitmq.client.impl.CredentialsProvider; import com.rabbitmq.client.impl.CredentialsProvider;
public class AliyunCredentialsProvider implements CredentialsProvider { public class AliyunCredentialsProvider implements CredentialsProvider {
/** /**
* Access Key ID * Access Key ID
*/ */
@@ -27,16 +25,14 @@ public class AliyunCredentialsProvider implements CredentialsProvider {
/** /**
* 资源主账号ID * 资源主账号ID
*/ */
private final long resourceOwnerId; private final long resourceOwnerId;
public AliyunCredentialsProvider(final String accessKey, final String accessSecret, final long resourceOwnerId){
public AliyunCredentialsProvider(final String accessKey, final String accessSecret, final long resourceOwnerId) {
this.AliyunAccessKey = accessKey; this.AliyunAccessKey = accessKey;
this.AliyunAccessSecret = accessSecret; this.AliyunAccessSecret = accessSecret;
this.resourceOwnerId = resourceOwnerId; this.resourceOwnerId = resourceOwnerId;
} }
@Override @Override
public String getUsername() { public String getUsername() {
return UserUtils.getUserName(AliyunAccessKey, resourceOwnerId); return UserUtils.getUserName(AliyunAccessKey, resourceOwnerId);
@@ -51,5 +47,4 @@ public class AliyunCredentialsProvider implements CredentialsProvider {
return null; return null;
} }
} }

8
server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
View File

@@ -40,7 +40,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
private MQProperties mqProperties; private MQProperties mqProperties;
private static final String CLOUD_ACCESS_CHANNEL = "cloud"; private static final String CLOUD_ACCESS_CHANNEL = "cloud";
private boolean databaseHash; private boolean databaseHash;
@Override @Override
public void init(MQProperties rocketMQProperties) { public void init(MQProperties rocketMQProperties) {
@@ -128,7 +128,8 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas, com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas,
message.getId(), message.getId(),
destination.getPartitionsNum(), destination.getPartitionsNum(),
destination.getPartitionHash(),databaseHash); destination.getPartitionHash(),
databaseHash);
int length = messages.length; int length = messages.length;
ExecutorTemplate template = new ExecutorTemplate(executor); ExecutorTemplate template = new ExecutorTemplate(executor);
@@ -171,7 +172,8 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
for (FlatMessage flatMessage : flatMessages) { for (FlatMessage flatMessage : flatMessages) {
FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
destination.getPartitionsNum(), destination.getPartitionsNum(),
destination.getPartitionHash(),databaseHash); destination.getPartitionHash(),
databaseHash);
int length = partitionFlatMessage.length; int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
partitionFlatMessages.get(i).add(partitionFlatMessage[i]); partitionFlatMessages.get(i).add(partitionFlatMessage[i]);