?? 本文提供的指令完全可以按順序逐一執(zhí)行，已進(jìn)行了多次測(cè)試。因此如果你是直接按照我本文寫的指令一條條執(zhí)行的，而非自定義修改過(guò)，執(zhí)行應(yīng)當(dāng)是沒(méi)有任何問(wèn)題的。

?? 本文講述：使用docker環(huán)境安裝mysql、canal、elasticsearch，基于binlog利用canal實(shí)現(xiàn)mysql的數(shù)據(jù)同步到elasticsearch中。

?? ???? 注意事項(xiàng)（如果你在運(yùn)行命令時(shí)遇到問(wèn)題一定看下這里）：

• 容器啟動(dòng)順序：mysql、es、canal、adapter

• 按照本文的命令啟動(dòng)容器正常情況下是沒(méi)有問(wèn)題的，如果遇到啟動(dòng)不了的情況，可以通過(guò) sudo docker logs 容器名 查看日志信息，我遇到的一個(gè)啟動(dòng)不了的情況是日志報(bào)錯(cuò)為內(nèi)存不足，這與你自身的服務(wù)器有關(guān)，因此在實(shí)操時(shí)我使用了兩臺(tái)虛擬機(jī)進(jìn)行操作。如果你的服務(wù)器或本地虛擬機(jī)內(nèi)存大小在2.5G以上應(yīng)該使用一臺(tái)就夠了。

• 很多容器中都只有 vi 命令，而沒(méi)有 vim 命令，關(guān)于 vim 命令的下載我試過(guò)很多次都沒(méi)有成功，并且有些容器的 vi 命令版本比較老導(dǎo)致使用起來(lái)一件很折磨人的事情，和vim有所不同。因此在本文中的容器內(nèi)的配置文件的修改，我的建議是復(fù)制一份配置文件到你的windows電腦進(jìn)行修改，然后刪除原配置文件新建一份同名文件，把你修改后的文件內(nèi)容拷貝上去即可。

• 你在自定義編寫 es_demo_collect.yml 的 es_demo_collect.yml 文件的 SQL 語(yǔ)句時(shí)需要注意（如果直接按照我寫的來(lái)是沒(méi)有問(wèn)題的）：

  1. 不要加頓號(hào)，例如：

     select b.`user_id`
     ...
     on b.`user_id`=u.`id`
     

     上述sql中的頓號(hào)必須全部刪除，即不能出現(xiàn)頓號(hào)。應(yīng)該改為如下：

     select b.user_id
     ...
     on b.user_id=u.id
     

  2. 外連接時(shí)必須將連接條件字段放入查詢中，例如 on b.user_id = u.id ，這里以 b.user_id/u.id 作為了連接條件，，則在 select 中必須查詢 b.user_id 或 u.id 。

• 你需要讓這些防火墻端口處于開(kāi)放狀態(tài)：3306，9200，9300，11111，8081

1.docker安裝

1.1 基于ubuntu

# 1.更新軟件包索引
sudo apt update

# 2.安裝必要的依賴軟件，來(lái)添加一個(gè)新的 HTTPS 軟件源，遇到 y/N 輸入y
sudo apt install apt-transport-https ca-certificates curl gnupg-agent software-properties-common

# 3.使用下面的 curl 導(dǎo)入源倉(cāng)庫(kù)的 GPG key
sudo curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add -

# 4.將 Docker APT 軟件源添加到你的系統(tǒng)
sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable"

# 5.更新軟件包索引
sudo apt update

# 6.安裝 Docker 最新版本，遇到 Y/n 輸入y
sudo apt install docker-ce docker-ce-cli containerd.io

# 7.安裝完成，查看docker版本
sudo docker --version

# 8.設(shè)置docker自啟動(dòng)
sudo systemctl enable docker

1.2 基于centos7

# 1.安裝 docker
yum install docker

# 2.檢驗(yàn)安裝是否成功
docker --version

# 3.啟動(dòng)
systemctl start docker

# 4.設(shè)置為自啟動(dòng)
systemctl enable docker

# 5.換鏡像源
sudo vim /etc/docker/daemon.json
#內(nèi)容如下：
{
 "registry-mirrors": ["https://m9r2r2uj.mirror.aliyuncs.com"]
}


# 6.重啟
sudo service docker restart

# 7.列出鏡像
docker images

# 8.查看運(yùn)行進(jìn)程
docker ps

