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 branch 'develop-tmp'

Browse Source 
# Conflicts:
#	kafka-client/pom.xml
#	kafka/pom.xml
This commit is contained in:
machey
2018-08-23 23:59:20 +08:00
parent 899592ce2a 9e3d4a7ffe
commit 4fe6c36c49
36 changed files with 136 additions and 848 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
kafka-client/pom.xml
View File

@@ -1,109 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>canal</artifactId>
<groupId>com.alibaba.otter</groupId>
<version>1.1.1-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.kafka.client</artifactId>
<packaging>jar</packaging>
<name>canal kafka client module for otter ${project.version}</name>
<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.protocol</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>1.1.1</version>
</dependency>
<!-- junit -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
</dependencies>
<profiles>
<profile>
<id>dev</id>
<activation>
<activeByDefault>true</activeByDefault>
<property>
<name>env</name>
<value>!javadoc</value>
</property>
</activation>
</profile>
<profile>
<id>javadoc</id>
<activation>
<property>
<name>env</name>
<value>javadoc</value>
</property>
</activation>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId>
<version>2.9.1</version>
<executions>
<execution>
<id>attach-javadocs</id>
<phase>package</phase>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
<configuration>
<aggregate>true</aggregate>
<show>public</show>
<nohelp>true</nohelp>
<header>${project.artifactId}-${project.version}</header>
<footer>${project.artifactId}-${project.version}</footer>
<doctitle>${project.artifactId}-${project.version}</doctitle>
<links>
<link>https://github.com/alibaba/canal</link>
</links>
<outputDirectory>${project.build.directory}/apidocs/apidocs/${project.version}</outputDirectory>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-scm-publish-plugin</artifactId>
<version>1.0-beta-2</version>
<executions>
<execution>
<id>attach-javadocs</id>
<phase>package</phase>
<goals>
<goal>publish-scm</goal>
</goals>
</execution>
</executions>
<configuration>
<checkoutDirectory>${project.build.directory}/scmpublish</checkoutDirectory>
<checkinComment>Publishing javadoc for ${project.artifactId}:${project.version}</checkinComment>
<content>${project.build.directory}/apidocs</content>
<skipDeletedFiles>true</skipDeletedFiles>
<pubScmUrl>scm:git:git@github.com:alibaba/canal.git</pubScmUrl>
<scmBranch>gh-pages</scmBranch>
</configuration>
</plugin>
</plugins>
</build>
</profile>
</profiles>
</project>

242
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java
View File

