操作步骤

说明

1

#Docker hub上查找mysql镜像

2

#从docker hub上(阿里云加速器)拉取mysql镜像到本地标签为5.7

docker pull mysql:5.7

3

#创建容器

docker run -di --name=mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=123456 mysql:5.7

-p 代表端口映射，格式为宿主机映射端口:容器运行端口

-e 代表添加环境变量，MYSQL_ROOT_PASSWORD是root用户的登录密码

4

#使用mysql客户端连接

5

#进入MySQL容器

docker exec -it mysql bash

操作步骤

说明

1

#进入MySQL容器

docker exec -it mysql bash

2

#安装vim、ifconfig等命令

#更新软件包

apt-get update

#安装vim

apt-get install -y vim

#安装ifconfig

apt-get install net-tools

3

#配置my.cnf

root@dfbf3fdefbdf:/# vim /etc/mysql/my.cnf

[mysqld]

log-bin=mysql-bin  #添加这一行就ok

binlog-format=ROW #选择row模式

server_id=1  #配置mysql replaction需要定义，不能和canal的slaveId重复

expire_logs_days=7 # binlog文件保存7天

max_binlog_size = 500m # 每个binlog日志文件大小

vim中使用鼠标右键粘贴，失败。vim的模式：

普通模式下输入“:set mouse-=a”，有效

4

#重启mysql容器（dfbf3fdefbdf是容器id）

docker restart dfbf3fdefbdf

5

#进入sql命令行

docker exec -it mysql bash

mysql -uroot -p123456

6

#验证 my.cnf 配置是否生效:

show variables like 'binlog_format';

show variables like 'log_bin';

show master status;

1

创建容器时报错：

WARNING: IPv4 forwarding is disabled. Networking will not work.

ab71b3f089ff3a7538811e79fd3d989b5958383968b1f1eb8ba1dc6f1da3335b

解决方式：

vim /usr/lib/sysctl.d/00-system.conf

添加如下代码：net.ipv4.ip_forward=1

重启网络：systemctl restart network

完成以后，删除当前容器，重新创建新容器。

服务器

主机名

IP

Canal

客户关系管理服务器

node1

192.168.88.10

源端

大数据服务器

node2

192.168.88.20

目标端

操作步骤

说明

1

#安装canalserver镜像

docker pull canal/canal-server:v1.1.2

2

#通过镜像生成canal-server容器

docker run -d --name canal-server \

-e canal.instance.master.address=192.168.88.10:3306 \

-e canal.instance.dbUsername=root \

-e canal.instance.dbPassword=123456 \

-p 11111:11111 \

-d canal/canal-server:v1.1.2

3

#进入canal-server容器

docker exec -it canal-server bash

4

#执行/export/servers/canal/bin目录中的 startup.sh 启动canal

cd /home/admin/canal-server/bin

[root@a9ec635e5c35 bin]# ./startup.sh

控制台如果输出如上，表示canal已经启动成功

5

#进入example日志文件查看是否有报错

cd canal-server/logs/example/

cat example.log

操作步骤

说明

1

#进入sql命令行

docker exec -it mysql bash

mysql -uroot -p123456

2

#创建mysql数据库

create database if not EXISTS test DEFAULT charset utf8 COLLATE utf8_general_ci;

3

#切换到test数据库

use test;

4

#在test数据库创建表

DROP  TABLE  IF  EXISTS  `userinfo`;

CREATE  TABLE  `userinfo` (

  `id`  int(11)  NOT  NULL  AUTO_INCREMENT,

  `name`  varchar(255)  DEFAULT  NULL,

  `age`  int(11)  DEFAULT  NULL,

  PRIMARY  KEY (`id`)

)  ENGINE=InnoDB  DEFAULT  CHARSET=utf8;

操作步骤

说明

1

#进入canal-server容器

docker exec -it canal-server bash

2

#修改 canal/conf目录中的 canal.properties 文件

vi /home/admin/canal-server/conf/canal.properties

#################################################

#########               common argument         ############# 

#################################################

canal.id = 1

canal.ip =