2.數(shù)據(jù)卷統(tǒng)一管理

# 1.根目錄下創(chuàng)建waveedu文件夾
sudo mkdir /wave
# 2.移動(dòng)到wave文件夾
cd /wave

3.安裝mysql

# 1.拉取鏡像
sudo docker pull mysql:5.7.36

# 2.創(chuàng)建容器，密碼設(shè)置為123456，啟動(dòng)mysql
sudo docker run --name mysql5736 \
-v /wave/mysql/log:/var/log/mysql \
-v /wave/mysql/data:/var/lib/mysql \
-v /wave/mysql/conf:/etc/mysql/conf.d \
-e MYSQL_ROOT_PASSWORD=123456 \
-p 3306:3306 \
-d mysql:5.7.36

# 3.查看容器狀態(tài)，可以看到一個(gè)容器名稱為mysql5736的容器在運(yùn)行
sudo docker ps

# 4./wave/mysql/conf 目錄(數(shù)據(jù)卷目錄)中新建 my.cnf 文件，并在其中寫入如下內(nèi)容
sudo vim /wave/mysql/conf/my.cnf

# 表示客戶端的默認(rèn)字符集為utf8。
[client]
default_character_set=utf8

# 表示MySQL命令行客戶端的默認(rèn)字符集為utf8。
[mysql]
default_character_set=utf8

# 表示MySQL服務(wù)器的默認(rèn)字符集為utf8，并且 character_set_server 設(shè)置MySQL服務(wù)器默認(rèn)使用utf8字符集。
[mysqld]
character_set_server=utf8

[mysqld]
#binlog setting
# 開(kāi)啟logbin
log-bin=mysql-bin
# binlog日志格式
binlog-format=ROW
# mysql主從備份serverId,canal中不能與此相同
server-id=1

# 5.重啟容器
sudo docker restart mysql5736

# 6.進(jìn)入容器終端
sudo docker exec -it mysql5736 /bin/bash

# 7.登錄mysql
mysql -uroot -p123456

# 8.查看字符編碼是否為utf8
show variables like 'character%';

# 顯示信息如下說(shuō)明配置ok
+--------------------------+----------------------------+
| Variable_name            | Value                      |
+--------------------------+----------------------------+
| character_set_client     | utf8                       |
| character_set_connection | utf8                       |
| character_set_database   | utf8                       |
| character_set_filesystem | binary                     |
| character_set_results    | utf8                       |
| character_set_server     | utf8                       |
| character_set_system     | utf8                       |
| character_sets_dir       | /usr/share/mysql/charsets/ |
+--------------------------+----------------------------+
8 rows in set (0.00 sec)

# 9.展示binlog日志文件
show master status; 

# 10.重啟日志
reset master;

# 11.查看binlog文件格式
show variables like 'binlog_format';

如果顯示信息和上面一致，說(shuō)明MySQL安裝與配置成功，退出容器終端。

# 12.退出mysql
exit

# 13.退出容器終端
exit

4.安裝elasticsearch

# 1.拉取鏡像
sudo docker pull elasticsearch:7.16.2
# 2.創(chuàng)建容器并啟動(dòng)
sudo docker run --name='es7162' \
-p 9200:9200 \
-p 9300:9300 \
-e "discovery.type=single-node" \
-d elasticsearch:7.16.2

# 3.在運(yùn)行的es7162容器中打開(kāi)一個(gè)新的終端會(huì)話
sudo docker exec -ites es7162 /bin/bash

# 4.刪除原配置文件
rm -rf config/elasticsearch.yml

# 5.新建配置文件
vi config/elasticsearch.yml

# 配置文件內(nèi)容為如下
cluster.name: fox-es
network.host: 0.0.0.0

node.name: node-1
http.port: 9200
http.cors.enabled: true
http.cors.allow-origin: "*"
node.master: true
node.data: true

# 6.退出容器會(huì)話
exit

# 7.重啟es
sudo docker restart es7162

瀏覽器測(cè)試：http://【服務(wù)器地址】:9200

注意防火墻需要開(kāi)放端口

5.es安裝ik中文分詞器

5.1 在線安裝

# 1.在運(yùn)行的es7362容器中打開(kāi)一個(gè)新的終端會(huì)話
sudo docker exec -ites es7162 /bin/bash

