搭建Canal+Zookeeper+Kafka+Redis二、canal能做什么

前言：模拟canal监听binlog，提供给其它服务运行

经过几个小时的实践：在自己电脑上（Mac环境）跑起来（小鹿乱撞），接下来一步一步取探究遇到的问题及实验结果，下图是前期开了几个窗口来观察这几个服务运行状态

搭建Canal+Zookeeper+Kafka+Redis二、canal能做什么

1、前期准备: mysql安装运行、kafka安装运行、zookeeper安装运行、canal安装运行

mysql安装不做过多介绍了，网上都可以找到

kafka搭建：

kafka安装：
安装 kafka
(安装kafka前应在本地先安装java的jdk，因为zookeeper需要依赖java虚拟机)
wget http://mirrors.shuosc.org/apache/kafka/2.0.0/kafka_2.11-2.0.0.tgz
tar -zxvf kafka_2.11-2.0.0.tgz
cd /usr/local/kafka_2.11-2.0.0/
vim /usr/local/kafka/config/server.properties

打开/config/server.properties配置文件，修改日志目录：
broker.id=1
log.dir=/data/kafka/logs-1

启动zookeeper（看bin下的命令是否带.sh）
bin/zookeeper-server-start -daemon ../config/zookeeper.properties &
bin/kafka-server-start ../config/server.properties

bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
Or
kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

查看topic命令：
kafka-topics --list --zookeeper localhost:2181

使用 kafka-console-producer.sh 发送消息：
kafka-console-producer --broker-list localhost:9092 --topic test

消费消息
使用 kafka-console-consumer.sh 接收消息并在终端打印：
 kafka-console-consumer --bootstrap-server localhost:9092 --topic test --from-beginning


安装mysql
查找mysql路径：
mysql --help --verbose | grep my.cnf

/etc/my.cnf /etc/mysql/my.cnf /usr/local/mysql/etc/my.cnf ~/.my.cnf

sudo /etc/my.cnf
内容：
[mysqld]
# log_bin
log-bin = mysql-bin #开启binlog
binlog-format = ROW #选择row模式
server_id = 1 #配置mysql replication需要定义，不能喝canal的slaveId重复

配置Canal：

找到canal.deployer-1.1.4/conf目录下的canal.properties配置文件：

# tcp, kafka, RocketMQ 这里选择kafka模式
canal.serverMode = kafka
# 解析器的线程数，打开此配置，不打开则会出现阻塞或者不进行解析的情况
canal.instance.parser.parallelThreadSize = 16
# 配置MQ的服务地址，这里配置的是kafka对应的地址和端口
canal.mq.servers = 127.0.0.1:9092
# 配置instance，在conf目录下要有example同名的目录，可以配置多个
canal.destinations = example
然后配置instance，找到/conf/example/instance.properties配置文件：

## mysql serverId , v1.0.26+ will autoGen(自动生成，不需配置)
# canal.instance.mysql.slaveId=0

# position info
canal.instance.master.address=127.0.0.1:3306
# 在Mysql执行 SHOW MASTER STATUS;查看当前数据库的binlog
canal.instance.master.journal.name=mysql-bin.000006
canal.instance.master.position=4596
# 账号密码
canal.instance.dbUsername=canal
[email protected]****
canal.instance.connectionCharset = UTF-8
#MQ队列名称
canal.mq.topic=canaltopic
#单队列模式的分区下标
canal.mq.partition=0
配置完成后，就可以启动canal了。

封装Redis客户端：

环境搭建完成后，我们可以写代码了。

首先引入Kafka和Redis的maven依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
在application.yml文件增加以下配置：

spring:  
  redis:
    host: 127.0.0.1
    port: 6379
    database: 0
    password: 123456
封装一个操作Redis的工具类：

@Component
public class RedisClient {

    /**
     * 获取redis模版
     */
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 设置redis的key-value
     */
    public void setString(String key, String value) {
        setString(key, value, null);
    }

    /**
     * 设置redis的key-value，带过期时间
     */
    public void setString(String key, String value, Long timeOut) {
        stringRedisTemplate.opsForValue().set(key, value);
        if (timeOut != null) {
            stringRedisTemplate.expire(key, timeOut, TimeUnit.SECONDS);
        }
    }

    /**
     * 获取redis中key对应的值
     */
    public String getString(String key) {
        return stringRedisTemplate.opsForValue().get(key);
    }

    /**
     * 删除redis中key对应的值
     */
    public Boolean deleteKey(String key) {
        return stringRedisTemplate.delete(key);
    }
}

创建MQ消费者进行同步：

在application.yml配置文件加上kafka的配置信息：

spring:
  kafka:
   # Kafka服务地址
    bootstrap-servers: 127.0.0.1:9092
    consumer:
      # 指定一个默认的组名
      group-id: consumer-group1
      #序列化反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringDeserializer
      value-serializer: org.apache.kafka.common.serialization.StringDeserializer
      # 批量抓取
      batch-size: 65536
      # 缓存容量
      buffer-memory: 524288
根据上面Kafka消费命令那里，我们知道了json数据的结构，可以创建一个CanalBean对象进行接收：

public class CanalBean {
    //数据
    private List<TbCommodityInfo> data;
    //数据库名称
    private String database;
    private long es;
    //递增，从1开始
    private int id;
    //是否是DDL语句
    private boolean isDdl;
    //表结构的字段类型
    private MysqlType mysqlType;
    //UPDATE语句，旧数据
    private String old;
    //主键名称
    private List<String> pkNames;
    //sql语句
    private String sql;
    private SqlType sqlType;
    //表名
    private String table;
    private long ts;
    //(新增)INSERT、(更新)UPDATE、(删除)DELETE、(删除表)ERASE等等
    private String type;
    //getter、setter方法
}
public class MysqlType {
    private String id;
    private String commodity_name;
    private String commodity_price;
    private String number;
    private String description;
    //getter、setter方法
}
public class SqlType {
    private int id;
    private int commodity_name;
    private int commodity_price;
    private int number;
    private int description;
}
最后就可以创建一个消费者CanalConsumer进行消费：

