canal+kafka实现mysql与redis数据同步(centos7)_canal kafka redis

为什么不直接使用 canal 同步数据到redis中？

回答:  数据同步的代码和业务逻辑代码搅合在一起不方便维护。

目前cannal的最新版支持三种消息队列，kafka , rocketmq(有bug) rabbitMq   因此本文使用kafka作为mysql同步数据到redis的消息队列

kafka2.8以后的版本不在依赖zookeeper   本文采用kafka2.12的版本使用自带的kraft代替zookeeper

官网搭建文档 ​​​​​​https://github.com/alibaba/canal/wiki/Canal-Admin-QuickStart  

配置mysql数据库 vim /etc/my.cnf

[mysqld]
server-id=1
log-bin = mysql-bin
binlog-format = ROW

重启数据库  systemctl restart mysqld

检查binlog是否开启

> show variables like '%log_bin%';

创建canal 用户给授权给canal_manager数据库
CREATE USER 'canal''@%' IDENTIFIED by 'canal';
GRANT ALL PRIVILEGES ON  canal_manager.* TO canal@'%' IDENTIFIED by 'canal';
 

搭建canal 官网 Releases · alibaba/canal · GitHub

获取安装包 canal.deployer-1.1.6.tar.gz     canal.admin-1.16.tar.gz

mkdir cannal

mv canal.deployer-1.1.6.tar.gz canal    && cd canal

tar -zxvf canal.deployer-1.1.6.tar.gz 

vim /conf/canal.properties

# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = kafka
#########                    Kafka                   #############
kafka.bootstrap.servers = 127.0.0.1:9092
#########               destinations            #############
canal.destinations = example

修改cannal配置文件 canal/conf/example

vim instance.properties

# position info
canal.instance.master.address=127.0.0.1:3306
canal.instance.master.journal.name=mysql-bin.000005    # SHOW MASTER STATUS 获取;
canal.instance.master.position=154                        #binlog日志位置
# username/password
canal.instance.dbUsername=cannal
canal.instance.dbPassword=yourpassword
# mq config   消息队列的topic 我这里的topic名称是devops 根据自身情况而定
canal.mq.topic=devops
 

启动canal应用  查看进程

./canal/bin/startup.sh   

查看日志是否运行成功  less canal.log

安装canal.admin 

安装zookeeper

yum -y install java-1.8.0-openjdk.x86_64
 
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz
 
 
 
tar -zxvf zookeeper-3.4.6.tar.gz
 
 
 
 
cd zookeeper-3.4.6/   && mkdir data   
 
 
 
cp ./conf/zoo_sample.cfg ./conf/zoo.cfg
 
 
 
vim zoo.cfg 
修改配置
dataDir=/usr/local/zookeeper-3.4.6/data
 
 
 
 
 
./zkServer.sh start
 

安装kafka3.0 

Kafka官网地址: Apache Kafka   去下载binary Downloads  二进制包 不要下载源码包

$ tar -xzf kafka_2.13-3.3.1.tgz
$ cd kafka_2.13-3.3.1/congfig/kraft/
 

修改配置 vim kraft/server.properties

node.id=1
controller.quorum.voters=1@localhost:9093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://localhost:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAI
log.dirs=/usr/local/script/kafka/logs/kraft_log/    #单机主要该这里 meta数据的存储位置

生成一个唯一集群id kafka3.x专有的   会生成uuid字符串

../../bin/kafka-storage.sh random-uuid

格式化操作，会在配置的kraft_log中出现一个meta表

../../bin/kafka-storage.sh format -t ODmFtRXAQl6zabcZfzScnxw -c /usr/local/script/kafka/config/kraft/server.properties

启动kafka

./bin/kafka-server-start.sh -daemon config/kraft/server.properties   后台启动

./bin/kafka-server-start.sh config/kraft/server.properties     前端启动

创建一个topic   topic名称为devops

./kafka-topics.sh --create  --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic devops

查看kafka topic列表

./bin/kafka-topics.sh --list --bootstrap-server localhost:9092

查看group id 

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

查看topic名称为devops的消息内容

./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic devops

验证是否消息可以正常生产与消费

创建生产者     输入hello

./kafka-console-producer.sh  --bootstrap-server localhost:9092  --topic devops

重新开一个中端，创建消费者   并在消费端查看

./kafka-console-consumer.sh --bootstrap-server  localhost:9092 -topic devops --from-beginning

至此kafka3.x搭建完毕

kafka 常用命令 

查看组列表    消费之后就会出现组id 

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

kafka 新建一个组 命名为devops 

springboot 中 kafka 报错

2024-02-14 14:03:34,138 - [Consumer clientId=consumer-devops-4, groupId=devops] Bootstrap broker 192.168.217.140:9092 (id: -1 rack: null) disconnected
2024-02-14 14:03:35,739 - [Consumer clientId=consumer-devops-3, groupId=devops] Node -1 disconnected.
2024-02-14 14:03:35,739 - [Consumer clientId=consumer-devops-3, groupId=devops] Connection to node -1 (/192.168.217.140:9092) could not be established. Broker may not be available.

原因： 9093端口 无法telnet   broke 代表一台kafka服务 说明kafka 服务有问题 没有对外开放 多测试几遍telnet  

# 允许外部端口连接                                            
listeners=PLAINTEXT://0.0.0.0:9092  
# 外部代理地址                                                
advertised.listeners=PLAINTEXT://121.201.64.12:9092