# 2.在線下載（服務(wù)器網(wǎng)速不好下載會(huì)比較緩慢或下載失敗，則建議使用離線安裝的方式）
./bin/elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.16.2/elasticsearch-analysis-ik-7.16.2.zip

# 3.退出容器會(huì)話
exit

# 4.重啟es
sudo docker restart es7162

# 5.查看日志是否加載ik分詞器成功
sudo docker logs es7162

5.2 離線安裝

首先打開(kāi)這個(gè)鏈接：https://github.com/medcl/elasticsearch-analysis-ik/releases/tag/v7.16.2，把分詞器插件下載下來(lái)

# 1.把下載的 zip 上傳至linux服務(wù)器的 wave 文件夾下

# 2.把插件復(fù)制到容器內(nèi)
sudo docker cp elasticsearch-analysis-ik-7.16.2.zip es7162:/usr/share/elasticsearch/plugins

# 3.進(jìn)入容器
sudo docker exec -it es7162 /bin/bash

# 4.切換目錄
cd /usr/share/elasticsearch/plugins/

# 5.創(chuàng)建ik目錄
mkdir ik

# 6.解壓zip文件
unzip elasticsearch-analysis-ik-7.16.2.zip -d ik

# 7.移除zip文件
rm -rf elasticsearch-analysis-ik-7.16.2.zip

# 8.退出容器會(huì)話
exit

# 9.重啟es
sudo docker restart es7162

使用 Apipost7 測(cè)試路徑：【POST方式】 http://【linux服務(wù)器IP】:9200/_analyze

// 傳參如下
{
    "analyzer": "ik_smart",
    "text": "java名字由來(lái)"
}

出現(xiàn)如上信息說(shuō)明ik分詞器配置成功。

5.安裝canal-server

# 1.拉取鏡像
sudo docker pull canal/canal-server:v1.1.5

# 2.創(chuàng)建容器并啟動(dòng)
# --name canal115 為容器命名為canal115
# -p 11111:11111 表示將容器的11111端口映射到主機(jī)的11111端口
# --link mysql5736:mysql5736 表示將mysql5736容器鏈接到canal115容器，并且在canal115容器中可以使用mysql5736作為主機(jī)名訪問(wèn)mysql5736容器
# -id canal/canal-server:v1.1.5 表示使用鏡像canal/canal-server:v1.1.5創(chuàng)建并啟動(dòng)一個(gè)新的后臺(tái)運(yùn)行的容器。
sudo docker run --name canal115 \
-p 11111:11111 \
--link mysql5736:mysql5736 \
-id canal/canal-server:v1.1.5

# 3.在運(yùn)行的canal115容器中打開(kāi)一個(gè)新的終端會(huì)話
sudo docker exec -it canal115 /bin/bash

# 4.修改配置文件
vi canal-server/conf/example/instance.properties

# 把0改成10，只要不和mysql的id相同就行
canal.instance.mysql.slaveId=10
# 修改成mysql對(duì)應(yīng)的賬號(hào)密碼，mysql5736就是mysql鏡像的鏈接別名
canal.instance.master.address=mysql5736:3306
canal.instance.dbUsername=root
canal.instance.dbPassword=123456

# 5.退出canal容器會(huì)話
exit

# 6.重啟canal
sudo docker restart canal115

# 7.在運(yùn)行的canal115容器中打開(kāi)一個(gè)新的終端會(huì)話
sudo docker exec -it canal115 /bin/bash

# 8.進(jìn)入到日志 example 目錄下 
cd canal-server/logs/example/

# 9.查看日志，實(shí)時(shí)查看指定日志文件的最后100行，并將新添加到日志文件中的內(nèi)容實(shí)時(shí)輸出到屏幕。
tail -100f example.log

截圖如下，說(shuō)明已經(jīng)鏈接上了mysql主機(jī)，此時(shí)mysql中的數(shù)據(jù)變化，都會(huì)在canal中有同步。

# 10.退出日志打印
ctrl + c

# 11.退出容器
exit

6.實(shí)戰(zhàn)開(kāi)發(fā)-數(shù)據(jù)庫(kù)設(shè)計(jì)

6.1 開(kāi)發(fā)背景

假設(shè)有一個(gè)博客模塊，即有一個(gè)博客圈，用戶可以通過(guò)搜索來(lái)獲取與自己輸入的關(guān)鍵字相關(guān)的數(shù)據(jù)列表，類似于CSDN的搜索。

