Canal+Springboot实现数据同步,全网唯一一个跑通了的案例
以下是一个简化的示例,展示如何在Spring Boot应用中使用Canal实现数据同步:
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
@Service
public class CanalService {
@PostConstruct
public void startCanal() {
// 连接到Canal Server
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress(AddressUtils.getHostIp(),
11111), "example", "", "");
int batchSize = 1000;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
while (true) {
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
long batchId = message.getId();
if (batchId == -1 || message.getEntries().isEmpty()) {
// 没有数据,休眠一会儿
Thread.sleep(1000);
} else {
dataHandle(message.getEntries());
connector.ack(batchId); // 确认消息消费成功
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
connector.disconnect();
}
}
private void dataHandle(List<CanalEntry.Entry> entrys) {
for (CanalEntry.Entry entry : entrys) {
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
continue;
}
CanalEntry.RowChange rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
switch (rowChage.getEventType()) {
case INSERT:
handleInsert(rowChage);
break;
case UPDATE:
handleUpdate(rowChage);
break;
case DELETE:
handleDelete(rowChage);
评论已关闭