canal +RocketMQ实现MySQL与ElasticSearch数据同步

Posted 请叫我东子

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了canal +RocketMQ实现MySQL与ElasticSearch数据同步相关的知识,希望对你有一定的参考价值。

1.引言

在很多业务情况下,我们都会在系统中引入ElasticSearch搜索引擎作为做全文检索的优化方案。

如果数据库数据发生更新,这时候就需要在业务代码中写一段同步更新ElasticSearch的代码。

这种数据同步的代码跟业务代码耦合性非常高,并且使得代码的可读性降低,于是乎,我们能不能把这些数据同步的代码抽出来形成一个独立的模块呢?肯定是可以的。

下面我会以一个CMS文章管理为例来演示canal+RocketMQ实现mysqlElasticSearch数据同步。

2.技术栈

如果你还对SpringBootcanalRocketMQMySQLElasticSearch 不是很了解的话,这里我为大家整理个它们的官网网站,如下

这里主要介绍一下canal,其他的自行学习。

2.1 canal定义

canal [kə’næl],译意为水道/管道/沟渠,主要用途是基于 MySQL 数据库增量日志解析,提供增量数据订阅和消费.。

2.2 canal工作原理

  • canal 模拟 MySQL slave 的交互协议,伪装自己为 MySQL slave ,向 MySQL master 发送 dump 协议
  • MySQL master 收到 dump 请求,开始推送 binary log 给 slave (即 canal )
  • canal 解析 binary log 对象(原始为 byte 流)
2.3 架构


说明:

  • server代表一个canal运行实例,对应于一个jvm
  • instance对应于一个数据队列 (1个server对应1…n个instance)

instance模块:

  • eventParser (数据源接入,模拟slave协议和master进行交互,协议解析)
  • eventSink (Parser和Store链接器,进行数据过滤,加工,分发的工作)
  • eventStore (数据存储)
  • metaManager (增量订阅&消费信息管理器)

到这里我们对canal有了一个初步的认识,接下我们就进入实战环节。

3.环境准备

3.1 MySQL 配置

对于自建 MySQL , 需要先开启 Binlog写入功能,配置binlog-formatROW 模式,my.cnf 中配置如下

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义,不要和 canal 的 slaveId 重复

**注意:**针对阿里云 RDS for MySQL , 默认打开了 binlog , 并且账号默认具有 binlog dump 权限 , 不需要任何权限或者 binlog 设置,可以直接跳过这一步

授权canal 连接 MySQL 账号具有作为 MySQL slave的权限, 如果已有账户可直接 使用grant 命令授权。

#创建用户名和密码都为canal
CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

3.2 canal的安装和配置

3.2.1 canal.admin安装和配置

canal提供web ui 进行Server管理、Instance管理。

3.2.1.1 下载 canal.admin, 访问 release 页面 , 选择需要的包下载, 如以 1.1.4版本为例

wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.admin-1.1.4.tar.gz

3.2.1.2 解压完成可以看到如下结构:


我们先配置canal.admin之后。通过web ui来配置 cancal server,这样使用界面操作非常的方便。

3.2.1.3 配置修改
vi conf/application.yml
server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8

spring.datasource:
  address: 127.0.0.1:3306
  database: canal_manager
  username: canal
  password: canal
  driver-class-name: com.mysql.jdbc.Driver
  url: jdbc:mysql://$spring.datasource.address/$spring.datasource.database?useUnicode=true&characterEncoding=UTF-8&useSSL=false
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1

canal:
  adminUser: admin
  adminPasswd: admin
3.2.1.4 初始化元数据库

初始化元数据库

mysql -h127.1 -uroot -p

# 导入初始化SQL
> source conf/canal_manager.sql
  • 初始化SQL脚本里会默认创建canal_manager的数据库,建议使用root等有超级权限的账号进行初始化

  • canal_manager.sql默认会在conf目录下,也可以通过链接下载 canal_manager.sql

3.2.1.5 启动
sh bin/startup.sh
3.2.1.6 启动成功,使用浏览器输入http://ip:8089/ 会跳转到登录界面


使用用户名:admin 密码为:123456 登录
登录成功,会自动跳转到如下界面。这时候我们的canal.admin就搭建成功了。

3.2.2 下载 canal.deployer, 访问 release 页面 , 选择需要的包下载, 如以 1.1.4版本为例

wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz


解压完成可以看到如下结构:

进入conf 目录。可以看到如下的配置文件。

我们先对canal.properties 不做任何修改。

使用canal_local.properties的配置覆盖canal.properties

# register ip
canal.register.ip =

# canal admin config
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster =

使用如下命令启动canal server

sh bin/startup.sh local

启动成功。同时我们在canal.admin web ui中刷新 server 管理,可以到canal server 已经启动成功。