由于只是 demo，因此這里會(huì)設(shè)計(jì)的很簡(jiǎn)單。

6.2 數(shù)據(jù)庫(kù)表設(shè)計(jì)

使用客戶端連接到容器mysql5736，執(zhí)行下面語(yǔ)句進(jìn)行建表操作。

必須在安裝 canal-adapter 容器前先進(jìn)行建表操作，否則對(duì) canal-adapter 修改配置后，canal-adapter會(huì)重啟失敗。

# 如果存在 es_demo 數(shù)據(jù)庫(kù)則刪除
DROP DATABASE IF EXISTS `es_demo`;

# 創(chuàng)建新數(shù)據(jù)庫(kù)
CREATE DATABASE `es_demo`;

# 使用數(shù)據(jù)庫(kù)
USE `es_demo`;

# 創(chuàng)建一張用戶表，注意自增主鍵id是從1000開(kāi)始
CREATE TABLE `user`(
	`id` INT NOT NULL AUTO_INCREMENT COMMENT '主鍵id',
	`username` VARCHAR(24) NOT NULL COMMENT '用戶名',
	`icon` VARCHAR(255) NOT NULL COMMENT '頭像url',
	PRIMARY KEY(`id`)
)ENGINE=INNODB AUTO_INCREMENT=1000 DEFAULT CHARSET=utf8mb4 COMMENT='用戶表';

# 創(chuàng)建一張博客表，注意自增主鍵id是從1000開(kāi)始
CREATE TABLE `blog`(
	`id` INT NOT NULL AUTO_INCREMENT COMMENT '主鍵id',
	`user_id` INT NOT NULL COMMENT '用戶id（雪花算法生成）',
	`title` VARCHAR(255) NOT NULL COMMENT '標(biāo)題',
	`tags` VARCHAR(64) NOT NULL COMMENT '標(biāo)簽',
	`introduce` VARCHAR(512) NOT NULL COMMENT '介紹',
	`content` TEXT NOT NULL COMMENT '文章內(nèi)容', 
	`create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '創(chuàng)建時(shí)間',
	`update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改時(shí)間',
	PRIMARY KEY(`id`),
	KEY `idx_user_create`(`user_id`,`create_time` DESC)
)ENGINE=INNODB AUTO_INCREMENT=1000 DEFAULT CHARSET=utf8mb4 COMMENT='博客信息表';

7.安裝canal-adapter

# 1.鏡像拉取
sudo docker pull slpcat/canal-adapter:v1.1.5

# 2.你可以選擇依舊在同一臺(tái)虛擬機(jī)上創(chuàng)建容器并啟動(dòng)，那么啟動(dòng)方式如下

sudo docker run --name adapter115 \
-p 8081:8081 \
--link mysql5736:mysql5736 \
--link canal115:canal115 \
--link es7162:es7162 \
-d slpcat/canal-adapter:v1.1.5

# 但我啟動(dòng)時(shí)由于服務(wù)器內(nèi)存不足導(dǎo)致啟動(dòng)失敗，因此我選擇在另一臺(tái)虛擬機(jī)上執(zhí)行下面的命令
# 2.我在這里選擇放到了另一臺(tái)虛擬機(jī)上運(yùn)行，那么就不再鏈接，后面修改配置文件時(shí)直接寫前三個(gè)容器的所在虛擬機(jī)IP
sudo docker run --name adapter115 \
-p 8081:8081 \
-d slpcat/canal-adapter:v1.1.5

# 3.進(jìn)入容器
sudo docker exec -it adapter115 /bin/bash

# 4.進(jìn)入 conf 目錄
cd conf/

# 5.刪除原配置文件
rm -rf application.yml

# 6.新建配置文件
vi application.yml

?? 與 mysql、es、canal-server 依舊是在一臺(tái)服務(wù)器時(shí)的application.yml信息配置：

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 tcp consumer
    canal.tcp.server.host: canal115:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  srcDataSources:
    defaultDS:
      # 注意mysql容器中需要存在es_demo數(shù)據(jù)庫(kù)，否則重啟會(huì)報(bào)錯(cuò)
      url: jdbc:mysql://mysql5736:3306/es_demo?useUnicode=true
      username: root
      password: 123456
  canalAdapters:
  - instance: example # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger
      - name: es7
        hosts: es7162:9200 # 127.0.0.1:9200 for rest mode
        properties:
          mode: rest
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: fox-es

