Canal同步Mysql数据到ES以及Springboot项目查询ES数据

1、环境准备

服务器：Centos7

Jdk版本：1.8

Mysql版本：5.7.44

Canal版本：1.17

Es版本：7.12.1

kibana版本：7.12.1

软件包下载地址：链接：https://pan.baidu.com/s/1jRpCJP0-hr9aIghC2ZbS4g 提取码：zzzz

IP地址	安装软件
192.168.50.210	Mysql，Canal
192.168.50.211	Es，Kibana

2、安装es以及kibana

2.1 安装docker

#设置源
wget https://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo -O /etc/yum.repos.d/docker-ce.repo
# 安装依赖
yum install -y yum-utils device-mapper-persistent-data lvm2
# 安装docker
yum install -y docker-ce
# 检查安装
docker -v
# 启动
systemctl start docker

设置容器镜像加速地址登录阿里云容器镜像服务进入到镜像工具 -> 镜像加速器

# https://xxxxxxx.mirror.aliyuncs.com 替换成你的地址 ！！！！！！！！！！！！
sudo mkdir -p /etc/docker
sudo tee /etc/docker/daemon.json <<-'EOF'
{"registry-mirrors": ["https://xxxxxxx.mirror.aliyuncs.com"]
}
EOF
sudo systemctl daemon-reload
sudo systemctl restart docker

设置开机启动

systemctl enable docker.service

2.2 安装es

# 创建容器网络 es-net
docker network create es-net
# docker 安装 es 
docker run -d \--name es \-e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \-e "discovery.type=single-node" \-v es-data:/usr/share/elasticsearch/data \-v es-plugins:/usr/share/elasticsearch/plugins \--privileged \--network es-net \-p 9200:9200 \-p 9300:9300 \
elasticsearch:7.12.1

# 开通端口
sudo iptables -A INPUT -p tcp --dport 9200 -j ACCEPT
sudo iptables -A INPUT -p tcp --dport 9300 -j ACCEPT

2.3 安装kibana

# docker 安装 kibana
docker run -d \
--name kibana \
-e ELASTICSEARCH_HOSTS=http://es:9200 \
-e "I18N_LOCALE=zh-CN" \
--network=es-net \
-p 5601:5601  \
kibana:7.12.1
# 开通端口
sudo iptables -A INPUT -p tcp --dport 5601 -j ACCEPT

3、安装mysql以及canel

3.1 安装Jdk

# 创建jdk安装路径
mkdir -p /opt/java
#将 jdk-8u301-linux-x64.tar.gz  放置 /opt/java
mv /youpath/jdk-8u301-linux-x64.tar.gz /opt/java
# 解压
cd /opt/java
tar -zxvf jdk-8u301-linux-x64.tar.gz
# 添加环境变量
vi /etc/profile
# 加入如下片段
JAVA_HOME=/opt/java/jdk1.8.0_301
JRE_HOME=/opt/java/jdk1.8.0_301/jre
PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
export JAVA_HOME JRE_HOME PATH CLASSPATH
# 保存
# 刷新环境变量
source /etc/profile
# 检查
java -version

3.2 安装Mysql

# 将安装包 mysql-5.7.44-linux-glibc2.12-x86_64.tar.gz 放入/opt下
cd /opt 
tar zxvf mysql-5.7.44-linux-glibc2.12-x86_64.tar.gz
mv mysql-5.7.44-linux-glibc2.12-x86_64 mysql
# 删除安装包
rm mysql-5.7.44-linux-glibc2.12-x86_64.tar.gz
# 添加环境变量
vi /etc/profile
# 加入如下代码段
export PATH=/opt/mysql/bin:$PATH
# 刷新环境变量
source /etc/profile
# 创建数据目录
mkdir -p /opt/mysql/data

# 创建用户 mysql
useradd -m mysql
# 将/opt/mysql 权限给到mysql用户
chown -R mysql:mysql /home/mysql/mysql-5.7.44

# 切换用户
su mysql

# 初始化mysql
mysqld --initialize  --user=mysql --basedir=/opt/mysql --datadir=/opt/mysql/data

如下：记录初始密码，下边要用到

# 编写配置文件
vi /etc/my.cnf
# 新增或者修改参数如下
[mysqld]
symbolic-links=0 # 禁用软连接
user=mysql # 用户
basedir=/opt/mysql
datadir=/opt/mysql/data
socket=/tmp/mysql.sock
lower_case_table_names=1
server-id=1
port=3306
log-bin=/opt/mysql/mysql-bin
binlog-format=ROW
expire-logs-days=15

#复制启动脚本
cp /opt/mysql/support-files/mysql.server /etc/init.d/mysqld
#启动mysql
/etc/init.d/mysqld start