@@ -1,242 +0,0 @@
package com.alibaba.otter.canal.kafka.client;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.common.utils.BooleanMutex;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.kafka.client.running.ClientRunningData;
import com.alibaba.otter.canal.kafka.client.running.ClientRunningListener;
import com.alibaba.otter.canal.kafka.client.running.ClientRunningMonitor;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
/**
* canal kafka 数据操作客户端
*
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
public class KafkaCanalConnector {
private KafkaConsumer<String, Message> kafkaConsumer;
private String topic;
private Integer partition;
private Properties properties;
private ClientRunningMonitor runningMonitor; // 运行控制
private ZkClientx zkClientx;
private BooleanMutex mutex = new BooleanMutex(false);
private volatile boolean connected = false;
private volatile boolean running = false;
public KafkaCanalConnector(String zkServers, String servers, String topic, Integer partition, String groupId) {
this.topic = topic;
this.partition = partition;
properties = new Properties();
properties.put("bootstrap.servers", servers);
properties.put("group.id", groupId);
properties.put("enable.auto.commit", false);
properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "latest"); //如果没有offset则从最后的offset开始读
properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置
properties.put("session.timeout.ms", "30000"); // 默认为30秒
properties.put("max.poll.records", "1"); // 所以一次只取一条数据
properties.put("key.deserializer", StringDeserializer.class.getName());
properties.put("value.deserializer", MessageDeserializer.class.getName());
if (zkServers != null) {
zkClientx = new ZkClientx(zkServers);
ClientRunningData clientData = new ClientRunningData();
clientData.setGroupId(groupId);
clientData.setAddress(AddressUtils.getHostIp());
runningMonitor = new ClientRunningMonitor();
runningMonitor.setTopic(topic);
runningMonitor.setZkClient(zkClientx);
runningMonitor.setClientData(clientData);
runningMonitor.setListener(new ClientRunningListener() {
public void processActiveEnter() {
mutex.set(true);
}
public void processActiveExit() {
mutex.set(false);
}
});
}
}
/**
* 重新设置sessionTime
*
* @param timeout
* @param unit
*/
public void setSessionTimeout(Long timeout, TimeUnit unit) {
long t = unit.toMillis(timeout);
properties.put("request.timeout.ms", String.valueOf(t + 60000));
properties.put("session.timeout.ms", String.valueOf(t));
}
/**
* 打开连接
*/
public void connect() {
if (connected) {
return;
}
if (runningMonitor != null) {
if (!runningMonitor.isStart()) {
runningMonitor.start();
}
}
connected = true;
if (kafkaConsumer == null) {
kafkaConsumer = new KafkaConsumer<String, Message>(properties);
}
}
/**
* 关闭链接
*/
public void disconnnect() {
kafkaConsumer.close();
connected = false;
if (runningMonitor.isStart()) {
runningMonitor.stop();
}
}
private void waitClientRunning() {
try {
if (zkClientx != null) {
if (!connected) {// 未调用connect
throw new CanalClientException("should connect first");
}
running = true;
mutex.get();// 阻塞等待
} else {
// 单机模式直接设置为running
running = true;
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new CanalClientException(e);
}
}
public boolean checkValid() {
if (zkClientx != null) {
return mutex.state();
} else {
return true;// 默认都放过
}
}
/**
* 订阅topic
*/
public void subscribe() {
waitClientRunning();
if (!running) {
return;
}
if (partition == null) {
kafkaConsumer.subscribe(Collections.singletonList(topic));
} else {
TopicPartition topicPartition = new TopicPartition(topic, partition);
kafkaConsumer.assign(Collections.singletonList(topicPartition));
}
}
/**
* 取消订阅
*/
public void unsubscribe() {
waitClientRunning();
if (!running) {
return;
}
kafkaConsumer.unsubscribe();
}
/**
* 获取数据，自动进行确认
*
* @return
*/
public Message get() {
return get(100L, TimeUnit.MILLISECONDS);
}
public Message get(Long timeout, TimeUnit unit) {
waitClientRunning();
if (!running) {
return null;
}
Message message = getWithoutAck(timeout, unit);
this.ack();
return message;
}
public Message getWithoutAck() {
return getWithoutAck(100L, TimeUnit.MILLISECONDS);
}
/**
* 获取数据，不进行确认，等待处理完成手工确认
*
* @return
*/
public Message getWithoutAck(Long timeout, TimeUnit unit) {
waitClientRunning();
if (!running) {
return null;
}
ConsumerRecords<String, Message> records = kafkaConsumer.poll(unit.toMillis(timeout)); // 基于配置，最多只能poll到一条数据
if (!records.isEmpty()) {
return records.iterator().next().value();
}
return null;
}
/**
* 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败
*/
public void ack() {
waitClientRunning();
if (!running) {
return;
}
kafkaConsumer.commitSync();
}
public void stopRunning() {
if (running) {
running = false; // 设置为非running状态
if (!mutex.state()) {
mutex.set(true); // 中断阻塞
}
}
}
}

62
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java
View File

