基于Docker結(jié)合Canal實(shí)現(xiàn)MySQL實(shí)時(shí)增量數(shù)據(jù)傳輸功能

docker network create --subnet=172.18.0.0/16 mynetwork

##生成mysql容器
docker run -d --name mysql --net mynetwork --ip 172.18.0.6 -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root mysql
##生成canal-server容器
docker run -d --name canal-server --net mynetwork --ip 172.18.0.4 -p 11111:11111 canal/canal-server
## 命令介紹
--net mynetwork #使用自定義網(wǎng)絡(luò)
--ip #指定分配ip

[mysqld]
log-bin=mysql-bin # 開啟binlog
binlog-format=ROW # 選擇ROW模式
server_id=1 # 配置MySQL replaction需要定義，不要和Canal的slaveId重復(fù)

mysql -uroot -proot
# 創(chuàng)建賬號(hào)
CREATE USER canal IDENTIFIED BY 'canal'; 
# 授予權(quán)限
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
# 刷新并應(yīng)用
FLUSH PRIVILEGES;

show variables like 'log_bin';
show variables like 'log_bin';
show master status;

docker exec -it canal-server bash
tail -100f canal-server/logs/example/example.log

# 下載對(duì)鏡像
docker pull elasticsearch:7.1.1
docker pull mobz/elasticsearch-head:5-alpine
# 創(chuàng)建容器并運(yùn)行
docker run -d --name elasticsearch --net mynetwork --ip 172.18.0.2 -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" elasticsearch:7.1.1
docker run -d --name elasticsearch-head --net mynetwork --ip 172.18.0.5 -p 9100:9100 mobz/elasticsearch-head:5-alpine

package com.example.canal.study.pojo;
import lombok.Data;
import java.io.Serializable;
// @Data 用戶生產(chǎn)getter、setter方法
@Data
public class Student implements Serializable {
private String id;
private String name;
private int age;
private String sex;
private String city;
} 

package com.example.canal.study.common;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.net.InetSocketAddress;
/**
* @author haha
*/
@Configuration
public class CanalConfig {
// @Value 獲取 application.properties配置中端內(nèi)容
@Value("${canal.server.ip}")
private String canalIp;
@Value("${canal.server.port}")
private Integer canalPort;
@Value("${canal.destination}")
private String destination;
@Value("${elasticSearch.server.ip}")
private String elasticSearchIp;
@Value("${elasticSearch.server.port}")
private Integer elasticSearchPort;
@Value("${zookeeper.server.ip}")
private String zkServerIp;
// 獲取簡(jiǎn)單canal-server連接
@Bean
public CanalConnector canalSimpleConnector() {
 CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalIp, canalPort), destination, "", "");
 return canalConnector;
}
// 通過(guò)連接zookeeper獲取canal-server連接
@Bean
public CanalConnector canalHaConnector() {
 CanalConnector canalConnector = CanalConnectors.newClusterConnector(zkServerIp, destination, "", "");
 return canalConnector;
}
// elasticsearch 7.x客戶端
@Bean
public RestHighLevelClient restHighLevelClient() {
 RestHighLevelClient client = new RestHighLevelClient(
   RestClient.builder(new HttpHost(elasticSearchIp, elasticSearchPort))
 );
 return client;
}
} 