这时候我们的canal.server 搭建已经成功。

3.2.3 在canal admin ui 中配置Instance管理

3.2.3.1 新建 Instance

选择Instance 管理-> 新建Instance
填写 Instance名称:cms_article
选择 选择所属主机集群
选择 载入模板
修改默认信息

#mysql serverId
canal.instance.mysql.slaveId = 1234
#position info,需要改成自己的数据库信息
canal.instance.master.address = 127.0.0.1:3306 
canal.instance.master.journal.name = 
canal.instance.master.position = 
canal.instance.master.timestamp = 
#canal.instance.standby.address = 
#canal.instance.standby.journal.name =
#canal.instance.standby.position = 
#canal.instance.standby.timestamp = 
#username/password,需要改成自己的数据库信息
canal.instance.dbUsername = canal  
canal.instance.dbPassword = canal
#改成自己的数据库信息(需要监听的数据库)
canal.instance.defaultDatabaseName = cms-manage
canal.instance.connectionCharset = UTF-8
#table regex 需要过滤的表 这里数据库的中所有表
canal.instance.filter.regex = .\\*\\\\..\\*

# MQ 配置 日志数据会发送到cms_article这个topic上
canal.mq.topic=cms_article
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\\\..*,.*\\\\..*
#单分区处理消息
canal.mq.partition=0

我们这里为了演示之创建一张表。

配置好之后,我需要点击保存。此时在Instances 管理中就可以看到此时的实例信息。

3.2.4 修改canal server 的配置文件,选择消息队列处理binlog

canal 1.1.1版本之后, 默认支持将canal server接收到的binlog数据直接投递到MQ, 目前默认支持的MQ系统有:

  • kafka: https://github.com/apache/kafka
  • RocketMQ : https://github.com/apache/rocketmq

本案例以RocketMQ为例

我们仍然使用web ui 界面操作。点击 server 管理 - > 点击配置

修改配置文件

# ...
# 可选项: tcp(默认), kafka, RocketMQ
canal.serverMode = RocketMQ
# ...
# kafka/rocketmq 集群配置: 192.168.1.117:9092,192.168.1.118:9092,192.168.1.119:9092 
canal.mq.servers = 192.168.0.200:9078
canal.mq.retries = 0
# flagMessage模式下可以调大该值, 但不要超过MQ消息体大小上限
canal.mq.batchSize = 16384
canal.mq.maxRequestSize = 1048576
# flatMessage模式下请将该值改大, 建议50-200
canal.mq.lingerMs = 1
canal.mq.bufferMemory = 33554432
# Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下)
canal.mq.canalBatchSize = 50
# Canal get数据的超时时间, 单位: 毫秒, 空为不限超时
canal.mq.canalGetTimeout = 100
# 是否为flat json格式对象
canal.mq.flatMessage = false
canal.mq.compressionType = none
canal.mq.acks = all
# kafka消息投递是否使用事务
canal.mq.transaction = false

修改好之后保存。会自动重启。

此时我们就可以在rocketmq的控制台看到一个cms_article topic已经自动创建了。

3.2.5 配置ElasticSearch启动

这里我使用的ElasticSearch 6.6.1

es 启动成功了。

我们使用 elasticsearch-head 连接是可以看到节点信息。一会我们就使用 elasticsearch-head 查询es中数据。

4.代码实战

4.1 创建一个springboot 项目

项目结构如下:

4.2 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">
    <parent>
        <artifactId>rockmq-samples</artifactId>
        <groupId>com.lidong.rocketmq</groupId>
        <version>1.0.0</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>springboot-canal-rocketmq-es</artifactId>
    <properties>
        <java.version>1.8</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <spring-boot.version>2.2.5.RELEASE</spring-boot.version>
    </properties>

    <dependencies>

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

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

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.0.4</version>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>$spring-boot.version</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.3.0.RELEASE</version>
                <configuration>
                    <mainClass>com.lidong.RocketmqSyncSamplesApplication</mainClass>
                </configuration>
                <executions>
                    <execution>
                        <id>repackage</id>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

  • spring-boot-starter-data-elasticsearch:操作es依赖库
  • rocketmq-spring-boot-starter:操作rocketmq依赖库

其他就不过多介绍了。大家一看就明白了。

4.3 application的配置
server:
  port: 8085
rocketmq:
  name-server: localhost:9876
spring:
  data:
    elasticsearch:
      cluster-nodes: localhost:9300
      cluster-name: my-application
      repositories:
        enabled: true
  • rocketmq.name-server : rocketmq的namesver
  • spring.data.elasticsearch.cluster-nodes :es 节点地址
  • spring.data.elasticsearch.cluster-name: es节点集群名称
  • spring.data.elasticsearch.repositories.enabled :开启es仓库使用