@@ -1,62 +0,0 @@
package com.alibaba.otter.canal.kafka.client;
/**
* canal kafka connectors创建工具类
*
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
public class KafkaCanalConnectors {
/**
* 创建kafka客户端链接，独立运行不注册zk信息
*
* @param servers
* @param topic
* @param partition
* @param groupId
* @return
*/
public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) {
return new KafkaCanalConnector(null, servers, topic, partition, groupId);
}
/**
* 创建kafka客户端链接，独立运行不注册zk信息
*
* @param servers
* @param topic
* @param groupId
* @return
*/
public static KafkaCanalConnector newKafkaConnector(String servers, String topic, String groupId) {
return new KafkaCanalConnector(null, servers, topic, null, groupId);
}
/**
* 创建kafka客户端链接
*
* @param zkServers
* @param servers
* @param topic
* @param partition
* @param groupId
* @return
*/
public static KafkaCanalConnector newKafkaConnector(String zkServers, String servers, String topic, Integer partition, String groupId) {
return new KafkaCanalConnector(zkServers, servers, topic, partition, groupId);
}
/**
* 创建kafka客户端链接
*
* @param zkServers
* @param servers
* @param topic
* @param groupId
* @return
*/
public static KafkaCanalConnector newKafkaConnector(String zkServers, String servers, String topic, String groupId) {
return new KafkaCanalConnector(zkServers, servers, topic, null, groupId);
}
}

64
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java
View File

@@ -1,64 +0,0 @@
package com.alibaba.otter.canal.kafka.client;
import java.util.Map;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.protobuf.ByteString;
/**
* Kafka Message类的反序列化
*
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
public class MessageDeserializer implements Deserializer<Message> {
private static Logger logger = LoggerFactory.getLogger(MessageDeserializer.class);
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
}
@Override
public Message deserialize(String topic, byte[] data) {
try {
if (data == null) {
return null;
}
else {
CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data);
switch (p.getType()) {
case MESSAGES: {
if (!p.getCompression().equals(CanalPacket.Compression.NONE)) {
throw new CanalClientException("compression is not supported in this connector");
}
CanalPacket.Messages messages = CanalPacket.Messages.parseFrom(p.getBody());
Message result = new Message(messages.getBatchId());
for (ByteString byteString : messages.getMessagesList()) {
result.addEntry(CanalEntry.Entry.parseFrom(byteString));
}
return result;
}
default:
break;
}
}
} catch (Exception e) {
logger.error("Error when deserializing byte[] to message ");
}
return null;
}
@Override
public void close() {
// nothing to do
}
}

39
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningData.java
View File

@@ -1,39 +0,0 @@
package com.alibaba.otter.canal.kafka.client.running;
/**
* client running状态信息
*
* @author machengyuan 2018-06-20 下午04:10:12
* @version 1.0.0
*/
public class ClientRunningData {
private String groupId;
private String address;
private boolean active = true;
public String getGroupId() {
return groupId;
}
public void setGroupId(String groupId) {
this.groupId = groupId;
}
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
public boolean isActive() {
return active;
}
public void setActive(boolean active) {
this.active = active;
}
}

21
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningListener.java
View File

@@ -1,21 +0,0 @@
package com.alibaba.otter.canal.kafka.client.running;
/**
* client running状态信息
*
* @author machengyuan 2018-06-20 下午04:10:12
* @version 1.0.0
*/
public interface ClientRunningListener {
/**
* 触发现在轮到自己做为active
*/
public void processActiveEnter();
/**
* 触发一下当前active模式失败
*/
public void processActiveExit();
}

281
kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningMonitor.java
View File