public static class TwoTuple<A, B> {
 public final A eventType;
 public final B columnMap;
 public TwoTuple(A a, B b) {
  eventType = a;
  columnMap = b;
 }
}
public static List<TwoTuple<EventType, Map>> printEntry(List<Entry> entrys) {
 List<TwoTuple<EventType, Map>> rows = new ArrayList<>();
 for (Entry entry : entrys) {
  // binlog event的事件事件
  long executeTime = entry.getHeader().getExecuteTime();
  // 當(dāng)前應(yīng)用獲取到該binlog鎖延遲的時(shí)間
  long delayTime = System.currentTimeMillis() - executeTime;
  Date date = new Date(entry.getHeader().getExecuteTime());
  SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  // 當(dāng)前的entry（binary log event）的條目類型屬于事務(wù)
  if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
   if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN) {
    TransactionBegin begin = null;
    try {
     begin = TransactionBegin.parseFrom(entry.getStoreValue());
    } catch (InvalidProtocolBufferException e) {
     throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
    }
    // 打印事務(wù)頭信息，執(zhí)行的線程id，事務(wù)耗時(shí)
    logger.info(transaction_format,
      new Object[]{entry.getHeader().getLogfileName(),
        String.valueOf(entry.getHeader().getLogfileOffset()),
        String.valueOf(entry.getHeader().getExecuteTime()),
        simpleDateFormat.format(date),
        entry.getHeader().getGtid(),
        String.valueOf(delayTime)});
    logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
    printXAInfo(begin.getPropsList());
   } else if (entry.getEntryType() == EntryType.TRANSACTIONEND) {
    TransactionEnd end = null;
    try {
     end = TransactionEnd.parseFrom(entry.getStoreValue());
    } catch (InvalidProtocolBufferException e) {
     throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
    }
    // 打印事務(wù)提交信息，事務(wù)id
    logger.info("----------------\n");
    logger.info(" END ----> transaction id: {}", end.getTransactionId());
    printXAInfo(end.getPropsList());
    logger.info(transaction_format,
      new Object[]{entry.getHeader().getLogfileName(),
        String.valueOf(entry.getHeader().getLogfileOffset()),
        String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
        entry.getHeader().getGtid(), String.valueOf(delayTime)});
   }
   continue;
  }
  // 當(dāng)前entry（binary log event）的條目類型屬于原始數(shù)據(jù)
  if (entry.getEntryType() == EntryType.ROWDATA) {
   RowChange rowChage = null;
   try {
    // 獲取儲(chǔ)存的內(nèi)容
    rowChage = RowChange.parseFrom(entry.getStoreValue());
   } catch (Exception e) {
    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
   }
   // 獲取當(dāng)前內(nèi)容的事件類型
   EventType eventType = rowChage.getEventType();
   logger.info(row_format,
     new Object[]{entry.getHeader().getLogfileName(),
       String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
       entry.getHeader().getTableName(), eventType,
       String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
       entry.getHeader().getGtid(), String.valueOf(delayTime)});
   // 事件類型是query或數(shù)據(jù)定義語(yǔ)言DDL直接打印sql語(yǔ)句，跳出繼續(xù)下一次循環(huán)
   if (eventType == EventType.QUERY || rowChage.getIsDdl()) {
    logger.info(" sql ----> " + rowChage.getSql() + SEP);
    continue;
   }
   printXAInfo(rowChage.getPropsList());
   // 循環(huán)當(dāng)前內(nèi)容條目的具體數(shù)據(jù)
   for (RowData rowData : rowChage.getRowDatasList()) {
    List<CanalEntry.Column> columns;
    // 事件類型是delete返回刪除前的列內(nèi)容，否則返回改變后列的內(nèi)容
    if (eventType == CanalEntry.EventType.DELETE) {
     columns = rowData.getBeforeColumnsList();
    } else {
     columns = rowData.getAfterColumnsList();
    }
    HashMap<String, Object> map = new HashMap<>(16);
    // 循環(huán)把列的name與value放入map中
    for (Column column: columns){
     map.put(column.getName(), column.getValue());
    }
    rows.add(new TwoTuple<>(eventType, map));
   }
  }
 }
 return rows;
} 