@Component
public class CanalConsumer {
 //日志记录
    private static Logger log = LoggerFactory.getLogger(CanalConsumer.class);
 //redis操作工具类
    @Resource
    private RedisClient redisClient;
 //监听的队列名称为：canaltopic
    @KafkaListener(topics = "canaltopic")
    public void receive(ConsumerRecord<?, ?> consumer) {
        String value = (String) consumer.value();
        log.info("topic名称:{},key:{},分区位置:{},下标:{},value:{}", consumer.topic(), consumer.key(),consumer.partition(), consumer.offset(), value);
        //转换为javaBean
        CanalBean canalBean = JSONObject.parseObject(value, CanalBean.class);
        //获取是否是DDL语句
        boolean isDdl = canalBean.getIsDdl();
        //获取类型
        String type = canalBean.getType();
        //不是DDL语句
        if (!isDdl) {
            List<TbCommodityInfo> tbCommodityInfos = canalBean.getData();
            //过期时间
            long TIME_OUT = 600L;
            if ("INSERT".equals(type)) {
                //新增语句
                for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
                    String id = tbCommodityInfo.getId();
                    //新增到redis中,过期时间是10分钟
                    redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);
                }
            } else if ("UPDATE".equals(type)) {
                //更新语句
                for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
                    String id = tbCommodityInfo.getId();
                    //更新到redis中,过期时间是10分钟
                    redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);
                }
            } else {
                //删除语句
                for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {
                    String id = tbCommodityInfo.getId();
                    //从redis中删除
                    redisClient.deleteKey(id);
                }
            }
        }
    }
}

测试MySQL与Redis同步：

mysql对应的表结构如下：

CREATE TABLE `tb_commodity_info` (
  `id` varchar(32) NOT NULL,
  `commodity_name` varchar(512) DEFAULT NULL COMMENT '商品名称',
  `commodity_price` varchar(36) DEFAULT '0' COMMENT '商品价格',
  `number` int(10) DEFAULT '0' COMMENT '商品数量',
  `description` varchar(2048) DEFAULT '' COMMENT '商品描述',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品信息表';
首先在MySQL创建表。然后启动项目，接着新增一条数据：

INSERT INTO `canaldb`.`tb_commodity_info` (`id`, `commodity_name`, `commodity_price`, `number`, `description`) VALUES ('3e71a81fd80711eaaed600163e046cc3', '叉烧包', '3.99', '3', '又大又香的叉烧包，老人小孩都喜欢');
tb_commodity_info表查到新增的数据：

搭建Canal+Zookeeper+Kafka+Redis二、canal能做什么

debug定位的时候：会监听到canaltopic消费队列，此处就可以进行逻辑处理了。

搭建Canal+Zookeeper+Kafka+Redis二、canal能做什么

以上就是自己动手做的🔗，如果感兴趣的话，也希望看到的能够亲手做一下，也希望能够记录自己。

接下来想说：虽说知其然不知其所以然也不行

Canal工作原理：

MySQL主备复制原理

MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

canal 工作原理

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

搭建Canal+Zookeeper+Kafka+Redis二、canal能做什么

二、canal能做什么

以下参考canal官网。

与其问canal能做什么，不如说数据同步有什么作用。

但是canal的数据同步不是全量的，而是增量。基于binary log增量订阅和消费，canal可以做：

数据库镜像
数据库实时备份
索引构建和实时维护
业务cache(缓存)刷新
带业务逻辑的增量数据处理

搭建Canal+Zookeeper+Kafka+Redis二、canal能做什么

未完待续......

备注：

https://github.com/alibaba/canal

https://mp.weixin.qq.com/s/4-3AlDygYxFOUAw0iqAsqg

搭建Canal+Zookeeper+Kafka+Redis二、canal能做什么

kafka搭建：

配置Canal：

封装Redis客户端：

创建MQ消费者进行同步：

测试MySQL与Redis同步：

Canal工作原理：

二、canal能做什么

继续阅读

Swagger-UI的使用一、添加依赖二、创建配置类三、注解

将swagger-ui中的接口导入postman

聊一聊Swagger-UI的那些事情软件简介Swagger 原生Bootstrap-ui Bootstrap的UIKnife4j Bootstrap的升级版的UI

springboot使用 swagger-ui 2.10.5 有关版本更新带来的问题

Spring Cloud整合Sleuth，当请求完成后，Zipkin没有链路信息

Redis订阅了一段时间后订阅失效了（ redisTemplate.convertAndSend）

2022秋招面试总结（cpp+java+测开）百度测开一面字节后端一面虾皮后端一面虾皮后端二面

数据迁移方法数据迁移原则数据迁移之双写方案数据迁移之级联同步方案

微服务-性能压测\缓存redis和分布式锁redisson和SpringCache

Nacos 2.0 升级前后性能对比压测

Spring数据和Redis

redis集群数据一致性_RedisRaft为Redis集群带来强大的数据一致性

java 8 Lamda 求 List 中 Bigdecimal类型的各种值求最大值求最小值求和求平均值

supervisor 管理redis 和httpd 环境centos7

Redis简介一(单机版)发展历程Redis

(SpringBoot)日志种类：log、monitor、access、out、gc、backup