@@ -1,281 +0,0 @@
package com.alibaba.otter.canal.kafka.client.running;
import java.text.MessageFormat;
import java.util.Random;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.exception.ZkException;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.common.utils.BooleanMutex;
import com.alibaba.otter.canal.common.utils.JsonUtils;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
/**
* kafka client running状态信息
*
* @author machengyuan 2018-06-20 下午04:10:12
* @version 1.0.0
*/
public class ClientRunningMonitor extends AbstractCanalLifeCycle {
private static final String TOPIC_ROOT_NODE = ZookeeperPathUtils.CANAL_ROOT_NODE
+ ZookeeperPathUtils.ZOOKEEPER_SEPARATOR + "topics";
private static final String TOPIC_NODE = TOPIC_ROOT_NODE + ZookeeperPathUtils.ZOOKEEPER_SEPARATOR
+ "{0}";
private static final String TOPIC_CLIENTID_NODE = TOPIC_NODE + ZookeeperPathUtils.ZOOKEEPER_SEPARATOR
+ "{1}";
private static final String TOPIC_CLIENTID_RUNNING_NODE = TOPIC_CLIENTID_NODE
+ ZookeeperPathUtils.ZOOKEEPER_SEPARATOR
+ ZookeeperPathUtils.RUNNING_NODE;
private static String getTopicClientRunning(String topic, String groupId) {
return MessageFormat.format(TOPIC_CLIENTID_RUNNING_NODE, topic, groupId);
}
private static String getClientIdNodePath(String topic, String groupId) {
return MessageFormat.format(TOPIC_CLIENTID_NODE, topic, groupId);
}
private static final Logger logger = LoggerFactory.getLogger(ClientRunningMonitor.class);
private ZkClientx zkClient;
private String topic;
private ClientRunningData clientData;
private IZkDataListener dataListener;
private BooleanMutex mutex = new BooleanMutex(false);
private volatile boolean release = false;
private volatile ClientRunningData activeData;
private ScheduledExecutorService delayExector = Executors.newScheduledThreadPool(1);
private ClientRunningListener listener;
private int delayTime = 5;
private static Integer virtualPort;
public ClientRunningMonitor(){
if (virtualPort == null) {
Random rand = new Random();
virtualPort = rand.nextInt(9000) + 1000;
}
dataListener = new IZkDataListener() {
public void handleDataChange(String dataPath, Object data) throws Exception {
MDC.put("kafkaTopic", topic);
ClientRunningData runningData = JsonUtils.unmarshalFromByte((byte[]) data, ClientRunningData.class);
if (!isMine(runningData.getAddress())) {
mutex.set(false);
}
if (!runningData.isActive() && isMine(runningData.getAddress())) { // 说明出现了主动释放的操作，并且本机之前是active
release = true;
releaseRunning();// 彻底释放mainstem
}
activeData = (ClientRunningData) runningData;
}
public void handleDataDeleted(String dataPath) throws Exception {
MDC.put("kafkaTopic", topic);
mutex.set(false);
// 触发一下退出,可能是人为干预的释放操作或者网络闪断引起的session expired timeout
processActiveExit();
if (!release && activeData != null && isMine(activeData.getAddress())) {
// 如果上一次active的状态就是本机，则即时触发一下active抢占
initRunning();
} else {
// 否则就是等待delayTime，避免因网络瞬端或者zk异常，导致出现频繁的切换操作
delayExector.schedule(new Runnable() {
public void run() {
initRunning();
}
}, delayTime, TimeUnit.SECONDS);
}
}
};
}
public void start() {
super.start();
String path = getTopicClientRunning(this.topic, clientData.getGroupId());
zkClient.subscribeDataChanges(path, dataListener);
initRunning();
}
public void stop() {
super.stop();
String path = getTopicClientRunning(this.topic, clientData.getGroupId());
zkClient.unsubscribeDataChanges(path, dataListener);
releaseRunning(); // 尝试一下release
// Fix issue #697
if (delayExector != null) {
delayExector.shutdown();
}
}
// 改动记录：
// 1,在方法上加synchronized关键字,保证同步顺序执行;
// 2,判断Zk上已经存在的activeData是否是本机，是的话把mutex重置为true，否则会导致死锁
// 3,增加异常处理，保证出现异常时，running节点能被删除,否则会导致死锁
public synchronized void initRunning() {
if (!isStart()) {
return;
}
String path = getTopicClientRunning(this.topic, clientData.getGroupId());
// 序列化
byte[] bytes = JsonUtils.marshalToByte(clientData);
try {
mutex.set(false);
zkClient.create(path, bytes, CreateMode.EPHEMERAL);
processActiveEnter();// 触发一下事件
activeData = clientData;
mutex.set(true);
} catch (ZkNodeExistsException e) {
bytes = zkClient.readData(path, true);
if (bytes == null) {// 如果不存在节点，立即尝试一次
initRunning();
} else {
activeData = JsonUtils.unmarshalFromByte(bytes, ClientRunningData.class);
// 如果发现已经存在,判断一下是否自己,避免活锁
if (activeData.getAddress().contains(":") && isMine(activeData.getAddress())) {
mutex.set(true);
}
}
} catch (ZkNoNodeException e) {
zkClient.createPersistent(getClientIdNodePath(this.topic, clientData.getGroupId()), true); // 尝试创建父节点
initRunning();
} catch (Throwable t) {
logger.error(MessageFormat.format("There is an error when execute initRunning method, with destination [{0}].",
topic),
t);
// 出现任何异常尝试release
releaseRunning();
throw new CanalClientException("something goes wrong in initRunning method. ", t);
}
}
/**
* 阻塞等待自己成为active，如果自己成为active，立马返回
*
* @throws InterruptedException
*/
public void waitForActive() throws InterruptedException {
initRunning();
mutex.get();
}
/**
* 检查当前的状态
*/
public boolean check() {
String path = getTopicClientRunning(this.topic, clientData.getGroupId());
// ZookeeperPathUtils.getDestinationClientRunning(this.destination,
// clientData.getClientId());
try {
byte[] bytes = zkClient.readData(path);
ClientRunningData eventData = JsonUtils.unmarshalFromByte(bytes, ClientRunningData.class);
activeData = eventData;// 更新下为最新值
// 检查下nid是否为自己
boolean result = isMine(activeData.getAddress());
if (!result) {
logger.warn("canal is running in [{}] , but not in [{}]",
activeData.getAddress(),
clientData.getAddress());
}
return result;
} catch (ZkNoNodeException e) {
logger.warn("canal is not run any in node");
return false;
} catch (ZkInterruptedException e) {
logger.warn("canal check is interrupt");
Thread.interrupted();// 清除interrupt标记
return check();
} catch (ZkException e) {
logger.warn("canal check is failed");
return false;
}
}
public boolean releaseRunning() {
if (check()) {
String path = getTopicClientRunning(this.topic, clientData.getGroupId());
zkClient.delete(path);
mutex.set(false);
processActiveExit();
return true;
}
return false;
}
// ====================== helper method ======================
private boolean isMine(String address) {
return address.equals(clientData.getAddress());
}
private void processActiveEnter() {
if (listener != null) {
// 触发回调
listener.processActiveEnter();
this.clientData.setAddress(/* address */AddressUtils.getHostIp() + ":" + virtualPort);
String path = getTopicClientRunning(this.topic, clientData.getGroupId());
// 序列化
byte[] bytes = JsonUtils.marshalToByte(clientData);
zkClient.writeData(path, bytes);
}
}
private void processActiveExit() {
if (listener != null) {
listener.processActiveExit();
}
}
public void setListener(ClientRunningListener listener) {
this.listener = listener;
}
// ===================== setter / getter =======================
public void setTopic(String topic) {
this.topic = topic;
}
public void setClientData(ClientRunningData clientData) {
this.clientData = clientData;
}
public void setDelayTime(int delayTime) {
this.delayTime = delayTime;
}
public void setZkClient(ZkClientx zkClient) {
this.zkClient = zkClient;
}
}

26
kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java
View File

@@ -1,26 +0,0 @@
package com.alibaba.otter.canal.kafka.client.running;
import org.junit.Assert;
/**
* Kafka 测试基类
*
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
public abstract class AbstractKafkaTest {
public static String topic = "example";
public static Integer partition = null;
public static String groupId = "g4";
public static String servers = "slave1:6667,slave2:6667,slave3:6667";
public static String zkServers = "slave1:2181,slave2:2181,slave3:2181";
public void sleep(long time) {
try {
Thread.sleep(time);
} catch (InterruptedException e) {
Assert.fail(e.getMessage());
}
}
}

144
kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java
View File

@@ -1,144 +0,0 @@
package com.alibaba.otter.canal.kafka.client.running;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
/**
* Kafka client example
*
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
public class CanalKafkaClientExample {
protected final static Logger logger = LoggerFactory.getLogger(CanalKafkaClientExample.class);
private KafkaCanalConnector connector;
private static volatile boolean running = false;
private Thread thread = null;
private Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
public void uncaughtException(Thread t, Throwable e) {
logger.error("parse events has an error", e);
}
};
public CanalKafkaClientExample(String zkServers, String servers, String topic, Integer partition, String groupId) {
connector = KafkaCanalConnectors.newKafkaConnector(zkServers, servers, topic, partition, groupId);
}
public static void main(String[] args) {
try {
final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(AbstractKafkaTest.zkServers,
AbstractKafkaTest.servers,
AbstractKafkaTest.topic,
AbstractKafkaTest.partition,
AbstractKafkaTest.groupId);
logger.info("## start the kafka consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId);
kafkaCanalClientExample.start();
logger.info("## the canal kafka consumer is running now ......");
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
try {
logger.info("## stop the kafka consumer");
kafkaCanalClientExample.stop();
} catch (Throwable e) {
logger.warn("##something goes wrong when stopping kafka consumer:", e);
} finally {
logger.info("## kafka consumer is down.");
}
}
});
while (running)
;
} catch (Throwable e) {
logger.error("## Something goes wrong when starting up the kafka consumer:", e);
System.exit(0);
}
}
public void start() {
Assert.notNull(connector, "connector is null");
thread = new Thread(new Runnable() {
public void run() {
process();
}
});
thread.setUncaughtExceptionHandler(handler);
thread.start();
running = true;
}
public void stop() {
if (!running) {
return;
}
connector.stopRunning();
running = false;
if (thread != null) {
try {
thread.join();
} catch (InterruptedException e) {
// ignore
}
}
}
private void process() {
while (!running)
;
while (running) {
try {
connector.connect();
connector.subscribe();
while (running) {
try {
Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); // 获取message
if (message == null) {
continue;
}
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// }
} else {
// printSummary(message, batchId, size);
// printEntry(message.getEntries());
logger.info(message.toString());
}
connector.ack(); // 提交确认
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
try {
connector.unsubscribe();
} catch (WakeupException e) {
// No-op. Continue process
}
connector.disconnnect();
}
}

62
kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java
View File

@@ -1,62 +0,0 @@
package com.alibaba.otter.canal.kafka.client.running;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.common.errors.WakeupException;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
/**
* Kafka consumer获取Message的测试例子
*
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
public class KafkaClientRunningTest extends AbstractKafkaTest {
private Logger logger = LoggerFactory.getLogger(KafkaClientRunningTest.class);
private boolean running = true;
@Test
public void testKafkaConsumer() {
final ExecutorService executor = Executors.newFixedThreadPool(1);
final KafkaCanalConnector connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId);
executor.submit(new Runnable() {
@Override
public void run() {
connector.connect();
connector.subscribe();
while (running) {
try {
Message message = connector.getWithoutAck(3L, TimeUnit.SECONDS);
if (message != null) {
System.out.println(message);
}
connector.ack();
} catch (WakeupException e) {
// ignore
}
}
connector.unsubscribe();
connector.disconnnect();
}
});
sleep(60000);
running = false;
executor.shutdown();
logger.info("shutdown completed");
}
}

19
kafka-client/src/test/resources/logback.xml
View File

@@ -1,19 +0,0 @@
<configuration scan="true" scanPeriod=" 5 seconds">
<jmxConfigurator />
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n
</pattern>
</encoder>
</appender>
<logger name="org.apache.kafka" additivity="false">
<level value="ERROR" />
<appender-ref ref="STDOUT" />
</logger>
<root level="INFO">
<appender-ref ref="STDOUT"/>
</root>
</configuration>