package com.example.canal.study.common;
import com.alibaba.fastjson.JSON;
import com.example.canal.study.pojo.Student;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.xcontent.XContentType;
import java.io.IOException;
import java.util.Map;
/**
* @author haha
*/
@Slf4j
@Component
public class ElasticUtils {
@Autowired
private RestHighLevelClient restHighLevelClient;
/**
 * 新增
 * @param student 
 * @param index 索引
 */
public void saveEs(Student student, String index) {
 IndexRequest indexRequest = new IndexRequest(index)
   .id(student.getId())
   .source(JSON.toJSONString(student), XContentType.JSON)
   .opType(DocWriteRequest.OpType.CREATE);
 try {
  IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
  log.info("保存數(shù)據(jù)至ElasticSearch成功：{}", response.getId());
 } catch (IOException e) {
  log.error("保存數(shù)據(jù)至elasticSearch失敗: {}", e);
 }
}
/**
 * 查看
 * @param index 索引
 * @param id _id
 * @throws IOException
 */
public void getEs(String index, String id) throws IOException {
 GetRequest getRequest = new GetRequest(index, id);
 GetResponse response = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
 Map<String, Object> fields = response.getSource();
 for (Map.Entry<String, Object> entry : fields.entrySet()) {
  System.out.println(entry.getKey() + ":" + entry.getValue());
 }
}
/**
 * 更新
 * @param student
 * @param index 索引
 * @throws IOException
 */
public void updateEs(Student student, String index) throws IOException {
 UpdateRequest updateRequest = new UpdateRequest(index, student.getId());
 updateRequest.upsert(JSON.toJSONString(student), XContentType.JSON);
 UpdateResponse response = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
 log.info("更新數(shù)據(jù)至ElasticSearch成功：{}", response.getId());
}
/**
 * 根據(jù)id刪除數(shù)據(jù)
 * @param index 索引
 * @param id _id
 * @throws IOException
 */
public void DeleteEs(String index, String id) throws IOException {
 DeleteRequest deleteRequest = new DeleteRequest(index, id);
 DeleteResponse response = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
 log.info("刪除數(shù)據(jù)至ElasticSearch成功：{}", response.getId());
}
} 