# 修改数据库密码  
mysql -uroot -p 
# 这里输入的是上边初始的默认密码mysql>set password=password('root');# 创建用户mysql>use mysql;mysql>CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';mysql>GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';mysql>FLUSH PRIVILEGES;# 创建数据库mysql>create database canal_test character set utf8mb4 collate utf8mb4_bin;mysql>use canal_test;# 创建表mysql>CREATE TABLE open_user(id          bigint(21)   not null auto_increment,user_name   varchar(255) CHARACTER SET utf8mb4 DEFAULT NULL,sex         int(11)                            DEFAULT 1,portrait    varchar(255) NOT NULL,create_time datetime                           DEFAULT NULL,update_time datetime                           DEFAULT NULL,primary key (id));# 退出mysql>exit;

3.3 安装canal.deployer

# 创建 canal用户
useradd -m canal
# 设置密码
passwd canal
# 密码为 canal，输入两遍即可
su canal
cd ~
mkdir canal-deployer
# 将 canal.deployer-1.1.7.tar.gz 移至 canal-deployer
mv /youpath/canal.deployer-1.1.7.tar.gz /home/canal/canal-deployer
cd /home/canal/canal-deployer
tar -zxvf canal.deployer-1.1.7.tar.gz
cd conf/example
vi instance.properties
# 修改如下配置
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
# 保存
# 进入到启动命令目录下
cd /home/canal/canal-deployer/bin
sh ./startup.sh
# 查看日志
tail -f /home/canal/canal-deployer/logs/example/example.log

3.4 安装canal.adapter

# 接上
cd ~
mkdir canal-adapter
# 将 canal.adapter-1.1.7.tar.gz 移至 canal-adapter
mv /youpath/canal.adapter-1.1.7.tar.gz /home/canal/canal-adapter
cd canal-adapter
tar -zxvf canal.adapter-1.1.7.tar.gz
# 修改配置文件为如下

server:port: 8081
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8default-property-inclusion: non_nullcanal.conf:mode: tcp #tcp kafka rocketMQ rabbitMQflatMessage: truezookeeperHosts:syncBatchSize: 1000retries: -1timeout:accessKey:secretKey:consumerProperties:# canal tcp consumercanal.tcp.server.host: 127.0.0.1:11111canal.tcp.zookeeper.hosts:canal.tcp.batch.size: 500canal.tcp.username:canal.tcp.password:srcDataSources:defaultDS:url: jdbc:mysql://127.0.0.1:3306/canal_test?useUnicode=true&useSSL=falseusername: canalpassword: canalcanalAdapters:- instance: example # canal instance Name or mq topic namegroups:- groupId: g1outerAdapters:- name: logger- name: es7hosts: http://192.168.50.211:9200 # es地址properties:mode: rest # or rest# security.auth: test:123456 #  only used for rest modecluster.name: docker-cluster

处理日期格式化为 yyyy-MM-dd HH:mm:ss 需将 client-adapter.es7x-1.1.7-jar-with-dependencies.jar 替换掉 /home/canal/canal-adapter/plugin 中的 client-adapter.es7x-1.1.7-jar-with-dependencies.jar

# 设置数据以及es映射信息
# 进入到配置目录下的es7目录
cd /home/canal/canal-adapter/conf/es7
# 创建文件 open_user.yml 内容如下：

dataSourceKey: defaultDS  #此配置为application.yml 的key
destination: example      #此配置为canal的name
groupId: g1
esMapping:_index: open_user_type: _doc_id: _idsql: "SELECT u.id AS _id,u.user_name AS userName,u.sex,u.portrait,u.create_time as createTime,u.update_time as updateTime FROM open_user u"commitBatch: 3000

# 启动
# 进入启动目录
cd /home/canal/canal-adapter/bin
sh startup.sh
tail -f /home/canal/canal-adapter/logs/adapter/adapter.log

3.5 初始数据

3.5.1 创建索引

进入kibana控制页面

打开 kibana

地址：http://192.168.50.211:5601

进入开发工具菜单

3.5.2 初始化Mysql数据

在mysql中增加记录

INSERT INTO canal_test.open_user (id, user_name, sex, portrait, create_time, update_time) VALUES (1, '张三', 2, '学生', '2023-11-02 16:31:21', '2023-11-02 16:39:20');
INSERT INTO canal_test.open_user (id, user_name, sex, portrait, create_time, update_time) VALUES (2, '李四', 1, '美术组组长', '2023-11-03 08:57:32', '2023-11-03 08:57:34');
INSERT INTO canal_test.open_user (id, user_name, sex, portrait, create_time, update_time) VALUES (3, '王五', 1, '班长', '2023-11-03 09:13:35', '2023-11-03 09:13:37');
INSERT INTO canal_test.open_user (id, user_name, sex, portrait, create_time, update_time) VALUES (4, '赵六', 1, '劳动委员', '2023-11-03 09:44:45', '2023-11-03 09:44:46');

3.5.3 初始化ES数据