?? 與 mysql、es、canal-server 依舊不在同一臺(tái)服務(wù)器時(shí)的application.yml信息配置：

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 tcp consumer，這個(gè)192.168.65.133就是我前三個(gè)容器所在虛擬機(jī)的IP
    canal.tcp.server.host: 192.168.65.133:11111
    canal.tcp.zookeeper.hosts:
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  srcDataSources:
    defaultDS:
      # 這個(gè)192.168.65.133就是我前三個(gè)容器所在虛擬機(jī)的IP，注意mysql容器中需要存在es_demo數(shù)據(jù)庫(kù)，否則重啟會(huì)報(bào)錯(cuò)
      url: jdbc:mysql://192.168.65.133:3306/es_demo?useUnicode=true
      username: root
      password: 123456
  canalAdapters:
  - instance: example # canal instance Name or mq topic name
    groups:
    - groupId: g1
      outerAdapters:
      - name: logger
      - name: es7
        # 這個(gè)192.168.65.133就是我前三個(gè)容器所在虛擬機(jī)的IP
        hosts: 192.168.65.133:9200 # 127.0.0.1:9200 for rest mode
        properties:
          mode: rest
          # security.auth: test:123456 #  only used for rest mode
          cluster.name: fox-es

# 7.進(jìn)入到 es7 目錄下，注意此時(shí)依舊在容器中
cd /opt/canal-adapter/conf/es7

# 8.刪除該目錄下所有配置文件
rm -rf biz_order.yml customer.yml mytest_user.yml

# 9.新建配置文件
vi es_demo_collect.yml

dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
  _index: es_demo_collect
  _id: _id
  _type: _doc
  upsert: true
#  pk: id
# 這里的select語(yǔ)句會(huì)與數(shù)據(jù)庫(kù)表設(shè)計(jì)的SQL相關(guān)聯(lián)，我會(huì)創(chuàng)建兩張表user和blog，創(chuàng)建的SQL我會(huì)在后面的操作中寫到
  sql: "
SELECT 
	b.id AS _id,
	b.user_id AS userId,
	b.title AS title,
	b.tags AS tags,
	b.introduce AS introduce,
	b.content AS content,
	b.create_time AS createTime,
	b.update_time AS updateTime,
	u.icon AS userIcon,
	u.username AS username
FROM 
	blog b
LEFT JOIN user u
ON b.user_id = u.id

"
#  objFields:
#    _labels: array:;
#   etlCondition: "where c.c_time>={}"
  commitBatch: 3000

注意對(duì)于時(shí)間類型，在后端一定要使用LocalDateTime或者LocalDate類型，如果是Date類型，需要自己手動(dòng)設(shè)置格式。

# 10.退出容器
exit

# 11.重啟容器
sudo docker restart adapter115

# 12.查看日志信息
sudo docker logs --tail 100 adapter115

8.實(shí)戰(zhàn)開(kāi)發(fā)-創(chuàng)建es索引與映射

使用Apipost建立索引與映射：【PUT】 http://192.168.65.133:9200/es_demo_collect

{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "userId": {
        "type": "long"
      },
      "username":{
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"  
      },
      "userIcon":{
        "type": "keyword",
        "index": false
      },
      "title": {
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"
      },
      "tags": {
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"
      },
      "introduce":{
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"
      },
      "content":{
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "type": "text"
      },
      "createTime":{
        "format": "date_optional_time||epoch_millis",
        "type": "date"
      },
      "updateTime":{
        "format": "date_optional_time||epoch_millis",
        "type": "date"
      }
    }
  }
}

刪除索引：DELETE http://192.168.65.133:9200/es_demo_collect

9.效果測(cè)試

9.1 新增記錄測(cè)試

在 canal-adapter 容器所在虛擬器上執(zhí)行：

# 實(shí)時(shí)檢測(cè) adapter 信息
sudo docker logs --tail 100  -f adapter115

對(duì)數(shù)據(jù)庫(kù)進(jìn)行如下SQL操作：

