canal同步elasticsearch定製開發，首先研究官方文件：https://github.com/alibaba/canal/wiki/ClientAPI，同時官方也提供了各種test開發例項，研究完官方文件和開發例項，canal的定製開發將不再高深莫測。

everything is easy

基於Kafka、RocketMQ的資料同步

Kafka、RocketMQ的配置和說明請參閱官方文件：https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart。

使用Kafka、RocketMQ時，一般都是採用多topic單分割槽模式，雖然存在熱點表寫入分割槽的效能問題，但是保證了binlog處理的順序性，這種方式在順序性和效能方面做了較好的平衡。

我們只需要根據表名接入相應的topic拉取資料進行處理後提交至ES即可，為了減輕elasticsearch伺服器壓力，我們可以採用一個小技巧，就是elasticsearch 批次提交，但需要在系統穩定和業務容忍度之間做下取捨，主要考量以下三個方面：

資料實時性要求日常業務處理頻率批次提交時間間隔

因此對於實時性要求極高的資料就直接寫入elasticsearch吧。

基於disruptor的資料同步

由於基礎設施建設匱乏，整體運維能力偏弱，我們沒法搭建Kafka、RocketMQ叢集，為了實現資料同步兼顧高併發要求，我們可以採用disruptor高效能佇列，同時使用串並行操作來最佳化資料同步。

Disruptor是英國外匯交易公司LMAX開發的一個高效能佇列，研發的初衷是解決記憶體佇列的延遲問題（在效能測試中發現竟然與I/O操作處於同樣的數量級），能夠在無鎖的情況下實現網路的Queue併發操作，基於Disruptor開發的系統單執行緒能支撐每秒600萬訂單。目前，包括Apache Storm、Camel、Log4j2等等知名的框架都在內部集成了Disruptor用來替代jdk的佇列，以此來獲得高效能。

責任鏈模式在面向物件程式設計裡是一種軟體設計模式，它包含了一些命令物件和一系列的處理物件。每一個處理物件決定它能處理哪些命令物件，它也知道如何將它不能處理的命令物件傳遞給該鏈中的下一個處理物件。該模式還描述了往該處理鏈的末尾新增新的處理物件的方法。

Chain of Responsibility Pattern