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

kafka生产端增加按pk hash到对应partition功能

Browse Source
This commit is contained in:
mcy
2018-09-20 16:48:14 +08:00
parent 0cda90cdea
commit 74890050d0
5 changed files with 79 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
deployer/src/main/resources/kafka.yml
View File

@@ -12,7 +12,9 @@ flatMessage: true
canalDestinations:
- canalDestination: example
topic: example
partition:
topic: expample
#partitionsNum: 3
#partitionHash:
# mytest.person: id

31
protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
View File

@@ -1,14 +1,8 @@
package com.alibaba.otter.canal.protocol;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.*;
import com.google.common.collect.Table;
import com.google.protobuf.ByteString;
/**
@@ -31,7 +25,7 @@ public class FlatMessage implements Serializable {
private List<Map<String, String>> data;
private List<Map<String, String>> old;
public FlatMessage() {
public FlatMessage(){
}
public FlatMessage(long id){
@@ -126,6 +120,12 @@ public class FlatMessage implements Serializable {
this.old = old;
}
/**
* 将Message转换为FlatMessage
*
* @param message 原生message
* @return FlatMessage列表
*/
public static List<FlatMessage> messageConverter(Message message) {
try {
if (message == null) {
@@ -231,11 +231,22 @@ public class FlatMessage implements Serializable {
}
}
/**
* 将FlatMessage按指定的字段值hash拆分
*
* @param flatMessage flatMessage
* @param partitionsNum 分区数量
* @param pkHashConfig hash映射
* @return 拆分后的flatMessage数组
*/
public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum,
Table<String, String, String> pkHashConfig) {
Map<String, String> pkHashConfig) {
if (partitionsNum == null) {
partitionsNum = 1;
}
FlatMessage[] partitionMessages = new FlatMessage[partitionsNum];
String pk = pkHashConfig.get(flatMessage.getDatabase(), flatMessage.getTable());
String pk = pkHashConfig.get(flatMessage.getDatabase() + "." + flatMessage.getTable());
if (pk == null || flatMessage.getIsDdl()) {
partitionMessages[0] = flatMessage;
} else {

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

@@ -67,15 +67,18 @@ public class CanalKafkaProducer {
}
}
public void send(KafkaProperties.Topic topic, Message message, Callback callback) {
public void send(KafkaProperties.CanalDestination canalDestination, Message message, Callback callback) {
try {
// producer.beginTransaction();
if (!kafkaProperties.getFlatMessage()) {
ProducerRecord<String, Message> record;
if (topic.getPartition() != null) {
record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);
if (canalDestination.getPartition() != null) {
record = new ProducerRecord<String, Message>(canalDestination.getTopic(),
canalDestination.getPartition(),
null,
message);
} else {
record = new ProducerRecord<String, Message>(topic.getTopic(), 0, null, message);
record = new ProducerRecord<String, Message>(canalDestination.getTopic(), 0, null, message);
}
producer.send(record);
@@ -84,19 +87,39 @@ public class CanalKafkaProducer {
List<FlatMessage> flatMessages = FlatMessage.messageConverter(message);
if (flatMessages != null) {
for (FlatMessage flatMessage : flatMessages) {
ProducerRecord<String, String> record;
if (topic.getPartition() != null) {
record = new ProducerRecord<String, String>(topic.getTopic(),
topic.getPartition(),
null,
JSON.toJSONString(flatMessage));
if (canalDestination.getPartition() != null) {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination
.getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage));
producer2.send(record);
} else {
record = new ProducerRecord<String, String>(topic.getTopic(),
0,
null,
JSON.toJSONString(flatMessage));
if (canalDestination.getPartitionHash() != null
&& !canalDestination.getPartitionHash().isEmpty()) {
FlatMessage[] partitionFlatMessage = FlatMessage.messagePartition(flatMessage,
canalDestination.getPartitionsNum(),
canalDestination.getPartitionHash());
int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) {
FlatMessage flatMessage1 = partitionFlatMessage[i];
if (flatMessage1 == null) {
continue;
}
ProducerRecord<String, String> record = new ProducerRecord<String, String>(
canalDestination.getTopic(),
i,
null,
JSON.toJSONString(partitionFlatMessage[i]));
producer2.send(record);
}
} else {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(
canalDestination.getTopic(),
0,
null,
JSON.toJSONString(flatMessage));
producer2.send(record);
}
}
producer2.send(record);
}
}
}
@@ -104,7 +127,7 @@ public class CanalKafkaProducer {
// producer.commitTransaction();
callback.commit();
if (logger.isDebugEnabled()) {
logger.debug("send message to kafka topic: {}", topic.getTopic());
logger.debug("send message to kafka topic: {}", canalDestination.getTopic());
}
} catch (Exception e) {
logger.error(e.getMessage(), e);

5
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java
View File

@@ -132,10 +132,7 @@ public class CanalKafkaStarter implements CanalServerStarter {
try {
int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size();
if (batchId != -1 && size != 0) {
Topic topic = new Topic();
topic.setTopic(destination.getTopic());
topic.setPartition(destination.getPartition());
canalKafkaProducer.send(topic, message, new CanalKafkaProducer.Callback() {
canalKafkaProducer.send(destination, message, new CanalKafkaProducer.Callback() {
@Override
public void commit() {

60
server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java
View File

@@ -1,9 +1,8 @@
package com.alibaba.otter.canal.kafka;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.Map;
/**
* kafka 配置项
@@ -27,10 +26,11 @@ public class KafkaProperties {
public static class CanalDestination {
private String canalDestination;
private String topic;
private Integer partition;
private Set<Topic> topics = new HashSet<Topic>();
private String canalDestination;
private String topic;
private Integer partition;
private Integer partitionsNum;
private Map<String, String> partitionHash;
public String getCanalDestination() {
return canalDestination;
@@ -56,52 +56,20 @@ public class KafkaProperties {
this.partition = partition;
}
public Set<Topic> getTopics() {
return topics;
public Integer getPartitionsNum() {
return partitionsNum;
}
public void setTopics(Set<Topic> topics) {
this.topics = topics;
}
}
public static class Topic {
private String topic;
private Integer partition;
public String getTopic() {
return topic;
public void setPartitionsNum(Integer partitionsNum) {
this.partitionsNum = partitionsNum;
}
public void setTopic(String topic) {
this.topic = topic;
public Map<String, String> getPartitionHash() {
return partitionHash;
}
public Integer getPartition() {
return partition;
}
public void setPartition(Integer partition) {
this.partition = partition;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Topic topic1 = (Topic) o;
if (topic != null ? !topic.equals(topic1.topic) : topic1.topic != null) return false;
return partition != null ? partition.equals(topic1.partition) : topic1.partition == null;
}
@Override
public int hashCode() {
int result = topic != null ? topic.hashCode() : 0;
result = 31 * result + (partition != null ? partition.hashCode() : 0);
return result;
public void setPartitionHash(Map<String, String> partitionHash) {
this.partitionHash = partitionHash;
}
}