# 在建表時(shí)，我們兩張表的自增主鍵id都是從1000開(kāi)始的，因此在插入時(shí) user_id 的測(cè)試數(shù)據(jù)也是從1000開(kāi)始
INSERT INTO `user` VALUES(NULL,'狐貍半面添','https://sangxin-fox');
INSERT INTO `user` VALUES(NULL,'逐浪者','https://zhulang-fox');
INSERT INTO `blog`(`user_id`,`title`,`tags`,`introduce`,`content`) VALUES 
(1001,'Java語(yǔ)言','編程,java,語(yǔ)言','Java的起源','Java最初是由任職于太陽(yáng)微系統(tǒng)的詹姆斯-高斯林（James-Gosling）等人于1990年代初開(kāi)發(fā)。最初被命名為Oak；當(dāng)時(shí)發(fā)現(xiàn)Oak被其他公司注冊(cè)了，不得不重新起名，當(dāng)時(shí)他們正在咖啡館喝著印尼爪哇（JAVA）島出產(chǎn)的咖啡，有人提議就叫JAVA怎么樣，然后就這樣定下來(lái)了。'),
(1000,'C語(yǔ)言','編程,C,語(yǔ)言','C的起源','C語(yǔ)言是由美國(guó)貝爾實(shí)驗(yàn)室的Dennis Ritchie于1972年設(shè)計(jì)發(fā)明的，最初在UNIX操作系統(tǒng)的DEC PDP-11計(jì)算機(jī)上使用。它由早期的編程語(yǔ)言BCPL（Basic Combind Programming Language）發(fā)展演變而來(lái)。');

查看 adapter 日志信息：

查看 es 信息，使用 apipost 發(fā)送請(qǐng)求：【GET】http://192.168.65.133:9200/es_demo_collect/_search

// 以下是該請(qǐng)求需要攜帶的json數(shù)據(jù)，表示查詢es_demo_collect索引中的全部文檔數(shù)據(jù)
{
    "query": {
        "match_all": {}
    }
}

9.2 修改記錄信息測(cè)試

同時(shí)，我們可以再測(cè)試修改，我們將用戶頭像路徑進(jìn)行修改，看看 es 是否同步了新的數(shù)據(jù)：

UPDATE `user` SET icon='https:///langlang' WHERE id=1001

查看 es 信息，使用 apipost 發(fā)送請(qǐng)求：【GET】http://192.168.65.133:9200/es_demo_collect/_search

// 以下是該請(qǐng)求需要攜帶的json數(shù)據(jù)，表示查詢es_demo_collect索引中的全部文檔數(shù)據(jù)
{
    "query": {
        "match_all": {}
    }
}

10.實(shí)戰(zhàn)開(kāi)發(fā)-后端代碼

以下只展示我認(rèn)為比較與本文相關(guān)的比較重要的文件，完整源碼的獲取鏈接我會(huì)放在文章的最后。

10.1 pom.xml依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.fox</groupId>
    <artifactId>elasticsearch-canal-demo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.6.3</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.8.0</version>
        </dependency>
        <!--fastjson依賴-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.33</version>
        </dependency>
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.7</version>
        </dependency>
    </dependencies>
</project>

10.2 application.yml配置

server:
  # 服務(wù)端口
  port: 9999
elasticsearch:
  # es訪問(wèn)ip
  hostname: 192.168.65.133
  # es訪問(wèn)port
  port: 9200
  blog:
    # 訪問(wèn)索引
    index: es_demo_collect
    # 搜索返回字段
    source_fields: userId,title,username,userIcon,introduce,createTime,updateTime

10.3 ElasticsearchConfig.java配置類

package com.fox.es.config;

import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author 狐貍半面添
 * @create 2023-03-22 17:51
 */
@Configuration
public class ElasticsearchConfig {
    @Value("${elasticsearch.hostname}")
    private String hostname;
    @Value("${elasticsearch.port}")
    private Integer port;

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        RestClientBuilder builder = RestClient.builder(
                new HttpHost(hostname, port, "http")
        );
        return new RestHighLevelClient(builder);
    }
}

10.4 ?測(cè)試是否連接 es 成功

package com.fox.es.controller;

import com.fox.es.entity.Result;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.MainResponse;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.IOException;

/**
 * @author 狐貍半面添
 * @create 2023-03-22 18:33
 */
@RestController
public class TestController {
    @Resource
    private RestHighLevelClient restHighLevelClient;

    /**
     * 用于測(cè)試是否連接 es 成功
     *
     * @return 返回 es 的基本信息，等價(jià)于訪問(wèn)：http://127.0.0.1:9200
     * @throws IOException 異常信息
     */
    @GetMapping("/getEsInfo")
    public Result getEsInfo() throws IOException {
        MainResponse info = restHighLevelClient.info(RequestOptions.DEFAULT);
        return Result.ok(info);
    }
}

