mysql+canal+kafka+elasticsearch构

1.相关背景：

在各个互联网公司的项目架构中。有一个系统必不可少，那就是各类的搜索系统。不管是b端商家端还是c端用户。都有各种各样的搜索需求。我们今天就基于mysql，canal，kafka，elasticsearch搭建一个简单的订单搜索平台。

大致流程如下：

1)当一条新的订单数据存入到数据库

2)canal监听到数据变更

3)数据变更存入到kafka中

4)消息监听系统监听kafka消息将数据存入到elasticsearch当中

5)构建搜索api提供用户查询

第一部分：

系统搭建阶段：

1.数据准备过程(mysql已经安装完成):

testdb.sql

DROP SCHEMA IF EXISTS demo_ds;

DROP SCHEMA IF EXISTS demo_ds_0;

DROP SCHEMA IF EXISTS demo_ds_1;

CREATE SCHEMA IF NOT EXISTS demo_ds;

CREATE SCHEMA IF NOT EXISTS demo_ds_0;

CREATE SCHEMA IF NOT EXISTS demo_ds_1;

CREATE TABLE IF NOT EXISTS demo_ds.t_order (order_id BIGINT NOT NULL AUTO_INCREMENT, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_id));

CREATE TABLE IF NOT EXISTS demo_ds.t_order_item (order_item_id BIGINT NOT NULL AUTO_INCREMENT, order_id BIGINT NOT NULL, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_item_id));

CREATE TABLE IF NOT EXISTS demo_ds_0.t_order_0 (order_id BIGINT NOT NULL AUTO_INCREMENT, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_id));

CREATE TABLE IF NOT EXISTS demo_ds_0.t_order_1 (order_id BIGINT NOT NULL AUTO_INCREMENT, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_id));

CREATE TABLE IF NOT EXISTS demo_ds_0.t_order_item_0 (order_item_id BIGINT NOT NULL AUTO_INCREMENT, order_id BIGINT NOT NULL, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_item_id));

CREATE TABLE IF NOT EXISTS demo_ds_0.t_order_item_1 (order_item_id BIGINT NOT NULL AUTO_INCREMENT, order_id BIGINT NOT NULL, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_item_id));

CREATE TABLE IF NOT EXISTS demo_ds_1.t_order_0 (order_id BIGINT NOT NULL AUTO_INCREMENT, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_id));

CREATE TABLE IF NOT EXISTS demo_ds_1.t_order_1 (order_id BIGINT NOT NULL AUTO_INCREMENT, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_id));

CREATE TABLE IF NOT EXISTS demo_ds_1.t_order_item_0 (order_item_id BIGINT NOT NULL AUTO_INCREMENT, order_id BIGINT NOT NULL, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_item_id));

CREATE TABLE IF NOT EXISTS demo_ds_1.t_order_item_1 (order_item_id BIGINT NOT NULL AUTO_INCREMENT, order_id BIGINT NOT NULL, user_id INT NOT NULL, status VARCHAR(50), PRIMARY KEY (order_item_id));

通过mysql客户端导入(source脚本):

导入完成后，查看数据库:

相关表:

2.canal安装：

canal相关介绍:

https://github.com/alibaba/canal

quick start:https://github.com/alibaba/canal/wiki/QuickStart

使用场景:

基于日志增量订阅和消费的业务包括

数据库镜像

数据库实时备份

索引构建和实时维护(拆分异构索引、倒排索引等)

业务 cache 刷新

带业务逻辑的增量数据处理

下面安装过程:

首先我们需要配置mysql.cnf

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

然后需要重启，使用如下命令:

mysql.server restart

接下来需要授权:

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

CREATE USER canalI DENTIFIED BY 'canal';

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENTON*.*TO'canal'@'%';--GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;

FLUSH PRIVILEGES;

接下来下载最新的canal发布包:

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

mkdir /tmp/canal

tar zxvf canal.deployer-1.1.3.tar.gz  -C /tmp/canal

解压目录如下：

配置修改

vi conf/example/instance.properties

## 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 =

canal.instance.connectionCharset = UTF-8

#table regex

canal.instance.filter.regex = .\*\\\\..\*

canal.instance.connectionCharset 代表数据库的编码方式对应到 java 中的编码类型，比如 UTF-8，GBK , ISO-8859-1

如果系统是1个 cpu，需要将 canal.instance.parser.parallel 设置为 false

启动

sh bin/startup.sh

关闭

sh bin/stop.sh