|
|
@@ -0,0 +1,168 @@
|
|
|
+package com.ruoyi.web.job;
|
|
|
+
|
|
|
+import cn.hutool.core.util.StrUtil;
|
|
|
+import com.alibaba.fastjson.JSON;
|
|
|
+import com.alibaba.fastjson.serializer.SerializerFeature;
|
|
|
+import com.alibaba.otter.canal.client.CanalConnector;
|
|
|
+import com.alibaba.otter.canal.client.CanalConnectors;
|
|
|
+import com.alibaba.otter.canal.protocol.CanalEntry;
|
|
|
+import com.alibaba.otter.canal.protocol.Message;
|
|
|
+import com.google.protobuf.InvalidProtocolBufferException;
|
|
|
+import com.ruoyi.framework.config.ElasticSearchClient;
|
|
|
+import com.ruoyi.web.core.config.CanalConfig;
|
|
|
+import lombok.Data;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.elasticsearch.action.index.IndexRequest;
|
|
|
+import org.elasticsearch.client.RequestOptions;
|
|
|
+import org.elasticsearch.client.RestHighLevelClient;
|
|
|
+import org.springframework.beans.BeansException;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.beans.factory.annotation.Value;
|
|
|
+import org.springframework.context.ApplicationContext;
|
|
|
+import org.springframework.context.ApplicationContextAware;
|
|
|
+import org.springframework.scheduling.annotation.Scheduled;
|
|
|
+import org.springframework.stereotype.Component;
|
|
|
+
|
|
|
+import javax.annotation.Resource;
|
|
|
+import java.io.IOException;
|
|
|
+import java.net.InetSocketAddress;
|
|
|
+import java.util.HashMap;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+
|
|
|
+/**
|
|
|
+ * @Description: TODO
|
|
|
+ * @Author: huangcheng
|
|
|
+ * @Date: 2021/8/16
|
|
|
+ * @Version V1.0
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+@Component
|
|
|
+public class CanalScheduling implements Runnable, ApplicationContextAware {
|
|
|
+
|
|
|
+ private ApplicationContext applicationContext;
|
|
|
+ @Autowired
|
|
|
+ private ElasticSearchClient client;
|
|
|
+// @Resource
|
|
|
+// private CanalConnector canalConnector;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private CanalConfig canalConfig;
|
|
|
+
|
|
|
+ @Override
|
|
|
+ @Scheduled(cron = "0/1 * * * * ?", fixedDelay = 100) //每隔100秒执行
|
|
|
+ public void run() {
|
|
|
+ for (CanalConfig.Config config : canalConfig.getConfigs()) {
|
|
|
+ // 创建链接
|
|
|
+ CanalConnector canalConnector = CanalConnectors
|
|
|
+ .newSingleConnector(new InetSocketAddress(config.getHostname(), config.getPort()), config.getDestination(), "", "");
|
|
|
+ long batchId = -1;
|
|
|
+ try {
|
|
|
+ canalConnector.connect();
|
|
|
+ canalConnector.subscribe();
|
|
|
+ canalConnector.rollback();
|
|
|
+
|
|
|
+ //每次拉取条数
|
|
|
+ int batchSize = 1000;
|
|
|
+ Message message = canalConnector.getWithoutAck(batchSize);
|
|
|
+ //批次id
|
|
|
+ batchId = message.getId();
|
|
|
+ List<CanalEntry.Entry> entries = message.getEntries();
|
|
|
+ if (batchId != -1 && entries.size() > 0) {
|
|
|
+ entries.forEach(entry -> {
|
|
|
+ //MySQL种my.cnf中配置的是binlog_format = ROW，这里只解析ROW类型
|
|
|
+ if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
|
|
|
+ //解析处理
|
|
|
+ publishCanalEvent(entry);
|
|
|
+ }
|
|
|
+ });
|
|
|
+ }
|
|
|
+ canalConnector.ack(batchId);
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.info("canal存在异常：{}", e.getMessage());
|
|
|
+ e.printStackTrace();
|
|
|
+ canalConnector.rollback(batchId);
|
|
|
+ } finally {
|
|
|
+ // 断开连接
|
|
|
+ canalConnector.disconnect();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private void publishCanalEvent(CanalEntry.Entry entry) {
|
|
|
+ //表名
|
|
|
+ String tableName = entry.getHeader().getTableName();
|
|
|
+ //数据库名
|
|
|
+ String database = entry.getHeader().getSchemaName();
|
|
|
+ // 操作类型
|
|
|
+ CanalEntry.EventType eventType = entry.getHeader().getEventType();
|
|
|
+
|
|
|
+ log.info(String.format("========> binlog[%s:%s] , name[%s,%s] , eventType[%s]",
|
|
|
+ entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
|
|
|
+ database, tableName,
|
|
|
+ eventType));
|
|
|
+
|
|
|
+ CanalEntry.RowChange rowChange;
|
|
|
+ try {
|
|
|
+ rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
|
|
|
+ } catch (InvalidProtocolBufferException e) {
|
|
|
+ e.printStackTrace();
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ rowChange.getRowDatasList().forEach(rowData -> {
|
|
|
+ //获取改变前的数据
|
|
|
+ List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
|
|
|
+ //获取改变后的数据
|
|
|
+ List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
|
|
|
+ Map<String, Object> beforeColumnsToMap = parseColumnsToMap(beforeColumnsList);
|
|
|
+ Map<String, Object> afterColumnsToMap = parseColumnsToMap(afterColumnsList);
|
|
|
+ try {
|
|
|
+ //插入es
|
|
|
+ indexES(beforeColumnsToMap, afterColumnsToMap, eventType, database, tableName);
|
|
|
+ } catch (IOException e) {
|
|
|
+ e.printStackTrace();
|
|
|
+ }
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ Map<String, Object> parseColumnsToMap(List<CanalEntry.Column> columns) {
|
|
|
+ Map<String, Object> map = new HashMap<>();
|
|
|
+ columns.forEach(column -> {
|
|
|
+ if (column == null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ map.put(column.getName(), column.getValue());
|
|
|
+ });
|
|
|
+ return map;
|
|
|
+ }
|
|
|
+
|
|
|
+ private void indexES(Map<String, Object> beforeDataMap, Map<String, Object> afterDataMap, CanalEntry.EventType eventType, String database, String table) throws IOException {
|
|
|
+ log.info("eventType:{},database:{},table:{}\nbeforeMap:{},\n afterMap:{}", eventType, database, table, beforeDataMap, afterDataMap);
|
|
|
+ if (!StrUtil.equals("test", database)) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ //不是user表中的数据不处理
|
|
|
+ if (!StrUtil.equals("user", table)) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 根据不同类型处理相应的逻辑
|
|
|
+ switch (eventType) {
|
|
|
+ case INSERT:
|
|
|
+ break;
|
|
|
+ case UPDATE:
|
|
|
+ break;
|
|
|
+ case DELETE:
|
|
|
+ break;
|
|
|
+ default:
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
|
|
|
+ this.applicationContext = applicationContext;
|
|
|
+ }
|
|
|
+}
|
459242451@qq.com