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

fixed issue #4923 , support fastJSON Feature.LargeObject

Browse Source
This commit is contained in:
jianghang.loujh
2023-11-03 12:39:53 +08:00
parent 55a5e7592b
commit 163f719971
4 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java
View File

@@ -9,7 +9,6 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import com.alibaba.otter.canal.common.utils.PropertiesUtils;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.Producer;
@@ -21,6 +20,7 @@ import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter; import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate; import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
import com.alibaba.otter.canal.common.utils.PropertiesUtils;
import com.alibaba.otter.canal.connector.core.producer.AbstractMQProducer; import com.alibaba.otter.canal.connector.core.producer.AbstractMQProducer;
import com.alibaba.otter.canal.connector.core.producer.MQDestination; import com.alibaba.otter.canal.connector.core.producer.MQDestination;
import com.alibaba.otter.canal.connector.core.producer.MQMessageUtils; import com.alibaba.otter.canal.connector.core.producer.MQMessageUtils;
@@ -249,13 +249,15 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro
FlatMessage flatMessagePart = partitionFlatMessage[i]; FlatMessage flatMessagePart = partitionFlatMessage[i];
if (flatMessagePart != null) { if (flatMessagePart != null) {
records.add(new ProducerRecord<>(topicName, i, null, JSON.toJSONBytes(flatMessagePart, records.add(new ProducerRecord<>(topicName, i, null, JSON.toJSONBytes(flatMessagePart,
JSONWriter.Feature.WriteNulls))); JSONWriter.Feature.WriteNulls,
JSONWriter.Feature.LargeObject)));
} }
} }
} else { } else {
final int partition = mqDestination.getPartition() != null ? mqDestination.getPartition() : 0; final int partition = mqDestination.getPartition() != null ? mqDestination.getPartition() : 0;
records.add(new ProducerRecord<>(topicName, partition, null, JSON.toJSONBytes(flatMessage, records.add(new ProducerRecord<>(topicName, partition, null, JSON.toJSONBytes(flatMessage,
JSONWriter.Feature.WriteNulls))); JSONWriter.Feature.WriteNulls,
JSONWriter.Feature.LargeObject)));
} }
} }
} }

3
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java
View File

@@ -13,6 +13,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.fastjson2.JSONWriter.Feature; import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate; import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
import com.alibaba.otter.canal.common.utils.NamedThreadFactory; import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
@@ -330,7 +331,7 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
try { try {
MessageId msgResultId = producer.newMessage() MessageId msgResultId = producer.newMessage()
.property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition)) .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition))
.value(JSON.toJSONBytes(f, Feature.WriteNulls)) .value(JSON.toJSONBytes(f, Feature.WriteNulls, JSONWriter.Feature.LargeObject))
.send() .send()
// //
; ;

3
connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java
View File

@@ -165,7 +165,8 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
// 串行分区 // 串行分区
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(datas, messageSub.getId()); List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(datas, messageSub.getId());
for (FlatMessage flatMessage : flatMessages) { for (FlatMessage flatMessage : flatMessages) {
byte[] message = JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls); byte[] message = JSON
.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject);
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("send message:{} to destination:{}", message, canalDestination.getCanalDestination()); logger.debug("send message:{} to destination:{}", message, canalDestination.getCanalDestination());
} }

4
connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java
View File

@@ -263,7 +263,7 @@ import java.util.stream.Collectors;
List<Message> messages = flatMessagePart.stream() List<Message> messages = flatMessagePart.stream()
.map(flatMessage -> new Message(topicName, .map(flatMessage -> new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls))) JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject)))
.collect(Collectors.toList()); .collect(Collectors.toList());
// 批量发送 // 批量发送
sendMessage(messages, index); sendMessage(messages, index);
@@ -278,7 +278,7 @@ import java.util.stream.Collectors;
List<Message> messages = flatMessages.stream() List<Message> messages = flatMessages.stream()
.map(flatMessage -> new Message(topicName, .map(flatMessage -> new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls))) JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls,JSONWriter.Feature.LargeObject)))
.collect(Collectors.toList()); .collect(Collectors.toList());
// 批量发送 // 批量发送
sendMessage(messages, partition); sendMessage(messages, partition);