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

Merge canal repo master

Browse Source
This commit is contained in:
duhengforever
2018-09-27 17:09:00 +08:00
parent a37a7ff142 ff592758cf
commit 5ae16b1c7c
73 changed files with 12936 additions and 10729 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java
View File

@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.common;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.CodedOutputStream;
@@ -31,14 +32,14 @@ public class CanalMessageSerializer {
// packet size
int size = 0;
size += CodedOutputStream.computeEnumSize(3,
CanalPacket.PacketType.MESSAGES.getNumber());
PacketType.MESSAGES.getNumber());
size += CodedOutputStream.computeTagSize(5)
+ CodedOutputStream.computeRawVarint32Size(messageSize)
+ messageSize;
// build data
byte[] body = new byte[size];
CodedOutputStream output = CodedOutputStream.newInstance(body);
output.writeEnum(3, CanalPacket.PacketType.MESSAGES.getNumber());
output.writeEnum(3, PacketType.MESSAGES.getNumber());
output.writeTag(5, WireFormat.WIRETYPE_LENGTH_DELIMITED);
output.writeRawVarint32(messageSize);
@@ -56,7 +57,7 @@ public class CanalMessageSerializer {
}
CanalPacket.Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
packetBuilder.setType(CanalPacket.PacketType.MESSAGES);
packetBuilder.setType(PacketType.MESSAGES);
packetBuilder.setBody(messageBuilder.build().toByteString());
return packetBuilder.build().toByteArray();
}

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