瀏覽器訪問(wèn)：http://localhost:9999/getEsInfo

10.5 ?搜索服務(wù)

10.5.1 controller層

package com.fox.es_canal.controller;

import com.fox.es_canal.constant.BlogConstants;
import com.fox.es_canal.entity.Result;
import com.fox.es_canal.service.BlogService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @author 狐貍半面添
 * @create 2023-03-22 20:16
 */
@RestController
@RequestMapping("/blog")
public class BlogController {

    @Resource
    private BlogService blogService;

    /**
     * 通過(guò)關(guān)鍵詞獲取數(shù)據(jù)列表
     *
     * @param keyWords 關(guān)鍵詞
     * @param pageNo   頁(yè)碼
     * @return 數(shù)據(jù)列表，按照相關(guān)性從高到低進(jìn)行排序
     */
    @GetMapping("/list")
    public Result list(@RequestParam("keyWords") String keyWords,
                       @RequestParam("pageNo") Integer pageNo) {
        // BlogConstants是我寫的一個(gè)常量類，里面定義了一個(gè)變量 SEARCH_PAGE_NUM = 15
        return blogService.list(keyWords, pageNo, BlogConstants.SEARCH_PAGE_NUM);
    }
}

10.5.2 service接口層

package com.fox.es_canal.service;

import com.fox.es_canal.entity.Result;

/**
 * @author 狐貍半面添
 * @create 2023-03-22 20:18
 */
public interface BlogService {

    /**
     * 通過(guò)關(guān)鍵詞獲取數(shù)據(jù)列表
     *
     * @param keyWords 關(guān)鍵詞
     * @param pageNo 頁(yè)碼
     * @param pageSize 每頁(yè)大小
     * @return 數(shù)據(jù)列表，按照相關(guān)性從高到低進(jìn)行排序
     */
    Result list(String keyWords, int pageNo, int pageSize);
}

10.5.3 service實(shí)現(xiàn)層

package com.fox.es_canal.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fox.es_canal.dto.BlogSimpleInfoDTO;
import com.fox.es_canal.entity.Result;
import com.fox.es_canal.service.BlogService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MultiMatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author 狐貍半面添
 * @create 2023-03-22 20:18
 */
@Slf4j
@Service
public class BlogServiceImpl implements BlogService {
    @Resource
    private RestHighLevelClient restHighLevelClient;
    @Value("${elasticsearch.blog.index}")
    private String blogIndexStore;
    @Value("${elasticsearch.blog.source_fields}")
    private String blogFields;


