关于分布式:使用-docker-运行-RocketMQ-Canal-ElasticSearch-Golang-示例

6次阅读

共计 5749 个字符，预计需要花费 15 分钟才能阅读完成。

在很多业务状况下，咱们都会在零碎中引入 ElasticSearch 搜索引擎作为做全文检索的优化计划。

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

上面我会以一个 blog 文章治理 为例来演示 canal+RocketMQ 用Golang实现 MySQL 与ElasticSearch的数据同步。

示例地址：https://gitee.com/thepoy/Rock…

尽量不要在 macOS 中应用，创立的容器多多少少会有问题，出问题时很难找到症结所在，而在 linux 零碎中应用则一切正常。

RocketMQ是没有官网镜像的，所以须要在本地创立：

cd rocketMQ
docker build --no-cache -f Dockerfile -t rocketmq:4.8.0 --build-arg version=4.8.0 .

可依据本人的需要对 Dockerfile 进行批改

批改环境变量文件 .env 中的主机地址为本人的 ip 地址，而后应用 rocketMQ 目录中的配置文件创立容器：

docker-compose --file compose.yml up

应用我的项目根目录中的配置文件创立 mysql、canal-admin 和canal-server容器：

cd ..
docker-compose --file compose.yml up

也有一个环境变量文件须要批改，另外，compos 文件中的信息也须要依据须要批改，如 mysql 的 root 明码。

创立 mysql 容器时也创立了 canal 账号，须要为这个账号受权。

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

http://localhost:8089，关上后 …admin账号登录，默认明码为123456，治理后盾的界面如下图所示：

因为 compose.yml 文件中曾经配置了 canal-server，所以在后盾中能看见曾经启动的一个 server。

点击侧边栏的 Instance 治理，抉择 新建 Instance，抉择那个惟一的主机，再点击 载入模板，批改上面的一些参数：

# 勾销第 3 行中 mysql slaveId 的正文，轻易批改为一个数字（不能是 1，因为 mysql 的 server_id=1）canal.instance.mysql.slaveId=1234
# 批改 mysql 的地址，canal-admin 容器中也有一个 mysql 实例，咱们不应用这个 mysql，而应用独自的 mysql 容器
canal.instance.master.address=192.168.31.129:3306
# 改成本人的数据库信息(须要监听的数据库, 新建一个 database 就能够)，这一行须要增加
canal.instance.defaultDatabaseName = blog
# table regex 须要过滤的表 这里数据库的中所有表
canal.instance.filter.regex = .\*\\..\*
# MQ 配置 日志数据会发送到 blog_articles 这个 topic 上
canal.mq.topic=blog_articles

实例名称轻易填一个就行。

创立好的新实例默认是进行状态，将其启动。

创立 database 和 table：

CREATE DATABASE IF NOT EXISTS `blog`;
USE blog;
CREATE TABLE IF NOT EXISTS `blog_articles` (
    `id` INT AUTO_INCREMENT PRIMARY KEY NOT NULL,
    `title` VARCHAR(100) NOT NULL UNIQUE,
    `content` TEXT NOT NULL,
    `created_date` VARCHAR(10) NOT NULL
);

批改上面的参数：

# 默信是 tcp, 批改为 rocketMQ
canal.serverMode = rocketMQ
##################################################
#########             RocketMQ         #############
##################################################
rocketmq.producer.group = blog
rocketmq.namesrv.addr = 192.168.31.129:9876

保留后 server 会重启，这时关上 rocketMQ 控制台，可能看到新减少了一个主题blog_articles：

能够通过增加一行数据来测试是否胜利：

INSERT INTO blog.blog_articles
(title, content, created_date)
VALUES('test1', '这是第 1 个测试文章', '2020-01-01');

增加后，在 rocketMQ 控制台查看音讯：

能够看到，增加数据的音讯曾经产生期待生产。

elasticsearch 容器会在应用配置文件创立 Canal 时一起创立，须要留神的是，如果你想批改 elasticsearch 的 tag，能够在 .env 文件中批改 ES_TAG 的值。

我没有创立 Kibana 容器，有需要的话能够自行创立。

当数据库发生变化时，Canal 会将变动信息发送到 RocketMQ 中，所以咱们只须要生产 RocketMQ 中的音讯就能够做到即时或很快地将变动的数据同步到 Elasticsearch 中。

常量

const (
    // topic 在 Canal 中曾经配置了，这里肯定不能写错
    topic              string = "blog_articles"
    // 消费者组能够自定义，但要与 2.5 节中设置的 rocketmq.producer.group 雷同
    consumerGroup      string = "blog"
)

从环境变量中获取host，并生成server：

var (
    server string
    Host   string
)

func init() {Host = os.Getenv("HOST")
    if Host == "" {Host = "localhost"}
    server = Host + ":9876"
}

构造体的设计

尽管代码中没有用到这个构造体，但我感觉须要拿进去聊一聊：

type ChangedData struct {
    // 变动的文档汇合
    Data []es.Document `json:"data"`
    // 发生变化的数据库
    Database string `json:"database"`
    // 数据库内执行工夫
    ES uint64 `json:"es"`
    // 就是 id
    ID uint `json:"id"`
    // 是否为 DDL 语句，create database、create table、alter table
    IsDDL bool `json:"isDdl"`
    // 表构造的字段类型
    MysqlType map[string]string `json:"mysqlType"`
    // 主键名称
    PrimaryKeyNames []string `json:"pkNames"`
    // sql 语句
    SQL string `json:"sql"`
    // sql 语句类型
    SqlType map[string]uint `json:"sqlType"`
    // 表名称
    Table string `json:"table"`
    // 操作类型，(新增)INSERT、(更新)UPDATE、(删除)DELETE、(删除表)ERASE 等等
    Type string `json:"type"`
    // 数据库内解析工夫
    Timestamp uint `json:"ts"`
    // 旧数据
    Old []map[string]string `json:"old"`
}