package com.example.canal.study.action;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.example.canal.study.common.CanalDataParser;
import com.example.canal.study.common.ElasticUtils;
import com.example.canal.study.pojo.Student;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.List;
import java.util.Map;
/**
* @author haha
*/
@Slf4j
@Component
public class BinLogElasticSearch {
@Autowired
private CanalConnector canalSimpleConnector;
@Autowired
private ElasticUtils elasticUtils;
//@Qualifier("canalHaConnector")使用名為canalHaConnector的bean
@Autowired
@Qualifier("canalHaConnector")
private CanalConnector canalHaConnector;
public void binLogToElasticSearch() throws IOException {
 openCanalConnector(canalHaConnector);
 // 輪詢拉取數(shù)據(jù)
 Integer batchSize = 5 * 1024;
 while (true) {
  Message message = canalHaConnector.getWithoutAck(batchSize);
//   Message message = canalSimpleConnector.getWithoutAck(batchSize);
  long id = message.getId();
  int size = message.getEntries().size();
  log.info("當(dāng)前監(jiān)控到binLog消息數(shù)量{}", size);
  if (id == -1 || size == 0) {
   try {
    // 等待2秒
    Thread.sleep(2000);
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  } else {
   //1. 解析message對(duì)象
   List<CanalEntry.Entry> entries = message.getEntries();
   List<CanalDataParser.TwoTuple<CanalEntry.EventType, Map>> rows = CanalDataParser.printEntry(entries);
   for (CanalDataParser.TwoTuple<CanalEntry.EventType, Map> tuple : rows) {
    if(tuple.eventType == CanalEntry.EventType.INSERT) {
     Student student = createStudent(tuple);
     // 2。將解析出的對(duì)象同步到elasticSearch中
     elasticUtils.saveEs(student, "student_index");
     // 3.消息確認(rèn)已處理
//     canalSimpleConnector.ack(id);
     canalHaConnector.ack(id);
    }
    if(tuple.eventType == CanalEntry.EventType.UPDATE){
     Student student = createStudent(tuple);
     elasticUtils.updateEs(student, "student_index");
     // 3.消息確認(rèn)已處理
//     canalSimpleConnector.ack(id);
     canalHaConnector.ack(id);
    }
    if(tuple.eventType == CanalEntry.EventType.DELETE){
     elasticUtils.DeleteEs("student_index", tuple.columnMap.get("id").toString());
     canalHaConnector.ack(id);
    }
   }
  }
 }
}
/**
 * 封裝數(shù)據(jù)至Student
 * @param tuple
 * @return
 */
private Student createStudent(CanalDataParser.TwoTuple<CanalEntry.EventType, Map> tuple){
 Student student = new Student();
 student.setId(tuple.columnMap.get("id").toString());
 student.setAge(Integer.parseInt(tuple.columnMap.get("age").toString()));
 student.setName(tuple.columnMap.get("name").toString());
 student.setSex(tuple.columnMap.get("sex").toString());
 student.setCity(tuple.columnMap.get("city").toString());
 return student;
}
/**
 * 打開canal連接
 *
 * @param canalConnector
 */
private void openCanalConnector(CanalConnector canalConnector) {
 //連接CanalServer
 canalConnector.connect();
 // 訂閱destination
 canalConnector.subscribe();
}
/**
 * 關(guān)閉canal連接
 *
 * @param canalConnector
 */
private void closeCanalConnector(CanalConnector canalConnector) {
 //關(guān)閉連接CanalServer
 canalConnector.disconnect();
 // 注銷訂閱destination
 canalConnector.unsubscribe();
}
} 

package com.example.canal.study;
import com.example.canal.study.action.BinLogElasticSearch;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author haha
*/
@SpringBootApplication
public class CanalDemoApplication implements ApplicationRunner {
@Autowired
private BinLogElasticSearch binLogElasticSearch;
public static void main(String[] args) {
 SpringApplication.run(CanalDemoApplication.class, args);
}
// 程序啟動(dòng)則執(zhí)行run方法
@Override
public void run(ApplicationArguments args) throws Exception {
 binLogElasticSearch.binLogToElasticSearch();
}
} 

server.port=8081
spring.application.name = canal-demo
canal.server.ip = 192.168.124.5
canal.server.port = 11111
canal.destination = example
zookeeper.server.ip = 192.168.124.5:2181
zookeeper.sasl.client = false
elasticSearch.server.ip = 192.168.124.5
elasticSearch.server.port = 9200

docker pull zookeeper
docker run -d --name zookeeper --net mynetwork --ip 172.18.0.3 -p 2181:2181 zookeeper
docker run -d --name canal-server2 --net mynetwork --ip 172.18.0.8 -p 11113:11113 canal/canal-server

canal.port=11113
canal.zkServers=172.18.0.3:2181
canal.instance.global.spring.xml = classpath:spring/default-instance.xml

canal.instance.mysql.slaveId = 1235 
#之前的canal slaveId是1234，保證slaveId不重復(fù)即可
canal.instance.master.address = 172.18.0.6:3306

[zk: localhost:2181(CONNECTED) 15] get /otter/canal/destinations/example/running 
{"active":true,"address":"172.18.0.4:11111","cid":1} 

[zk: localhost:2181(CONNECTED) 0] get /otter/canal/destinations/example/running
{"active":true,"address":"172.18.0.4:11111","cid":1} 

CanalConnector connector = CanalConnectors.newClusterConnector("172.18.0.3:2181", "example", "", "");

[zk: localhost:2181(CONNECTED) 4] get /otter/canal/destinations/example/1001/running
{"active":true,"address":"192.168.124.5:59887","clientId":1001} 

[zk: localhost:2181(CONNECTED) 5] get /otter/canal/destinations/example/1001/cursor

{"@type":"com.alibaba.otter.canal.protocol.position.LogPosition","identity":{"slaveId":-1,"sourceAddress":{"address":"mysql.mynetwork","port":3306}},"postion":{"included":false,"journalName":"binlog.000004","position":2169,"timestamp":1562672817000}} 

docker exec -it canal-server bash
cd canal-server/bin
sh stop.sh

[zk: localhost:2181(CONNECTED) 19] get /otter/canal/destinations/example/running
{"active":true,"address":"172.18.0.8:11111","cid":1} 

[root@localhost /usr/local/elasticsearch-head-master]# cd ../elasticsearch-5.5.2/config/
[root@localhost /usr/local/elasticsearch-5.5.2/config]# vim elasticsearch.yml 
# 文件末尾加上如下配置
http.cors.enabled: true
http.cors.allow-origin: "*"

#6886行 contentType: "application/x-www-form-urlencoded
改成 contentType: "application/json;charset=UTF-8"
 #7574行 var inspectData = s.contentType === "application/x-www-form-urlencoded" &&
改成 var inspectData = s.contentType === "application/json;charset=UTF-8" &&

#pom中除了加入依賴
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>7.1.1</version>
</dependency>
#還需加入
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>7.1.1</version>
</dependency>

如果創(chuàng)建時(shí)未指定 --restart=always ,可通過(guò)update 命令
docker update --restart=always [containerID]