    public Result list(String keyWords, int pageNo, int pageSize) {
        // 1.設(shè)置索引 - blog
        SearchRequest searchRequest = new SearchRequest(blogIndexStore);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        // 2.source源字段過(guò)慮
        String[] sourceFieldsArray = blogFields.split(",");
        searchSourceBuilder.fetchSource(sourceFieldsArray, new String[]{});

        // 3.關(guān)鍵字
        if (StringUtils.hasText(keyWords)) {
            // 哪些字段匹配關(guān)鍵字
            MultiMatchQueryBuilder multiMatchQueryBuilder = QueryBuilders.multiMatchQuery(keyWords, "title", "tags", "username", "introduce", "content");
            // 設(shè)置匹配占比（表示最少匹配的子句個(gè)數(shù),例如有五個(gè)可選子句,最少的匹配個(gè)數(shù)為5*70%=3.5.向下取整為3,這就表示五個(gè)子句最少要匹配其中三個(gè)才能查到）
            multiMatchQueryBuilder.minimumShouldMatch("70%");
            // 提升字段的Boost值
            multiMatchQueryBuilder.field("title", 15);
            multiMatchQueryBuilder.field("tags", 10);
            multiMatchQueryBuilder.field("introduce", 7);
            multiMatchQueryBuilder.field("content", 3);
            multiMatchQueryBuilder.field("username", 3);

            boolQueryBuilder.must(multiMatchQueryBuilder);
        }

        // 4.分頁(yè)
        int start = (pageNo - 1) * pageSize;
        searchSourceBuilder.from(start);
        searchSourceBuilder.size(pageSize);

        // 布爾查詢
        searchSourceBuilder.query(boolQueryBuilder);

        // 6.高亮設(shè)置
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.preTags("<font color='red'>");
        highlightBuilder.postTags("</font>");
        // 設(shè)置高亮字段
        ArrayList<HighlightBuilder.Field> fields = new ArrayList<>();
        fields.add(new HighlightBuilder.Field("title"));
        fields.add(new HighlightBuilder.Field("introduce"));
        fields.add(new HighlightBuilder.Field("username"));
        highlightBuilder.fields().addAll(fields);
        searchSourceBuilder.highlighter(highlightBuilder);

        // 請(qǐng)求搜索
        searchRequest.source(searchSourceBuilder);

        SearchResponse searchResponse;
        try {
            searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("博客搜索異常：{}", e.getMessage());
            return Result.error(e.getMessage());
        }

        // 結(jié)果集處理
        SearchHits hits = searchResponse.getHits();
        SearchHit[] searchHits = hits.getHits();
        // 記錄總數(shù)
        long totalHitsCount = hits.getTotalHits().value;
        // 數(shù)據(jù)列表
        List<BlogSimpleInfoDTO> list = new ArrayList<>();

        for (SearchHit hit : searchHits) {

            JSONObject jsonObject = JSONObject.parseObject(hit.getSourceAsString());
            BlogSimpleInfoDTO blog = new BlogSimpleInfoDTO();
            blog.setId(Integer.parseInt(hit.getId()));
            blog.setUsername(jsonObject.getString("username"));
            blog.setTitle(jsonObject.getString("title"));
            blog.setUserId(Long.parseLong(jsonObject.getString("userId")));
            blog.setUserIcon(jsonObject.getString("userIcon"));
            blog.setIntroduce(jsonObject.getString("introduce"));
            blog.setCreateTime(LocalDateTime.parse(jsonObject.getString("createTime"), DateTimeFormatter.ISO_OFFSET_DATE_TIME));
            blog.setUpdateTime(LocalDateTime.parse(jsonObject.getString("updateTime"), DateTimeFormatter.ISO_OFFSET_DATE_TIME));

            // 取出高亮字段內(nèi)容
            Map<String, HighlightField> highlightFields = hit.getHighlightFields();
            if (highlightFields != null) {
                blog.setTitle(parseHighlightStr(blog.getTitle(), highlightFields.get("title")));
                blog.setIntroduce(parseHighlightStr(blog.getIntroduce(), highlightFields.get("introduce")));
                blog.setUsername(parseHighlightStr(blog.getUsername(), highlightFields.get("username")));
            }

            list.add(blog);
        }

        // 封裝信息返回前端
        HashMap<String, Object> resultMap = new HashMap<>(4);
        // 頁(yè)碼
        resultMap.put("pageNo", pageNo);
        // 每頁(yè)記錄數(shù)量
        resultMap.put("pageSize", pageSize);
        // 總記錄數(shù)
        resultMap.put("total", totalHitsCount);
        // 該頁(yè)信息
        resultMap.put("items", list);

        return Result.ok(resultMap);

    }

    public String parseHighlightStr(String text, HighlightField field) {
        if (field != null) {
            Text[] fragments = field.getFragments();
            StringBuilder stringBuilder = new StringBuilder();
            for (Text str : fragments) {
                stringBuilder.append(str.string());
            }
            return stringBuilder.toString();
        } else {
            return text;
        }
    }
}

10.5.4 效果測(cè)試

這里我們使用 apipost7 或?yàn)g覽器 進(jìn)行測(cè)試：

11.源碼獲取

Java源碼地址：Mr-Write/SpringbootDemo: 各種demo案例 (github.com)

對(duì)應(yīng)的是 elasticsearch-canal-demo 包模塊。

12.其它說(shuō)明

當(dāng)我們?cè)贘ava中寫出往MySQL數(shù)據(jù)庫(kù)添加、刪除、修改博客記錄的操作接口時(shí)，會(huì)同時(shí)通過(guò) Canal 同步到es中，因?yàn)?canal 同步的本質(zhì)還是去讀 MySQL的 binlog 日志。由于比較簡(jiǎn)單，在這里就不做演示了。文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-431448.html

到了這里，關(guān)于docker環(huán)境安裝mysql、canal、elasticsearch，基于binlog利用canal實(shí)現(xiàn)mysql的數(shù)據(jù)同步到elasticsearch中的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！