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

adapter增加同步异常重试

Browse Source
This commit is contained in:
mcy
2018-12-23 18:16:47 +08:00
parent 34b5f9c927
commit 354d64069f
8 changed files with 233 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,11 +2,7 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -80,16 +76,17 @@ public abstract class AbstractCanalAdapterWorker {
futures.forEach(future -> {
try {
if (!future.get()) {
logger.error("Outer adapter write failed");
throw new RuntimeException("Outer adapter sync failed! ");
}
} catch (InterruptedException | ExecutionException e) {
// ignore
} catch (Exception e) {
future.cancel(true);
throw new RuntimeException(e);
}
});
});
}
private void writeOut(final List<FlatMessage> flatMessages) {
protected void writeOut(final List<FlatMessage> flatMessages) {
List<Future<Boolean>> futures = new ArrayList<>();
// 组间适配器并行运行
canalOuterAdapters.forEach(outerAdapters -> {
@@ -119,63 +116,64 @@ public abstract class AbstractCanalAdapterWorker {
futures.forEach(future -> {
try {
if (!future.get()) {
logger.error("Outer adapter write failed");
throw new RuntimeException("Outer adapter sync failed! ");
}
} catch (InterruptedException | ExecutionException e) {
// ignore
} catch (Exception e) {
future.cancel(true);
throw new RuntimeException(e);
}
});
});
}
@SuppressWarnings("unchecked")
protected void mqWriteOutData(int retry, long timeout, final boolean flatMessage, CanalMQConnector connector,
protected boolean mqWriteOutData(int retry, long timeout, int i, final boolean flatMessage, CanalMQConnector connector,
ExecutorService workerExecutor) {
for (int i = 0; i < retry; i++) {
try {
List<?> messages;
if (!flatMessage) {
messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS);
} else {
messages = connector.getFlatListWithoutAck(100L, TimeUnit.MILLISECONDS);
}
if (messages != null) {
Future<Boolean> future = workerExecutor.submit(() -> {
if (flatMessage) {
// batch write
writeOut((List<FlatMessage>) messages);
} else {
for (final Object message : messages) {
writeOut((Message) message);
}
try {
List<?> messages;
if (!flatMessage) {
messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS);
} else {
messages = connector.getFlatListWithoutAck(100L, TimeUnit.MILLISECONDS);
}
if (messages != null && !messages.isEmpty()) {
Future<Boolean> future = workerExecutor.submit(() -> {
if (flatMessage) {
// batch write
writeOut((List<FlatMessage>) messages);
} else {
for (final Object message : messages) {
writeOut((Message) message);
}
return true;
});
try {
future.get(timeout, TimeUnit.MILLISECONDS);
} catch (Exception e) {
future.cancel(true);
throw e;
}
return true;
});
try {
future.get(timeout, TimeUnit.MILLISECONDS);
} catch (Exception e) {
future.cancel(true);
throw e;
}
connector.ack();
break;
} catch (Throwable e) {
if (i == retry - 1) {
connector.ack();
} else {
connector.rollback();
}
logger.error(e.getMessage(), e);
try {
TimeUnit.SECONDS.sleep(1L);
} catch (InterruptedException e1) {
// ignore
}
}
return true;
} catch (Throwable e) {
if (i == retry - 1) {
connector.ack();
logger.error(e.getMessage() + " Error sync but ACK!");
return true;
} else {
connector.rollback();
logger.error(e.getMessage() + " Error sync and rollback, execute times: " + (i + 1));
}
try {
Thread.sleep(500);
} catch (InterruptedException e1) {
// ignore
}
}
return false;
}
/**

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

@@ -3,8 +3,11 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import org.apache.kafka.common.errors.WakeupException;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
@@ -61,7 +64,17 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
connector.disconnect();
break;
}
mqWriteOutData(retry, timeout, flatMessage, connector, workerExecutor);
if (retry == -1) {
retry = Integer.MAX_VALUE;
}
for (int i = 0; i < retry; i++) {
if (!running) {
break;
}
if (mqWriteOutData(retry, timeout, i, flatMessage, connector, workerExecutor)) {
break;
}
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);

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

@@ -3,7 +3,11 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import org.apache.kafka.common.errors.WakeupException;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
@@ -39,7 +43,8 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
;
ExecutorService workerExecutor = Executors.newSingleThreadExecutor();
int retry = canalClientConfig.getRetries() == null || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
int retry = canalClientConfig.getRetries() == null
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒
while (running) {
@@ -56,7 +61,17 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
connector.disconnect();
break;
}
mqWriteOutData(retry, timeout, flatMessage, connector, workerExecutor);
if (retry == -1) {
retry = Integer.MAX_VALUE;
}
for (int i = 0; i < retry; i++) {
if (!running) {
break;
}
if (mqWriteOutData(retry, timeout, i, flatMessage, connector, workerExecutor)) {
break;
}
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);

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

@@ -20,10 +20,10 @@ import com.alibaba.otter.canal.protocol.Message;
*/
public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
private static final int BATCH_SIZE = 50;
private static final int SO_TIMEOUT = 0;
private static final int BATCH_SIZE = 50;
private static final int SO_TIMEOUT = 0;
private CanalConnector connector;
private CanalConnector connector;
/**
* 单台client适配器worker的构造方法
@@ -61,10 +61,14 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
while (!running)
; // waiting until running == true
ExecutorService workerExecutor = Executors.newSingleThreadExecutor();
int retry = canalClientConfig.getRetries() == null
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
long timeout = canalClientConfig.getTimeout() == null ? 300000 : canalClientConfig.getTimeout(); // 默认超时5分钟
if (retry == -1) {
// 重试次数-1代表异常时一直阻塞重试
retry = Integer.MAX_VALUE;
}
// long timeout = canalClientConfig.getTimeout() == null ? 300000 :
// canalClientConfig.getTimeout(); // 默认超时5分钟
Integer batchSize = canalClientConfig.getBatchSize();
if (batchSize == null) {
batchSize = BATCH_SIZE;
@@ -90,6 +94,9 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
}
for (int i = 0; i < retry; i++) {
if (!running) {
break;
}
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
long batchId = message.getId();
try {
@@ -97,29 +104,19 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
if (batchId == -1 || size == 0) {
Thread.sleep(500);
} else {
Future<Boolean> future = workerExecutor.submit(() -> {
if (logger.isDebugEnabled()) {
logger.debug("destination: {} batchId: {} batchSize: {} ",
canalDestination,
batchId,
size);
}
long begin = System.currentTimeMillis();
writeOut(message);
if (logger.isDebugEnabled()) {
logger.debug("destination: {} batchId: {} elapsed time: {} ms",
canalDestination,
batchId,
System.currentTimeMillis() - begin);
}
return true;
});
try {
future.get(timeout, TimeUnit.MILLISECONDS);
} catch (Exception e) {
future.cancel(true);
throw e;
if (logger.isDebugEnabled()) {
logger.debug("destination: {} batchId: {} batchSize: {} ",
canalDestination,
batchId,
size);
}
long begin = System.currentTimeMillis();
writeOut(message);
if (logger.isDebugEnabled()) {
logger.debug("destination: {} batchId: {} elapsed time: {} ms",
canalDestination,
batchId,
System.currentTimeMillis() - begin);
}
}
connector.ack(batchId); // 提交确认
@@ -127,10 +124,11 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
} catch (Exception e) {
if (i != retry - 1) {
connector.rollback(batchId); // 处理失败, 回滚数据
logger.error(e.getMessage() + " Error sync and rollback, execute times: " + (i + 1));
} else {
connector.ack(batchId);
logger.error(e.getMessage() + " Error sync but ACK!");
}
logger.error("sync error!", e);
Thread.sleep(500);
}
}
@@ -151,8 +149,6 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
}
}
}
workerExecutor.shutdown();
}
@Override

9
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java
View File

@@ -5,7 +5,10 @@ import java.sql.SQLException;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import javax.sql.DataSource;
@@ -152,8 +155,8 @@ public class RdbAdapter implements OuterAdapter {
try {
future1.get();
future2.get();
} catch (ExecutionException | InterruptedException e) {
// ignore
} catch (Exception e) {
throw new RuntimeException(e);
}
}

101
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java
View File

@@ -47,63 +47,58 @@ public class RdbMirrorDbSyncService {
* @param dmls 批量 DML
*/
public void sync(List<Dml> dmls) {
try {
List<Dml> dmlList = new ArrayList<>();
for (Dml dml : dmls) {
String destination = StringUtils.trimToEmpty(dml.getDestination());
String database = dml.getDatabase();
MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(destination + "." + database);
List<Dml> dmlList = new ArrayList<>();
for (Dml dml : dmls) {
String destination = StringUtils.trimToEmpty(dml.getDestination());
String database = dml.getDatabase();
MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(destination + "." + database);
if (mirrorDbConfig == null) {
continue;
}
if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
// DDL
if (logger.isDebugEnabled()) {
logger.debug("DDL: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
}
executeDdl(mirrorDbConfig, dml);
rdbSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable());
mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置
} else {
// DML
initMappingConfig(dml.getTable(), mirrorDbConfig.getMappingConfig(), mirrorDbConfig, dml);
dmlList.add(dml);
}
}
if (!dmlList.isEmpty()) {
rdbSyncService.sync(dmlList, dml -> {
MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(dml.getDestination() + "." + dml.getDatabase());
if (mirrorDbConfig == null) {
continue;
return false;
}
if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
// DDL
if (logger.isDebugEnabled()) {
logger.debug("DDL: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
}
executeDdl(mirrorDbConfig, dml);
rdbSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable());
mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置
String table = dml.getTable();
MappingConfig config = mirrorDbConfig.getTableConfig().get(table);
if (config == null) {
return false;
}
if (config.getConcurrent()) {
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
singleDmls.forEach(singleDml -> {
int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData());
RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml);
rdbSyncService.getDmlsPartition()[hash].add(syncItem);
});
} else {
// DML
initMappingConfig(dml.getTable(), mirrorDbConfig.getMappingConfig(), mirrorDbConfig, dml);
dmlList.add(dml);
int hash = 0;
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
singleDmls.forEach(singleDml -> {
RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml);
rdbSyncService.getDmlsPartition()[hash].add(syncItem);
});
}
}
if (!dmlList.isEmpty()) {
rdbSyncService.sync(dmlList, dml -> {
MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache
.get(dml.getDestination() + "." + dml.getDatabase());
if (mirrorDbConfig == null) {
return false;
}
String table = dml.getTable();
MappingConfig config = mirrorDbConfig.getTableConfig().get(table);
if (config == null) {
return false;
}
if (config.getConcurrent()) {
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
singleDmls.forEach(singleDml -> {
int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData());
RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml);
rdbSyncService.getDmlsPartition()[hash].add(syncItem);
});
} else {
int hash = 0;
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
singleDmls.forEach(singleDml -> {
RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml);
rdbSyncService.getDmlsPartition()[hash].add(syncItem);
});
}
return true;
});
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
return true;
});
}
}

77
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
View File

@@ -133,49 +133,44 @@ public class RdbSyncService {
* @param dmls 批量 DML
*/
public void sync(Map<String, Map<String, MappingConfig>> mappingConfig, List<Dml> dmls) {
try {
sync(dmls, dml -> {
if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
// DDL
columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
sync(dmls, dml -> {
if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) {
// DDL
columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
return false;
} else {
// DML
String destination = StringUtils.trimToEmpty(dml.getDestination());
String database = dml.getDatabase();
String table = dml.getTable();
Map<String, MappingConfig> configMap = mappingConfig.get(destination + "." + database + "." + table);
if (configMap == null) {
return false;
} else {
// DML
String destination = StringUtils.trimToEmpty(dml.getDestination());
String database = dml.getDatabase();
String table = dml.getTable();
Map<String, MappingConfig> configMap = mappingConfig
.get(destination + "." + database + "." + table);
if (configMap == null) {
return false;
}
boolean executed = false;
for (MappingConfig config : configMap.values()) {
if (config.getConcurrent()) {
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
singleDmls.forEach(singleDml -> {
int hash = pkHash(config.getDbMapping(), singleDml.getData());
SyncItem syncItem = new SyncItem(config, singleDml);
dmlsPartition[hash].add(syncItem);
});
} else {
int hash = 0;
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
singleDmls.forEach(singleDml -> {
SyncItem syncItem = new SyncItem(config, singleDml);
dmlsPartition[hash].add(syncItem);
});
}
executed = true;
}
return executed;
}
});
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
boolean executed = false;
for (MappingConfig config : configMap.values()) {
if (config.getConcurrent()) {
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
singleDmls.forEach(singleDml -> {
int hash = pkHash(config.getDbMapping(), singleDml.getData());
SyncItem syncItem = new SyncItem(config, singleDml);
dmlsPartition[hash].add(syncItem);
});
} else {
int hash = 0;
List<SingleDml> singleDmls = SingleDml.dml2SingleDmls(dml);
singleDmls.forEach(singleDml -> {
SyncItem syncItem = new SyncItem(config, singleDml);
dmlsPartition[hash].add(syncItem);
});
}
executed = true;
}
return executed;
}
});
}
/**

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

@@ -1,9 +1,6 @@
package com.alibaba.otter.canal.client.kafka;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.*;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -22,7 +19,7 @@ import com.google.common.collect.Lists;
/**
* canal kafka 数据操作客户端
*
*
* <pre>
* 注意点：
* 1. 相比于canal {@linkplain SimpleCanalConnector}, 这里get和ack操作不能有并发, 必须是一个线程执行get后，内存里执行完毕ack后再取下一个get
@@ -34,14 +31,16 @@ import com.google.common.collect.Lists;
public class KafkaCanalConnector implements CanalMQConnector {
private KafkaConsumer<String, Message> kafkaConsumer;
private KafkaConsumer<String, String> kafkaConsumer2; // 用于扁平message的数据消费
private KafkaConsumer<String, String> kafkaConsumer2; // 用于扁平message的数据消费
private String topic;
private Integer partition;
private Properties properties;
private volatile boolean connected = false;
private volatile boolean running = false;
private volatile boolean connected = false;
private volatile boolean running = false;
private boolean flatMessage;
private Map<Integer, Long> currentOffsets = new HashMap<>();
public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
boolean flatMessage){
this.topic = topic;
@@ -79,6 +78,7 @@ public class KafkaCanalConnector implements CanalMQConnector {
connected = true;
if (kafkaConsumer == null && !flatMessage) {
kafkaConsumer = new KafkaConsumer<String, Message>(properties);
}
if (kafkaConsumer2 == null && flatMessage) {
kafkaConsumer2 = new KafkaConsumer<String, String>(properties);
@@ -176,6 +176,11 @@ public class KafkaCanalConnector implements CanalMQConnector {
ConsumerRecords<String, Message> records = kafkaConsumer.poll(unit.toMillis(timeout));
currentOffsets.clear();
for (TopicPartition topicPartition : records.partitions()) {
currentOffsets.put(topicPartition.partition(), kafkaConsumer.position(topicPartition));
}
if (!records.isEmpty()) {
List<Message> messages = new ArrayList<>();
for (ConsumerRecord<String, Message> record : records) {
@@ -208,6 +213,12 @@ public class KafkaCanalConnector implements CanalMQConnector {
}
ConsumerRecords<String, String> records = kafkaConsumer2.poll(unit.toMillis(timeout));
currentOffsets.clear();
for (TopicPartition topicPartition : records.partitions()) {
currentOffsets.put(topicPartition.partition(), kafkaConsumer2.position(topicPartition));
}
if (!records.isEmpty()) {
List<FlatMessage> flatMessages = new ArrayList<>();
for (ConsumerRecord<String, String> record : records) {
@@ -222,7 +233,22 @@ public class KafkaCanalConnector implements CanalMQConnector {
}
@Override
public void rollback() throws CanalClientException {
public void rollback() {
waitClientRunning();
if (!running) {
return;
}
// 回滚所有分区
if (kafkaConsumer != null) {
for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
}
}
if (kafkaConsumer2 != null) {
for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
kafkaConsumer2.seek(new TopicPartition(topic, entry.getKey()), entry.getValue() - 1);
}
}
}
/**