其中 es.Document 构造如下：

type Document struct {
    ID          string `json:"id,omitempty"`
    Title       string `json:"title,omitempty"`
    Content     string `json:"content,omitempty"`
    CreatedDate string `json:"created_date,omitempty"`
}

应用第三方 json 库

这也是为什么没用到下面的构造体的起因。

应用 json 规范库解决音讯数据并同步到 es 中，齐全是小题大做，会节约很多的性能。

data := gjson.Get(string(msg.body), "data")

应用 gjson 库，能够不便地从 json 字符串中获取想要的数据，并进行后续解决，无需将整个 json 反序列化。

应用 context 阻塞或退出生产线程

启动生产订阅后，阻塞多久，就会生产多久，为了可能管制何时完结生产，这里应用 context 的cancle()函数管制：

    err = c.Start()
    ...

    select {case <-ctx.Done():
        fmt.Println(strings.Repeat("*", 60))
        fmt.Println("shutdown consumer")
        fmt.Println(strings.Repeat("*", 60))
    }

    err = c.Shutdown()
    ...

es 的代码是通用的，没有特地阐明的意义，间接看代码即可。

联合 RocketMQ 和 Elasticsearch 的代码，就能实现 音讯的即时生产 和文档的即时更新。

须要从音讯中取出的数据

下面的构造体对每个字段都有正文，此示例只取 data、old 和type三个字段：

// 将音讯体解析成 gjson.Result
body := gjson.Parse(string(msg.Body))
// 从音讯体中取 data
data := body.Get("data").Array()
// 从音讯体中取 old
old := body.Get("old").Array()
// 从音讯体中取 type
canalTypeStr := body.Get("type").String()

依据不同的操作以不同的形式更新数据

本示例中的仅包含非 DDL 操作，仅限于根本的增、删、改，因为数据已同步到 es 中，所以查应该在 es 中进行。

switch canalType {
    case canal.DELETE:
    ...
    case canal.UPDATE:
    ...
    case canal.INSERT:
    ...
    default:
    log.Fatal("未知操作", canalType)
}

设置环境变量（可选操作）：

export HOST=192.168.31.129

运行示例，示例我的项目在 core 目录中：

cd core
go run main.go

而后在数据库中增加一篇文章：

INSERT INTO blog.blog_articles
(title, content, created_date)
VALUES('test9', '这是第 9 篇测试文章', '2020-01-01');

在终端中就能看见日志：

...
2021/05/08 15:05:41 已创立新的文档：map[content: 这是第 9 篇测试文章 created_date:2020-01-01 id:10 title:test9]
...

在 es 中查问一下 id=10 的文档：

curl -X GET "http://localhost:9200/canal_es/_doc/10?pretty"

查问后果：

{
  "_index" : "canal_es",
  "_type" : "_doc",
  "_id" : "10",
  "_version" : 1,
  "_seq_no" : 14,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "content" : "这是第 9 篇测试文章",
    "created_date" : "2020-01-01",
    "id" : "10",
    "title" : "test9"
  }
}

在数据库中更新一下这篇文章的创立日期：

UPDATE blog.blog_articles
SET created_date='2009-04-15'
WHERE id=10;

终端日志：

2021/05/08 15:15:08 文档已存在，行将更新...
[200 OK] {"_index":"canal_es","_type":"_doc","_id":"10","_version":2,"result":"updated","_shards":{"total":2,"successful":1,"failed":0},"_seq_no":15,"_primary_term":1}
2021/05/08 15:15:08 已更新文档：id=10, new-data=map[created_date:2009-04-15]

再查问一下这篇文章信息，后果为：

{
  "_index" : "canal_es",
  "_type" : "_doc",
  "_id" : "10",
  "_version" : 2,
  "_seq_no" : 15,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "content" : "这是第 9 篇测试文章",
    "created_date" : "2009-04-15",
    "id" : "10",
    "title" : "test9"
  }
}

可见，创立日期曾经更新。

上面删除这篇文章：

DELETE FROM blog.blog_articles
WHERE id=10;

终端日志：

2021/05/08 15:18:03 行将删除文档  10
2021/05/08 15:18:03 已删除：{"id":"10","title":"test9","content":"这是第 9 篇测试文章","created_date":"2009-04-15"}

再查问一下这篇文档：

{
  "_index" : "canal_es",
  "_type" : "_doc",
  "_id" : "10",
  "found" : false
}

es 中也已删除此文章。

示例完结。

正文完

分布式

发表至：分布式

2021-05-28

0

关于分布式:浅析大数据框架-Hadoop～

关于分布式:分布式框架原理与实践李林峰笔记

关于分布式:离线实时一体化数仓与湖仓一体云原生大数据平台的持续演进

关于分布式:EventBridge消息路由｜高效构建消息路由能力

关于java:别再乱打日志了这样才是定位-bug-打日志的方式

关于分布式:使用-docker-运行-RocketMQ-Canal-ElasticSearch-Golang-示例

0 引言

1 RocketMQ

2 Canal

2.1 创立容器

2.2 为 canal 账号受权

2.3 关上 canal 治理后盾

2.4 配置实例 / Instance

2.5 配置 canal-server

3 Elasticsearch

4 代码设计

4.1 RocketMQ

常量

构造体的设计

应用第三方 json 库

应用 context 阻塞或退出生产线程

4.2 Elasticsearch

4.3 二者联合

须要从音讯中取出的数据

依据不同的操作以不同的形式更新数据

5 操作后果

站内搜索