客快物流大数据项目(二十六)：客户关系管理服务器
Posted Lansonli
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了客快物流大数据项目(二十六)：客户关系管理服务器相关的知识，希望对你有一定的参考价值。
目录
客户关系管理服务器
客户关系关系服务器需要部署Docker环境，将Mysql与Canal部署到Docker容器中
一、MySQL
1、Mysql部署
| 操作步骤 | 说明 |
| 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 | |
| |
2、Mysql开启binlog日志
- 用来记录mysql中的增加 、 删除 、 修改、清空操作，select操作 不会 保存到binlog中
- 必须要 打开 mysql中的binlog功能，才会生成binlog日志
- binlog日志就是一系列的二进制文件
-rw-rw---- 1 mysql mysql 669 11⽉月 10 21:29 mysql-bin.000001
-rw-rw---- 1 mysql mysql 126 11⽉月 10 22:06 mysql-bin.000002
-rw-rw---- 1 mysql mysql 11799 11⽉月 15 18:17 mysql-bin.00000 开启步骤：
| 操作步骤 | 说明 |
| 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; | |
| |
3、可能遇到的问题
| 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 完成以后，删除当前容器，重新创建新容器。 |
二、Canal介绍
1、Canal简介
canal [kə'næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费
早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。
基于日志增量订阅和消费的业务包括
- 数据库镜像
- 数据库实时备份
- 索引构建和实时维护(拆分异构索引、倒排索引等)
- 业务 cache 刷新
- 带业务逻辑的增量数据处理
当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x
2、Mysql的主备复制原理
- MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
- MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
- MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据
3、Canal的工作原理
- canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
- MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
- canal 解析 binary log 对象(原始为 byte 流)
4、Canal的架构
- server 代表一个 canal 运行实例，对应于一个 jvm
- instance 对应于一个数据队列 （1个 canal server 对应 1..n 个 instance )
- instance 下的子模块
- eventParser: 数据源接入，模拟 slave 协议和 master 进行交互，协议解析
- eventSink: Parser 和 Store 链接器，进行数据过滤，加工，分发的工作
- eventStore: 数据存储
- metaManager: 增量订阅 & 消费信息管理器
EventParser在向mysql发送dump命令之前会先从Log Position中获取上次解析成功的位置(如果是第一次启动，则获取初始指定位置或者当前数据段binlog位点)。mysql接受到dump命令后，由EventParser从mysql上pull binlog数据进行解析并传递给EventSink(传递给EventSink模块进行数据存储，是一个阻塞操作，直到存储成功 )，传送成功之后更新Log Position。流程图如下：
- EventSink起到一个类似channel的功能，可以对数据进行过滤、分发/路由(1:n)、归并(n:1)和加工。EventSink是连接EventParser和EventStore的桥梁。
- EventStore实现模式是内存模式，内存结构为环形队列，由三个指针(Put、Get和Ack)标识数据存储和读取的位置。
- MetaManager是增量订阅&消费信息管理器，增量订阅和消费之间的协议包括get/ack/rollback，分别为：
- Message getWithoutAck(int batchSize)，允许指定batchSize，一次可以获取多条，每次返回的对象为Message，包含的内容为：batch id[唯一标识]和entries[具体的数据对象]
- void rollback(long batchId)，顾名思义，回滚上次的get请求，重新获取数据。基于get获取的batchId进行提交，避免误操作
- void ack(long batchId)，顾名思议，确认已经消费成功，通知server删除数据。基于get获取的batchId进行提交，避免误操作
三、Canal安装部署
| 服务器 | 主机名 | IP | Canal |
| 客户关系管理服务器 | node1 | 192.168.88.10 | 源端 |
| 大数据服务器 | node2 | 192.168.88.20 | 目标端 |
1、Canal部署
| 操作步骤 | 说明 |
| 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 | |
| |
2、Canal测试
- 需求：
将mysql的数据写入到kafka集群
1）创建测试数据库
| 操作步骤 | 说明 |
| 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; |
2）配置canalserver端
| 操作步骤 | 说明 |
| 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 客快物流大数据项目(六十六)：车辆主题 客快物流大数据项目(二十七)：Cloudera Manager简单介绍 |
