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 异步发送 flat message

Browse Source
This commit is contained in:
mcy
2018-12-27 17:52:33 +08:00
parent d9e259dde4
commit 150a13824b
1 changed files with 27 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -124,6 +124,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
} }
if (record != null) { if (record != null) {
// 同步发送原生message
producer.send(record).get(); producer.send(record).get();
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
@@ -141,13 +142,17 @@ public class CanalKafkaProducer implements CanalMQProducer {
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(message); List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(message);
if (flatMessages != null) { if (flatMessages != null) {
for (FlatMessage flatMessage : flatMessages) { for (FlatMessage flatMessage : flatMessages) {
if (canalDestination.getPartition() != null) { if (StringUtils.isEmpty(canalDestination.getPartitionHash())) {
try { try {
Integer partition = canalDestination.getPartition();
if (partition == null) {
partition = 0;
}
ProducerRecord<String, String> record = new ProducerRecord<String, String>(topicName, ProducerRecord<String, String> record = new ProducerRecord<String, String>(topicName,
canalDestination.getPartition(), partition,
null, null,
JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue)); JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue));
producer2.send(record).get(); producer2.send(record);
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
// producer.abortTransaction(); // producer.abortTransaction();
@@ -155,43 +160,27 @@ public class CanalKafkaProducer implements CanalMQProducer {
return; return;
} }
} else { } else {
if (canalDestination.getPartitionHash() != null FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
&& !canalDestination.getPartitionHash().isEmpty()) { canalDestination.getPartitionsNum(),
FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, canalDestination.getPartitionHash());
canalDestination.getPartitionsNum(), int length = partitionFlatMessage.length;
canalDestination.getPartitionHash()); for (int i = 0; i < length; i++) {
int length = partitionFlatMessage.length; FlatMessage flatMessagePart = partitionFlatMessage[i];
for (int i = 0; i < length; i++) { if (flatMessagePart != null) {
FlatMessage flatMessagePart = partitionFlatMessage[i]; try {
if (flatMessagePart != null) { ProducerRecord<String, String> record = new ProducerRecord<String, String>(
try { topicName,
ProducerRecord<String, String> record = new ProducerRecord<String, String>( i,
topicName, null,
i, JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue));
null, producer2.send(record);
JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue)); } catch (Exception e) {
producer2.send(record).get(); logger.error(e.getMessage(), e);
} catch (Exception e) { // producer.abortTransaction();
logger.error(e.getMessage(), e); callback.rollback();
// producer.abortTransaction(); return;
callback.rollback();
return;
}
} }
} }
} else {
try {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(topicName,
0,
null,
JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue));
producer2.send(record).get();
} catch (Exception e) {
logger.error(e.getMessage(), e);
// producer.abortTransaction();
callback.rollback();
return;
}
} }
} }
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {