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

Modify canal-client config and polish consumer implementation

Browse Source
This commit is contained in:
duhengforever
2018-09-30 09:40:05 +08:00
parent af03f07cc7
commit fa8588f766
3 changed files with 10 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
View File

@@ -26,6 +26,7 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId,
List<List<CanalOuterAdapter>> canalOuterAdapters) { List<List<CanalOuterAdapter>> canalOuterAdapters) {
logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId);
this.canalOuterAdapters = canalOuterAdapters; this.canalOuterAdapters = canalOuterAdapters;
this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
this.topic = topic; this.topic = topic;

6
client-launcher/src/main/resources/canal-client.yml
View File

@@ -1,6 +1,6 @@
#canalServerHost: 127.0.0.1:11111 #canalServerHost: 127.0.0.1:11111
#zookeeperHosts: slave1:2181 #zookeeperHosts: slave1:2181
bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers:host1:9876;host2:9876
flatMessage: true flatMessage: true
#canalInstances: #canalInstances:
@@ -14,9 +14,9 @@ flatMessage: true
mqTopics: mqTopics:
- mqMode: rocketmq - mqMode: rocketmq
- topic: example topic: example
groups: groups:
- groupId: example_g1 - groupId: example
outAdapters: outAdapters:
- name: logger - name: logger
# - name: hbase # - name: hbase

11
client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java
View File

@@ -43,6 +43,9 @@ public class RocketMQCanalConnector implements CanalConnector {
@Override @Override
public void connect() throws CanalClientException { public void connect() throws CanalClientException {
rocketMQConsumer = new DefaultMQPushConsumer(groupName); rocketMQConsumer = new DefaultMQPushConsumer(groupName);
if (!StringUtils.isBlank(nameServer)) {
rocketMQConsumer.setNamesrvAddr(nameServer);
}
} }
@Override @Override
@@ -61,18 +64,16 @@ public class RocketMQCanalConnector implements CanalConnector {
return; return;
} }
try { try {
rocketMQConsumer.subscribe(topic, "*");
if (rocketMQConsumer == null) { if (rocketMQConsumer == null) {
rocketMQConsumer = new DefaultMQPushConsumer(groupName); this.connect();
if (!StringUtils.isBlank(nameServer)) {
rocketMQConsumer.setNamesrvAddr(nameServer);
}
} }
rocketMQConsumer.subscribe(topic, "*");
rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() { rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() {
@Override @Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts, public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts,
ConsumeOrderlyContext context) { ConsumeOrderlyContext context) {
context.setAutoCommit(true); context.setAutoCommit(true);
logger.info("xxxx");
boolean isSuccess = process(messageExts); boolean isSuccess = process(messageExts);
if (isSuccess) { if (isSuccess) {
logger.info("Dispatch success!"); logger.info("Dispatch success!");