@@ -1,9 +1,8 @@
package com.alibaba.otter.canal.common;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.Map;
/**
* kafka 配置项
@@ -13,23 +12,26 @@ import java.util.Set;
*/
public class MQProperties {
private String servers = "localhost:6667";
private int retries = 0;
private int batchSize = 16384;
private int lingerMs = 1;
private long bufferMemory = 33554432L;
private boolean filterTransactionEntry = true;
private int canalBatchSize = 5;
private String producerGroup = "Canal-Producer";
private String servers = "localhost:6667";
private int retries = 0;
private int batchSize = 16384;
private int lingerMs = 1;
private long bufferMemory = 33554432L;
private boolean filterTransactionEntry = true;
private String producerGroup = "Canal-Producer";
private int canalBatchSize = 50;
private Long canalGetTimeout;
private boolean flatMessage = true;
private List<CanalDestination> canalDestinations = new ArrayList<CanalDestination>();
private List<CanalDestination> canalDestinations = new ArrayList<CanalDestination>();
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;
@@ -55,54 +57,23 @@ public class MQProperties {
this.partition = partition;
}
public Set<Topic> getTopics() {
return topics;
public Integer getPartitionsNum() {
return partitionsNum;
}
public void setTopics(Set<Topic> topics) {
this.topics = topics;
public void setPartitionsNum(Integer partitionsNum) {
this.partitionsNum = partitionsNum;
}
public Map<String, String> getPartitionHash() {
return partitionHash;
}
public void setPartitionHash(Map<String, String> partitionHash) {
this.partitionHash = partitionHash;
}
}
public static class Topic {
private String topic;
private Integer partition;
public String getTopic() {
return topic;
}
public void setTopic(String topic) {
this.topic = topic;
}
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 String getServers() {
return servers;
@@ -152,6 +123,22 @@ public class MQProperties {
this.canalBatchSize = canalBatchSize;
}
public Long getCanalGetTimeout() {
return canalGetTimeout;
}
public void setCanalGetTimeout(Long canalGetTimeout) {
this.canalGetTimeout = canalGetTimeout;
}
public boolean getFlatMessage() {
return flatMessage;
}
public void setFlatMessage(boolean flatMessage) {
this.flatMessage = flatMessage;
}
public List<CanalDestination> getCanalDestinations() {
return canalDestinations;
}

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

@@ -1,10 +1,12 @@
package com.alibaba.otter.canal.kafka;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.spi.CanalMQProducer;
import java.io.IOException;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -12,8 +14,6 @@ import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.protocol.Message;
/**
* kafka producer 主操作类
*
@@ -26,8 +26,13 @@ public class CanalKafkaProducer implements CanalMQProducer {
private Producer<String, Message> producer;
private Producer<String, String> producer2; // 用于扁平message的数据投递
private MQProperties kafkaProperties;
@Override
public void init(MQProperties kafkaProperties) {
this.kafkaProperties = kafkaProperties;
Properties properties = new Properties();
properties.put("bootstrap.servers", kafkaProperties.getServers());
properties.put("acks", "all");
@@ -36,15 +41,27 @@ public class CanalKafkaProducer implements CanalMQProducer {
properties.put("linger.ms", kafkaProperties.getLingerMs());
properties.put("buffer.memory", kafkaProperties.getBufferMemory());
properties.put("key.serializer", StringSerializer.class.getName());
properties.put("value.serializer", MessageSerializer.class.getName());
producer = new KafkaProducer<String, Message>(properties);
if (!kafkaProperties.getFlatMessage()) {
properties.put("value.serializer", MessageSerializer.class.getName());
producer = new KafkaProducer<String, Message>(properties);
} else {
properties.put("value.serializer", StringSerializer.class.getName());
producer2 = new KafkaProducer<String, String>(properties);
}
// producer.initTransactions();
}
@Override
public void stop() {
try {
logger.info("## stop the kafka producer");
producer.close();
if (producer != null) {
producer.close();
}
if (producer2 != null) {
producer2.close();
}
} catch (Throwable e) {
logger.warn("##something goes wrong when stopping kafka producer:", e);
} finally {
@@ -53,16 +70,72 @@ public class CanalKafkaProducer implements CanalMQProducer {
}
@Override
public void send(MQProperties.Topic topic, Message message) throws IOException {
ProducerRecord<String, Message> record;
if (topic.getPartition() != null) {
record = new ProducerRecord<String, Message>(topic.getTopic(), topic.getPartition(), null, message);
} else {
record = new ProducerRecord<String, Message>(topic.getTopic(), message);
}
producer.send(record);
if (logger.isDebugEnabled()) {
logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString());
public void send(MQProperties.CanalDestination canalDestination, Message message, Callback callback) {
try {
// producer.beginTransaction();
if (!kafkaProperties.getFlatMessage()) {
ProducerRecord<String, Message> record;
if (canalDestination.getPartition() != null) {
record = new ProducerRecord<String, Message>(canalDestination.getTopic(),
canalDestination.getPartition(),
null,
message);
} else {
record = new ProducerRecord<String, Message>(canalDestination.getTopic(), 0, null, message);
}
producer.send(record);
} else {
// 发送扁平数据json
List<FlatMessage> flatMessages = FlatMessage.messageConverter(message);
if (flatMessages != null) {
for (FlatMessage flatMessage : flatMessages) {
if (canalDestination.getPartition() != null) {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination
.getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage));
producer2.send(record);
} else {
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 flatMessagePart = partitionFlatMessage[i];
if (flatMessagePart != null) {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(
canalDestination.getTopic(),
i,
null,
JSON.toJSONString(flatMessagePart));
producer2.send(record);
}
}
} else {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(
canalDestination.getTopic(),
0,
null,
JSON.toJSONString(flatMessage));
producer2.send(record);
}
}
}
}
}
// producer.commitTransaction();
callback.commit();
if (logger.isDebugEnabled()) {
logger.debug("send message to kafka topic: {}", canalDestination.getTopic());
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
// producer.abortTransaction();
callback.rollback();
}
}
}

15
server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java
View File

@@ -1,20 +1,9 @@
package com.alibaba.otter.canal.kafka;
import com.alibaba.otter.canal.common.CanalMessageSerializer;
import java.util.List;
import java.util.Map;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.CodedOutputStream;
import com.google.protobuf.WireFormat;
import java.util.Map;
import org.apache.kafka.common.serialization.Serializer;
/**
* Kafka Message类的序列化

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

@@ -35,17 +35,24 @@ public class CanalRocketMQProducer implements CanalMQProducer {
}
@Override
public void send(final MQProperties.Topic topic, com.alibaba.otter.canal.protocol.Message data) {
public void send(final MQProperties.CanalDestination destination, com.alibaba.otter.canal.protocol.Message data,
Callback callback) {
try {
Message message = new Message(topic.getTopic(), CanalMessageSerializer.serializer(data));
Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data));
this.defaultMQProducer.send(message, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
return mqs.get(topic.getPartition());
int partition = 0;
if (destination.getPartition() != null) {
partition = destination.getPartition();
}
return mqs.get(partition);
}
}, null);
callback.commit();
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
logger.error("Send message error!", e);
callback.rollback();
}
}

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

@@ -1,6 +1,7 @@
package com.alibaba.otter.canal.server;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.kafka.CanalKafkaProducer;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
@@ -9,17 +10,18 @@ import java.io.FileInputStream;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
public class CanalMQStarter {
private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class);
private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class);
private static final String CLASSPATH_URL_PREFIX = "classpath:";
private volatile boolean running = false;
private volatile boolean running = false;
private ExecutorService executorService;
@@ -27,7 +29,7 @@ public class CanalMQStarter {
private MQProperties properties;
public CanalMQStarter(CanalMQProducer canalMQProducer){
public CanalMQStarter(CanalMQProducer canalMQProducer) {
this.canalMQProducer = canalMQProducer;
}
@@ -51,6 +53,7 @@ public class CanalMQStarter {
if (properties.isFilterTransactionEntry()) {
System.setProperty("canal.instance.filter.transaction.entry", "true");
}
// 对应每个instance启动一个worker线程
List<MQProperties.CanalDestination> destinations = properties.getCanalDestinations();
@@ -95,8 +98,8 @@ public class CanalMQStarter {
while (!running)
;
logger.info("## start the canal consumer: {}.", destination.getCanalDestination());
CanalServerWithEmbedded server = CanalServerWithEmbedded.instance();
ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, "");
final CanalServerWithEmbedded server = CanalServerWithEmbedded.instance();
final ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, "");
while (running) {
try {
if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) {
@@ -111,27 +114,41 @@ public class CanalMQStarter {
logger.info("## the canal consumer {} is running now ......", destination.getCanalDestination());
while (running) {
Message message = server.getWithoutAck(clientIdentity, properties.getCanalBatchSize()); // 获取指定数量的数据
long batchId = message.getId();
Message message;
if (properties.getCanalGetTimeout() != null) {
message = server.getWithoutAck(clientIdentity,
properties.getCanalBatchSize(),
properties.getCanalGetTimeout(),
TimeUnit.MILLISECONDS);
} else {
message = server.getWithoutAck(clientIdentity, properties.getCanalBatchSize());
}
final long batchId = message.getId();
try {
int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size();
if (batchId != -1 && size != 0) {
if (!StringUtils.isEmpty(destination.getTopic())) {
MQProperties.Topic topic = new MQProperties.Topic();
topic.setTopic(destination.getTopic());
topic.setPartition(destination.getPartition());
destination.getTopics().add(topic);
}
for (MQProperties.Topic topic : destination.getTopics()) {
canalMQProducer.send(topic, message); // 发送message到所有topic
canalMQProducer.send(destination, message, new CanalKafkaProducer.Callback() {
@Override
public void commit() {
server.ack(clientIdentity, batchId); // 提交确认
}
@Override
public void rollback() {
server.rollback(clientIdentity, batchId);
}
}); // 发送message到topic
} else {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
// ignore
}
}
if (batchId != -1) {
server.ack(clientIdentity, batchId); // 提交确认
}
} catch (Exception e) {
server.rollback(clientIdentity);
logger.error(e.getMessage(), e);
}
}

7
server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java
View File

@@ -311,9 +311,10 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
}
if (CollectionUtils.isEmpty(events.getEvents())) {
logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result is null",
clientIdentity.getClientId(),
batchSize);
// logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result
// is null",
// clientIdentity.getClientId(),
// batchSize);
return new Message(-1, true, new ArrayList()); // 返回空包，避免生成batchId，浪费性能
} else {
// 记录到流式信息

3
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java
View File

@@ -3,7 +3,6 @@ package com.alibaba.otter.canal.server.netty.handler;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.jboss.netty.buffer.ChannelBuffer;
@@ -34,6 +33,7 @@ import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.netty.NettyUtils;
import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator;
import com.google.protobuf.ByteString;
import com.google.protobuf.CodedOutputStream;
import com.google.protobuf.WireFormat;
@@ -50,6 +50,7 @@ public class SessionHandler extends SimpleChannelHandler {
this.embeddedServer = embeddedServer;
}
@SuppressWarnings("deprecation")
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
logger.info("message receives in session handler...");
long start = System.nanoTime();

18
server/src/main/java/com/alibaba/otter/canal/server/netty/listener/ChannelFutureAggregator.java
View File

@@ -2,7 +2,7 @@ package com.alibaba.otter.canal.server.netty.listener;
import com.alibaba.otter.canal.protocol.CanalPacket;
import com.google.common.base.Preconditions;
import com.google.protobuf.GeneratedMessage;
import com.google.protobuf.GeneratedMessageV3;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
@@ -16,19 +16,19 @@ public class ChannelFutureAggregator implements ChannelFutureListener {
private ClientRequestResult result;
public ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency, boolean empty) {
public ChannelFutureAggregator(String destination, GeneratedMessageV3 request, CanalPacket.PacketType type, int amount, long latency, boolean empty) {
this(destination, request, type, amount, latency, empty, (short) 0);
}
public ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency) {
public ChannelFutureAggregator(String destination, GeneratedMessageV3 request, CanalPacket.PacketType type, int amount, long latency) {
this(destination, request, type, amount, latency, false, (short) 0);
}
public ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency, short errorCode) {
public ChannelFutureAggregator(String destination, GeneratedMessageV3 request, CanalPacket.PacketType type, int amount, long latency, short errorCode) {
this(destination, request, type, amount, latency, false, errorCode);
}
private ChannelFutureAggregator(String destination, GeneratedMessage request, CanalPacket.PacketType type, int amount, long latency, boolean empty, short errorCode) {
private ChannelFutureAggregator(String destination, GeneratedMessageV3 request, CanalPacket.PacketType type, int amount, long latency, boolean empty, short errorCode) {
this.result = new ClientRequestResult.Builder()
.destination(destination)
.type(type)
@@ -56,7 +56,7 @@ public class ChannelFutureAggregator implements ChannelFutureListener {
private String destination;
private CanalPacket.PacketType type;
private GeneratedMessage request;
private GeneratedMessageV3 request;
private int amount;
private long latency;
private short errorCode;
@@ -81,7 +81,7 @@ public class ChannelFutureAggregator implements ChannelFutureListener {
private String destination;
private CanalPacket.PacketType type;
private GeneratedMessage request;
private GeneratedMessageV3 request;
private int amount;
private long latency;
private short errorCode;
@@ -98,7 +98,7 @@ public class ChannelFutureAggregator implements ChannelFutureListener {
return this;
}
Builder request(GeneratedMessage request) {
Builder request(GeneratedMessageV3 request) {
this.request = request;
return this;
}
@@ -153,7 +153,7 @@ public class ChannelFutureAggregator implements ChannelFutureListener {
return type;
}
public GeneratedMessage getRequest() {
public GeneratedMessageV3 getRequest() {
return request;
}

12
server/src/main/java/com/alibaba/otter/canal/spi/CanalMQProducer.java
View File

@@ -1,6 +1,7 @@
package com.alibaba.otter.canal.spi;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.kafka.CanalKafkaProducer;
import com.alibaba.otter.canal.protocol.Message;
import java.io.IOException;
@@ -15,14 +16,21 @@ public interface CanalMQProducer {
/**
* Send canal message to related topic
*
* @param topic MQ topic
* @param canalDestination canal mq destination
* @param message canal message
* @throws IOException
*/
void send(MQProperties.Topic topic, Message message) throws IOException;
void send(MQProperties.CanalDestination canalDestination, Message message, Callback callback) throws IOException;
/**
* Stop MQ producer service
*/
void stop();
interface Callback {
void commit();
void rollback();
}
}

4
server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java
View File

@@ -66,6 +66,8 @@ public class CanalServerWithNettyTest {
@Test
public void testAuth() {
try {
SocketChannel channel = SocketChannel.open();
channel.connect(new InetSocketAddress("127.0.0.1", 1088));
@@ -80,7 +82,7 @@ public class CanalServerWithNettyTest {
}
//
Handshake handshake = Handshake.parseFrom(p.getBody());
System.out.println(handshake.getSupportedCompressionsList());
System.out.println(handshake.getSupportedCompressions());
//
ClientAuth ca = ClientAuth.newBuilder()
.setUsername("")