canal.port = 11111  #canal-server监听的端口（TCP模式下，非TCP模式不监听1111端口）

canal.metrics.pull.port = 11112 #canal-server metrics.pull监听的端口

canal.zkServers =   #集群模式下要配置zookeeper进行协调配置，单机模式可以不用配置

# flush data to zk

canal.zookeeper.flush.period = 1000

canal.withoutNetty = false

# tcp, kafka, RocketMQ

canal.serverMode = kafka #canal-server运行的模式，TCP模式就是直连客户端，不经过中间件。kafka和mq是消息队列的模式

# flush meta cursor/parse position to file

canal.file.data.dir = ${canal.conf.dir} #存放数据的路径

canal.file.flush.period = 1000

## memory store RingBuffer size, should be Math.pow(2,n)

canal.instance.memory.buffer.size = 16384

## memory store RingBuffer used memory unit size , default 1kb

canal.instance.memory.buffer.memunit = 1024 

## meory store gets mode used MEMSIZE or ITEMSIZE

canal.instance.memory.batch.mode = MEMSIZE

canal.instance.memory.rawEntry = true

## detecing config　　#这里是心跳检查的配置，做HA时会用到

canal.instance.detecting.enable = false

#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()

canal.instance.detecting.sql = select 1

canal.instance.detecting.interval.time = 3

canal.instance.detecting.retry.threshold = 3

canal.instance.detecting.heartbeatHaEnable = false

# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery

canal.instance.transaction.size =  1024

# mysql fallback connected to new master should fallback times

canal.instance.fallbackIntervalInSeconds = 60

# network config

canal.instance.network.receiveBufferSize = 16384

canal.instance.network.sendBufferSize = 16384

canal.instance.network.soTimeout = 30

# binlog filter config　　

#binlog过滤的配置，指定过滤那些SQL

canal.instance.filter.druid.ddl = true

canal.instance.filter.query.dcl = false

canal.instance.filter.query.dml = false

canal.instance.filter.query.ddl = false

canal.instance.filter.table.error = false

canal.instance.filter.rows = false

canal.instance.filter.transaction.entry = false

# binlog format/image check　

#binlog格式检测，使用ROW模式，非ROW模式也不会报错，但是同步不到数据

canal.instance.binlog.format = ROW,STATEMENT,MIXED 

canal.instance.binlog.image = FULL,MINIMAL,NOBLOB

# binlog ddl isolation

canal.instance.get.ddl.isolation = false

# parallel parser config

#并行解析配置，如果是单个CPU就把下面这个true改为false

canal.instance.parser.parallel = true

## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()

#canal.instance.parser.parallelThreadSize = 16

## disruptor ringbuffer size, must be power of 2

canal.instance.parser.parallelBufferSize = 256

# table meta tsdb info

canal.instance.tsdb.enable = true

canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}

canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;

canal.instance.tsdb.dbUsername = canal

canal.instance.tsdb.dbPassword = canal

# dump snapshot interval, default 24 hour

canal.instance.tsdb.snapshot.interval = 24

# purge snapshot expire , default 360 hour(15 days)

canal.instance.tsdb.snapshot.expire = 360

# aliyun ak/sk , support rds/mq

canal.aliyun.accesskey =

canal.aliyun.secretkey =

#################################################

#########               destinations            ############# 

#################################################

#canal-server创建的实例，在这里指定你要创建的实例的名字，比如test1，test2等，逗号隔开

canal.destinations = example

# conf root dir

canal.conf.dir = ../conf

# auto scan instance dir add/remove and start/stop instance

canal.auto.scan = true

canal.auto.scan.interval = 5

canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml

#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml

canal.instance.global.mode = spring

canal.instance.global.lazy = false

#canal.instance.global.manager.address = 127.0.0.1:1099

#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml

canal.instance.global.spring.xml = classpath:spring/file-instance.xml

#canal.instance.global.spring.xml = classpath:spring/default-instance.xml

##################################################

#########                    MQ                      #############

##################################################

#kafka为bootstrap.servers rocketMQ中为nameserver列表