进入canal 安装服务器直接调用canal-adapter的Rest API：如下：

curl -X POST http://127.0.0.1:8081/etl/es7/open_user.yml

3.5.4 测试修改数据

1）查看 adapter.log 日志

 tail -f /home/canal/canal-adapter/logs/adapter/adapter.log

2）修改数据库 open_user 表中的数据

update open_user set user_name = '章三' where id = 1;

日志输出如下：

2023-11-03 16:12:02.477 [pool-3-thread-1] INFO  c.a.o.canal.client.adapter.logger.LoggerAdapterExample - DML: {"data":[{"id":1,"user_name":"章三","sex":2,"portrait":"学生","create_time":1698913881000,"update_time":1698914360000}],"database":"canal_test","destination":"example","es":1698999121000,"groupId":"g1","isDdl":false,"old":[{"user_name":"张三"}],"pkNames":["id"],"sql":"","table":"open_user","ts":1698999122129,"type":"UPDATE"}
2023-11-03 16:12:02.477 [pool-3-thread-1] DEBUG c.a.o.canal.client.adapter.es.core.support.ESSyncUtil - typeConvert valClass:class java.lang.String val:章三 esType:text
2023-11-03 16:12:02.483 [pool-3-thread-1] DEBUG c.a.o.canal.client.adapter.es.core.service.ESSyncService - DML: {"data":[{"id":1,"user_name":"章三","sex":2,"portrait":"学生","create_time":1698913881000,"update_time":1698914360000}],"database":"canal_test","destination":"example","es":1698999121000,"groupId":"g1","isDdl":false,"old":[{"user_name":"张三"}],"pkNames":["id"],"sql":"","table":"open_user","ts":1698999122129,"type":"UPDATE"} 
Affected indexes: open_user

4、Spring-boot集成

4.1 创建springboot工程引入如下依赖

    	<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId></dependency>

4.2 创建实体类

package com.example.demo.model;import com.alibaba.fastjson.annotation.JSONField;
import com.fasterxml.jackson.annotation.JsonFormat;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;import java.io.Serializable;
import java.util.Date;@Document(indexName = "open_user", type = "_doc")
public class OpenUser implements Serializable {@Idprivate String id;@Field(type = FieldType.Text)private String userName;@Field(type = FieldType.Text)private String sex;@Field(type = FieldType.Text)private String portrait;@Field(type = FieldType.Date)@JSONField(format = "yyyy-MM-dd HH:mm:ss")@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")private Date createTime;@Field(type = FieldType.Date)@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")@JSONField(format = "yyyy-MM-dd HH:mm:ss")private Date updateTime;public String getId() {return id;}public void setId(String id) {this.id = id;}public String getUserName() {return userName;}public void setUserName(String userName) {this.userName = userName;}public String getSex() {return sex;}public void setSex(String sex) {this.sex = sex;}public String getPortrait() {return portrait;}public void setPortrait(String portrait) {this.portrait = portrait;}public Date getCreateTime() {return createTime;}public void setCreateTime(Date createTime) {this.createTime = createTime;}public Date getUpdateTime() {return updateTime;}public void setUpdateTime(Date updateTime) {this.updateTime = updateTime;}@Overridepublic String toString() {return "OpenUser{" + "id='" + id + '\'' + ", userName='" + userName + '\'' + ", sex='" + sex + '\''+ ", portrait='" + portrait + '\'' + ", createTime=" + createTime + '\'' + ", updateTime=" + updateTime + '}';}
}

4.3 创建接口

package com.example.demo.controller;import com.example.demo.model.OpenUser;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;@RestController
public class SearchController {@Autowiredprivate ElasticsearchTemplate elasticsearchTemplate;@PostMapping("/findOpenUserByUserName")public Page<OpenUser> findOpenUserByUserName(@RequestParam(value = "userName") String userName,@RequestParam(value = "pageNum", required = false) Integer pageNum,@RequestParam(value = "pageSize", required = false) Integer pageSize) {if (StringUtils.isBlank(userName)) {return null;}if (pageNum == null || pageNum < 0) {pageNum = 0; // if page is null, page = 0 size default 1}if (pageSize == null || pageSize < 0) {pageSize = 10; // if size is null, size default 10}// 分页，根据时间倒序Pageable pageable = PageRequest.of(pageNum, pageSize, Sort.Direction.DESC, "createTime");// 查询姓名QueryBuilder builder = null;if (userName.matches("^[A-Za-z0-9]+$")) {builder = QueryBuilders.boolQuery().must(QueryBuilders.wildcardQuery("userName", ("*" + userName + "*").toLowerCase()));} else {builder = QueryBuilders.boolQuery().must(QueryBuilders.matchPhraseQuery("userName", userName.toLowerCase()));}SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(builder).withPageable(pageable).build();return elasticsearchTemplate.queryForPage(searchQuery, OpenUser.class);}
}