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

Resolve conflict

Browse Source
This commit is contained in:
duhengforever
2018-10-28 17:00:33 +08:00
parent ebaf78f5aa
commit 86fd77f9c4
1 changed files with 15 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,9 +15,8 @@ import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
/** /**
* kafka对应的client适配器工作线程 * rocketmq对应的client适配器工作线程
* *
* @author rewerma 2018-8-19 下午11:30:49
* @version 1.0.0 * @version 1.0.0
*/ */
public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
@@ -65,13 +64,19 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
message = connector.getFlatMessageWithoutAck(); message = connector.getFlatMessageWithoutAck();
} }
if (message != null) { if (message != null) {
<<<<<<< HEAD:client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
final Object msg = message; final Object msg = message;
executor.submit(new Runnable() { executor.submit(new Runnable() {
@Override @Override
public void run() { public void run() {
try { try {
if (logger.isDebugEnabled()) {
logger.debug("topic: {} batchId: {} batchSize: {} ",
topic,
message.getId(),
message.getEntries().size());
}
if (msg != null) { if (msg != null) {
long begin = System.currentTimeMillis();
if (msg instanceof Message) { if (msg instanceof Message) {
Message receive = (Message) msg; Message receive = (Message) msg;
writeOut(receive, topic); writeOut(receive, topic);
@@ -81,31 +86,17 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
writeOut(receive); writeOut(receive);
connector.ack(receive.getId()); connector.ack(receive.getId());
} }
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
=======
executor.submit(() -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("topic: {} batchId: {} batchSize: {} ",
topic,
message.getId(),
message.getEntries().size());
}
long begin = System.currentTimeMillis();
writeOut(message);
long now = System.currentTimeMillis(); long now = System.currentTimeMillis();
if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { if ((now - begin) > 5 * 60 * 1000) {
logger.error("topic: {} batchId {} elapsed time: {} ms", logger.error("topic: {} batchId {} elapsed time: {} ms",
topic, topic,
message.getId(), message.getId(),
now - begin); now - begin);
} }
}
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
>>>>>>> 2c6bd3ee1b23f30fdb2b1f748805284a4fed872c:client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java }
} }
connector.ack(message.getId()); connector.ack(message.getId());
}); });