canal.mq.servers = 192.168.88.20:9092

#发送失败重试次数

canal.mq.retries = 0

#kafka为ProducerConfig.BATCH_SIZE_CONFIG rocketMQ无意义

canal.mq.batchSize = 16384

#kafka为ProducerConfig.MAX_REQUEST_SIZE_CONFIG rocketMQ无意义

canal.mq.maxRequestSize = 1048576

#kafka为ProducerConfig.LINGER_MS_CONFIG , 如果是flatMessage格式建议将该值调大, 如: 200 rocketMQ无意义

canal.mq.lingerMs = 1

#kafka为ProducerConfig.BUFFER_MEMORY_CONFIG rocketMQ无意义

canal.mq.bufferMemory = 33554432

#获取canal数据的批次大小

canal.mq.canalBatchSize = 50

#获取canal数据的超时时间

canal.mq.canalGetTimeout = 100

#是否为json格式 如果设置为false,对应MQ收到的消息为protobuf格式 需要通过CanalMessageDeserializer进行解码

canal.mq.flatMessage = true

canal.mq.compressionType = none

canal.mq.acks = all

修改内容如下

#默认为TCP，也就是你通过官方的example可以在终端查看数据，我们修改为kafka

canal.serverMode = kafka

canal.destinations = example

#kafka地址

canal.mq.servers = 192.168.88.20:9092

3

#修改 canal/conf/example目录中的 instance.properties  文件

vi /home/admin/canal-server/conf/example/instance.properties

#################################################

## mysql serverId , v1.0.26+ will autoGen 

# canal.instance.mysql.slaveId=0

# enable gtid use true/false

canal.instance.gtidon=false

# position info

canal.instance.master.address=192.168.88.10:3306

canal.instance.master.journal.name=

canal.instance.master.position=

canal.instance.master.timestamp=

canal.instance.master.gtid=

# rds oss binlog

canal.instance.rds.accesskey=

canal.instance.rds.secretkey=

canal.instance.rds.instanceId=

# table meta tsdb info

canal.instance.tsdb.enable=true

#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb

#canal.instance.tsdb.dbUsername=canal

#canal.instance.tsdb.dbPassword=canal

#canal.instance.standby.address =

#canal.instance.standby.journal.name =

#canal.instance.standby.position =

#canal.instance.standby.timestamp =

#canal.instance.standby.gtid=

# username/password

canal.instance.dbUsername=root

canal.instance.dbPassword=123456

canal.instance.connectionCharset = UTF-8

# 连接默认数据库

canal.instance.defaultDatabaseName =test

# enable druid Decrypt database password

canal.instance.enableDruid=false

#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

# table regex

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

# table black regex

canal.instance.filter.black.regex=

# mq config

#mq里的topic名

canal.mq.topic=example

canal.mq.partition=0

# hash partition config

#canal.mq.partitionsNum=3

#散列规则定义 库名.表名 : 唯一主键，比如mytest.person: id 1.1.3版本支持新语法

#canal.mq.partitionHash=mytest.person:id,mytest.role:id

#################################################

修改内容如下

# position info

canal.instance.master.address=192.168.88.10:3306

# username/password

canal.instance.dbUsername=root

canal.instance.dbPassword=123456

# 数据库及表过滤，这里我只抽取sourcedb的日志

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

# mq config

canal.mq.topic=example

4

#重启canal-server

bin/restart.sh

5

#启动大数据服务器的kafka集群

bin/kafka-server-start.sh config/server.properties > /dev/null 2>&1 &

6

#启动kafka的消费者命令行

cd /export/services/kafka

./kafka-simple-consumer-shell.sh  --broker-list node2:9092  --topic example

或者

./kafka-console-consumer.sh --bootstrap-server node2:9092 --topic example --from-beginning

7

#在mysql数据库中插入一条sql语句

8

#观察kafka消费者命令行的输出

1

#启动canal-server后，example.log日志错误如下

错误原因：