4.4 创建es操作的实体类和仓库类
4.4.1 EsCmsArticle实体类
package com.lidong.canal.es.entity;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import java.io.Serializable;
import java.util.Date;

/**
 * 文章详情
 *
 * String indexName();//索引库的名称,个人建议以项目的名称命名
 * String type() default "";//类型,个人建议以实体的名称命名
 * short shards() default 5;//默认分区数
 * short replicas() default 1;//每个分区默认的备份数
 * String refreshInterval() default "1s";//刷新间隔
 * String indexStoreType() default "fs";//索引文件存储类型
 *
 **/

@Document(indexName = "canal-rocketmq-es", type = "cms-article")
public class EsCmsArticle implements  Serializable 

	@Id
	private Long courseId;

	/** 标题 */
	private String title;

	/** 摘要 */
	private String abstractX;

	/** 内容 */
	private String content;

	/** 年龄段 */
	private String ageRange;

	/** 图片 */
	private String image;

	/** 查看次数 */
	private Long viewNumber;

	/** 作者 */
	private String author;

	/** 来源 */
	private String source;

	/** 所属分类 */
	private Long classId;

	/** 关键字 */
	private String keyWords;

	/** 描述 */
	private String description;

	/** 文章url */
	private String url;

	/**
	 * 文章状态
	 */
	private Integer status;

	/**
	 * 创建时间
	 */
	private Date createTime;

	/**
	 * 修改时间
	 */
	private Date updateTime;

	public void setCourseId(Long courseId)
	
		this.courseId = courseId;
	

	public Long getCourseId()
	
		return courseId;
	
	public void setTitle(String title)
	
		this.title = title;
	

	public String getTitle()
	
		return title;
	
	public void setAbstractX(String abstractX)
	
		this.abstractX = abstractX;
	

	public String getAbstractX()
	
		return abstractX;
	
	public void setContent(String content)
	
		this.content = content;
	

	public String getContent()
	
		return content;
	
	public void setAgeRange(String ageRange)
	
		this.ageRange = ageRange;
	

	public String getAgeRange()
	
		return ageRange;
	
	public void setImage(String image)
	
		this.image = image;
	

	public String getImage()
	
		return image;
	
	public void setViewNumber(Long viewNumber)
	
		this.viewNumber = viewNumber;
	

	public Long getViewNumber()
	
		return viewNumber;
	
	public void setAuthor(String author)
	
		this.author = author;
	

	public String getAuthor()
	
		return author;
	
	public void setSource(String source)
	
		this.source = source;
	

	public String getSource()
	
		return source;
	
	public void setClassId(Long classId)
	
		this.classId = classId;
	

	public Long getClassId()
	
		return classId;
	
	public void setKeyWords(String keyWords)
	
		this.keyWords = keyWords;
	

	public String getKeyWords()
	
		return keyWords;
	
	public void setDescription(String description)
	
		this.description = description;
	

	public String getDescription()
	
		return description;
	
	public void setUrl(String url)
	
		this.url = url;
	

	public String getUrl()
	
		return url;
	


	public Integer getStatus() 
		return status;
	

	public void setStatus(Integer status) 
		this.status = status;
	

	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;
	

	@Override
	public String toString() 
		return "CmsArticle" +
				"courseId=" + courseId +
				", title='" + title + '\\'' +
				", abstractX='" + abstractX + '\\'' +
				", content='" + content + '\\'' +
				", ageRange='" + ageRange + '\\'' +
				", image='" + image + '\\'' +
				", viewNumber=" + viewNumber +
				", author='" + author + '\\'' +
				", source='" + source + '\\'' +
				", classId=" + classId +
				", keyWords='" + keyWords + '\\'' +
				", description='" + description + '\\'' +
				", url='" + url + '\\'' +
				", status=" + status +
				", createTime=" + createTime +
				", updateTime=" + updateTime +
				'';
	



4.4.2 CmsArticleRepository 仓库类
package com.lidong.canal.rocketmq;

import com.alibaba.fastjson.JSON;
import com.lidong.canal.bean.CanalBean;
import com.lidong.canal.bean.CmsArticle;
import com.lidong.canal.es.entity.EsCmsArticle;
import com.lidong.canal.es.repository.CmsArticleRepository;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Optional;

@Component
@RocketMQMessageListener(
        topic = "cms_article"

以上是关于canal +RocketMQ实现MySQL与ElasticSearch数据同步的主要内容,如果未能解决你的问题,请参考以下文章

Canal利用canal实现mysql实时增量备份并对接kafka

mysql同步数据到redis-增量同步

Canal+Kafka实现MySql与Redis数据一致性

docker部署canal 1.1.6 rocketmq 分区顺序性

docker部署canal 1.1.6 rocketmq 分区顺序性

canal+Kafka实现mysql与redis数据同步