開發(fā)中，我們會發(fā)現(xiàn)關(guān)系型數(shù)據(jù)庫，對于數(shù)據(jù)分析來說，會有很大的不方便性，當(dāng)數(shù)據(jù)量過大時，查詢效率更加的慢。

首先我們能需要了解一下 主從數(shù)據(jù)庫，請移步到 docker mysql8.0主從庫

canal-server呢， 其實我認為canal就是類似從庫的概念，監(jiān)聽主庫的變化，

canal-adapter: 基于canal server訂閱Mysql binglog日志增量同步數(shù)據(jù)的一款工具

閑話不說了，開始吧！

安裝mysql數(shù)據(jù)庫

1. 新建mysql.cnf文件，存儲到 D:\docker\mysql\master\conf文件夾下

[mysqld]
server-id=100
log-bin=mysql-bin
binlog_cache_size=1M
# binlog日志格式
binlog_format=row

2. 啟動

docker run -itd -p 3307:3306 --name mysql-master -v D:\docker\mysql\master\conf:/etc/mysql/conf.d -v D:\docker\mysql\master\data:/var/lib/mysql -e MYSQL_ROOT_PASSWORD=123456 mysql:8.0

3. 修改mysql.cnf文件權(quán)限

docker exec -it mysql-master chmod 644 /etc/mysql/conf.d/mysql.cnf

4. 重啟

安裝 canal-server

1. 下載

docker pull canal/canal-server:v1.1.5

2. 啟動臨時容器

docker run --rm --name canal canal/canal-server:v1.1.5

3. 將容器中的instance.properties 復(fù)制到本地 D:\docker\canal\server文件夾下

docker cp canal:/home/admin/canal-server/conf/example/instance.properties .

4. 編輯instance.properties， 其他保持不變

## server_id 與mysql數(shù)據(jù)庫不要設(shè)置一樣
canal.instance.mysql.slaveId=300

# mysql數(shù)據(jù)庫ip與端口
canal.instance.master.address=192.168.1.8:3307

# 訪問主庫的用戶名密碼
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

5. 啟動canal-server

docker run -p 11111:11111 --name canal -v D:\docker\canel\server\conf\instance.properties:/home/admin/canal-server/conf/example/instance.properties  -v D:\docker\canel\server\logs:/home/admin/canal-server/logs -itd canal/canal-server:v1.1.5

6. 驗證啟動成功

docker logs -f canal
...
==> INIT DONE
==> RUN /home/admin/app.sh
==> START ...
start canal ...
start canal successful
==> START SUCCESSFUL ...

在mysql應(yīng)用中創(chuàng)建一個數(shù)據(jù)庫

CREATE DATABASE `canaltest` CHARACTER SET 'utf8mb4' COLLATE 'utf8mb4_general_ci';

驗證canal是否正常連接

PS D:\docker\canel> docker exec -it canal tail -1f canal-server/logs/example/example.log
2023-09-08 21:36:08.627 [MultiStageCoprocessor-other-example-0] 
WARN  c.a.o.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta - 
dup apply for sql : 
CREATE DATABASE `canaltest` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci'

安裝canal-adapter

1. 下載鏡像

docker pull slpcat/canal-adapter:v1.1.5

2. 臨時容器

docker run --rm --name adapter-tmp slpcat/canal-adapter:v1.1.5

3. 復(fù)制application.yml到本地 D:\docker\canel\adapter\conf

docker cp adapter-tmp:/opt/canal-adapter/conf/application.yml .

4. 修改文件內(nèi)容

server:
  port: 8081
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
    default-property-inclusion: non_null

canal.conf:
  mode: tcp #tcp kafka rocketMQ rabbitMQ
  flatMessage: true
  zookeeperHosts:
  syncBatchSize: 1000
  retries: 0
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    # canal-server的ip:端口
    canal.tcp.server.host: 192.168.1.8:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  srcDataSources:
    defaultDS:
      # 數(shù)據(jù)庫來源，用戶名、密碼 canaltest
      url: jdbc:mysql://192.168.1.8:3307/canaltest?useUnicode=true
      username: canal
      password: canal
  canalAdapters:
  - instance: example # canal-server 實例名稱，這個跟canal-server的instance.properties的文件夾名稱一致就好
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger
      - name: es7  #這個是容器內(nèi)conf目錄es7文件夾名稱
        hosts: 192.168.1.8:9210 # es對外可訪問地址
        properties:
          mode: rest
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: docker-cluster  # 集群的名稱，docker啟動單例es時，默認為docker-cluster

5. 啟動adapter

docker run -itd --name canal-adapter -p 8081:8081 -v D:\docker\canel\adapter\conf\application.yml:/opt/canal-adapter/conf/application.yml slpcat/canal-adapter:v1.1.5

6. 新增一個表映射索引文件

docker exec -it canal-adapter /bin/bash

cd /opt/canal-adapter/conf/es7/
touch my_test.yml

文件內(nèi)容如下

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  # es中索引
  _index: mytest_user
  _id: _id
  _type: _doc
  upsert: true
  sql: "
      SELECT
              id as _id,
              id,
              name,
              role_id,
              c_time
      FROM
              user
      "
  commitBatch: 3000

Mysql 創(chuàng)建表，追加數(shù)據(jù)

-- 創(chuàng)建數(shù)據(jù)庫, 因前面配置canal-adapter時，配置的數(shù)據(jù)庫為canaltest
CREATE DATABASE `canaltest` CHARACTER SET 'utf8mb4' COLLATE 'utf8mb4_general_ci';

-- 創(chuàng)建表
CREATE TABLE `user` (
  `id` int NOT NULL,
  `name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `role_id` int DEFAULT NULL,
  `c_time` datetime DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

-- 新增數(shù)據(jù)
INSERT INTO `canaltest`.`user` (`id`, `name`, `role_id`, `c_time`) VALUES (2, 'haha', 1, '2023-09-05 19:53:31');

ES中創(chuàng)建索引

PUT localhost:9210/mytest_user
{
    "mappings": {
        "properties": {
            "c_time": {
                "type": "date",
                "format": "date_optional_time||epoch_millis"
            },
            "name": {
                "type": "text"
            },
            "role_id": {
                "type": "long"
            },
            "id": {
                "type": "long"
            }
        }
    }
}

查詢ES中是否存在數(shù)據(jù)

GET localhost:9210/mytest_user/_search
{
    "took": 12,
    "timed_out": false,
    "_shards": {
        "total": 1,
        "successful": 1,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": {
            "value": 1,
            "relation": "eq"
        },
        "max_score": 1.0,
        "hits": [
            {
                "_index": "mytest_user",
                "_id": "1",
                "_score": 1.0,
                "_source": {
                    "id": 1,
                    "name": "haha",
                    "role_id": 1,
                    "c_time": "2023-09-05T19:53:31+08:00"
                }
            }
        ]
    }
}

至此， msyql中數(shù)據(jù)變化，canal都會將變化數(shù)據(jù)同步到es中