启动docker时，docker进程会创建一个名为docker0的虚拟网桥，用于宿主机与容器之间的通信。当启动一个docker容器时，docker容器将会附加到虚拟网桥上，容器内的报文通过docker0向外转发。

如果docker容器访问宿主机，那么docker0网桥将报文直接转发到本机，报文的源地址是docker0网段的地址。而如果docker容器访问宿主机以外的机器，docker的SNAT网桥会将报文的源地址转换为宿主机的地址，通过宿主机的网卡向外发送。

因此，当docker容器访问宿主机时，如果宿主机服务端口会被防火墙拦截，从而无法连通宿主机，出现No route to host的错误。

而访问宿主机所在局域网内的其他机器，由于报文的源地址是宿主机ip，因此，不会被目的机器防火墙拦截，所以可以访问。

因此，也可以通过开放相对应的端口即可。

解决方式：

firewall-cmd --zone=public --add-port=3306/tcp --permanent    

firewall-cmd --reload

确认下防火墙是否关闭，如果没有关闭需要关掉

查看防火墙状态：firewall-cmd --state

停止firewall：systemctl stop firewalld.service

禁止firewall开机启动：systemctl disable firewalld.service

操作步骤

说明

1

#进入canal-server容器

docker exec -it canal-server bash

2

#修改 canal/conf目录中的 canal.properties 文件

vi canal-server/conf/canal.properties

canal.mq.servers = 192.168.88.20:9092

3

#修改 canal/conf/example目录中的 instance.properties 文件

canal.instance.master.address=192.168.88.20:3306

canal.instance.dbUsername=root

canal.instance.dbPassword=123456

canal.instance.defaultDatabaseName =itcast_crm

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

4

#重启canal-server

/home/admin/canal-server/bin/restart.sh

操作步骤

说明

1

打开DBeaver客户端软件

2

新建连接

3

配置远程连接参数

4

点击测试是否连接成功

5

创建数据库，库名：it_crm

6

打开：4.资料\2.数据库脚本\it_crm.sql

操作步骤

说明

1

打开mysql客户端程序

2

创建crm数据库

3

执行数据表创建脚本

-- 客户地址表
CREATE TABLE `crm_address` (
  `id` bigint NOT NULL,
  `name` varchar(50) DEFAULT NULL,
  `tel` varchar(20) DEFAULT NULL,
  `mobile` varchar(20) DEFAULT NULL,
  `detail_addr` varchar(100) DEFAULT NULL,
  `area_id` bigint DEFAULT NULL,
  `gis_addr` varchar(20) DEFAULT NULL,
  `cdt` datetime DEFAULT NULL,
  `udt` datetime DEFAULT NULL,
  `remark` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- 客户表
DROP TABLE IF EXISTS `crm_customer`;
CREATE TABLE `crm_customer` (
  `id` bigint NOT NULL,
  `name` varchar(50) DEFAULT NULL,
  `tel` varchar(20) DEFAULT NULL,
  `mobile` varchar(20) DEFAULT NULL,
  `email` varchar(50) DEFAULT NULL,
  `type` bigint DEFAULT NULL,
  `is_own_reg` bigint DEFAULT NULL,
  `reg_dt` datetime DEFAULT NULL,
  `reg_channel_id` bigint DEFAULT NULL,
  `state` int DEFAULT NULL,
  `cdt` datetime DEFAULT NULL,
  `udt` datetime DEFAULT NULL,
  `last_login_dt` datetime DEFAULT NULL,
  `remark` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- 客户-地址-关联表
DROP TABLE IF EXISTS `crm_consumer_address_map`;
CREATE TABLE `crm_consumer_address_map` (
  `id` bigint NOT NULL,
  `consumer_id` bigint DEFAULT NULL,
  `address_id` bigint DEFAULT NULL,
  `cdt` datetime DEFAULT NULL,
  `udt` datetime DEFAULT NULL,
  `remark` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

4

插入一条数据

INSERT INTO `crm_address` VALUES ('10001', '葛秋红', null, '17*******47', '恒大影城南侧小金庄', '130903', null, '2020-02-02 18:51:39', '2020-02-02 18:51:39', null);