Canal synchronizes MySQL data changes to Kafka (CentOS deployment)

First step ： Environmental preparation

• System CentOS7.9： see CentOS edition : cat /etc/redhat-release; The machine memory is at least 2 individual G Memory , because canal The default maximum memory for startup is 2G.
• jdk edition jdk1.8： Please refer to ：[https://blog.csdn.net/u010132847/article/details/117002124?spm=1001.2014.3001.5501][https_blog.csdn.net_u010132847_article_details_117002124_spm_1001.2014.3001.5501]
• Deploy mysql8 Version of the machine ip： 192.168.187.132,mysql8 For version installation, see ：[https://blog.csdn.net/u010132847/article/details/116905080?spm=1001.2014.3001.5501][https_blog.csdn.net_u010132847_article_details_116905080_spm_1001.2014.3001.5501]
• install canal+kafka Machine ip： 192.168.187.131

The second step ：mysql First open Binlog Write function , To configure binlog-format by ROW Pattern .

• mysql8 The version is enabled by default binlog Functional , This step can actually be ignored .

The third step ： Create a new database canal Account and give it the right to synchronize the database .

• A new user ：create user ‘canal’@‘%’ identified by ‘Canal123!’;

• Give users canal Of SELECT、REPLICATION SLAVE and REPLICATION CLIENT jurisdiction ：

• “mysql_native_password ” Special treatment of ：mysql8 The default with sha2 This plugin To encrypt the password , If the plug-in is not used in the production of user passwords , When the client accesses, it will report “caching_sha2_password Auth failed” Error of , The following error displays .

2020-05-06 17:19:24.944 [destination = example , address = /192.168.187.132:3306 , EventParser] ERROR com.alibaba.otter.canal.common.alarm.LogAlarmHandler - destination:example[com.alibaba.otter.canal.parse.exception.CanalParseException: java.io.IOException: connect /192.168.187.132:3306 failure
Caused by: java.io.IOException: connect /127.0.0.1:3306 failure
        at com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector.connect(MysqlConnector.java:83)
        at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.connect(MysqlConnection.java:89)
        at com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser.preDump(MysqlEventParser.java:86)
        at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$3.run(AbstractEventParser.java:183)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.IOException: caching_sha2_password Auth failed
        at com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector.negotiate(MysqlConnector.java:257)
        at com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector.connect(MysqlConnector.java:80)
        ... 4 more
]

• Therefore, you need to use keywords “mysql_native_password ” The plug-in assigns the account password , as follows ：

The expression is as follows ：alter user ‘canal’@‘%’ identified with mysql_native_password by ‘Canal123!’;

Step four ： Installation and deployment kafka( Include zookeeper)

• Download installation package ：wget [https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.8.0/kafka_2.13-2.8.0.tgz][https_mirrors.tuna.tsinghua.edu.cn_apache_kafka_2.8.0_kafka_2.13-2.8.0.tgz]
• Unzip the installation package ：tar -zxvf kafka_2.13-2.8.0.tgz

• stay /opt/kafka_2.13-2.8.0/bin The directory can see these startup scripts ( With a little attention, you will find that it also contains zookeeper Script , Yes ,kafka The installation package contains zookeeper Service deployment , There is no need to download the deployment separately , Start directly here )：

• stay /opt/kafka_2.13-2.8.0/config The directory can see the configuration of the service ：

• Start... First zookeeper( The configuration file zookeeper.properties No change required )：

command ：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties &

remarks ： here “-daemon” Indicates that the daemon starts .Daemon Process is also called daemon ,linux Most service processes are implemented through daemons . such as 0 Process of no. ( Scheduling process ) ,1 Process of no. (init process ). From its name guard, we can see that it usually runs when the machine starts , Shut down before stopping . So it should not be affected by the terminal . At the same time, it actually runs in the background .

• Restart kafka（ The configuration file config/server.properties Need to change , as follows ）:

zookeeper.connect=192.168.187.131:2181
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://localhost:9092 # This machine ip
# ...

Then start the execution command ：bin/kafka-server-start.sh -daemon config/server.properties &

stop it kafka Service execution command ：bin/kafka-server-stop.sh

Step five ： Installation and deployment canal

• download canal: wget [https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz][https_github.com_alibaba_canal_releases_download_canal-1.1.5_canal.deployer-1.1.5.tar.gz]
• decompression canal（canal Unpack the installation package and put it into the new canal File decompression ）: tar -zxvf [canal.deployer-1.1.5.tar.gz][https_github.com_alibaba_canal_releases_download_canal-1.1.5_canal.deployer-1.1.5.tar.gz]

• The unzipped directory is explained as follows ：

- bin    #  Operation and maintenance script file 
- conf   #  Profile directory 
  canal_local.properties  # canal Local configuration , Generally, it does not need to be changed 
  canal.properties        # canal Service configuration 
  logback.xml             # logback Log configuration 
  metrics                 #  Measurement statistics configuration 
  spring                  # spring- The instance configuration , Main and binlog Position calculation 、 Some policy configurations are related , Can be in canal.properties Choose any of the configuration files 
  example                 #  Instance configuration folder , Generally speaking, a single database corresponds to a separate instance configuration folder 
    instance.properties   #  The instance configuration , Generally refers to the configuration of a single database 
- lib    #  Service dependency package 
- logs   #  Log file output directory 
- plugin #  Supported plug-in directories 
  connector.kafka-1.1.5-jar-with-dependencies.jar     #kafka Dependency package 
  connector.rabbitmq-1.1.5-jar-with-dependencies.jar  #rabbitmq Dependency package 
  connector.rocketmq-1.1.5-jar-with-dependencies.jar  #rocketmq Dependency package 

• Modify the configuration file ：

The configuration file config/canal.properties find “canal.serverMode” Change the value of to “kafka”, Nothing else needs to be changed .

The configuration file config/exmaple/instance.properties：

find “canal.instance.master.address”, Change its value to installation mysql Server's ip And port number .

find “canal.instance.dbUsername” and “canal.instance.dbPassword” The value of is changed to the previously created mysql Database backup account canal And the password , Nothing else needs to be changed .

• start-up canal service ： find /opt/kafka/bin Catalog , Execute the script startup.sh.

Step six ： Test the data synchronization function

• Use other tools to create databases 、 The account number of the table permission is mysql The service creates a database and a table .

• We noticed that config/example/instanct.properties There is one “example” Of topic To configure , This is a kafka Test use topic, There is no need to create , Let's take this topic To test , Here's the picture ：

• stay kafka Installation directory directory /opt/kafka_2.13-2.8.0, perform kafka Of topic Consumer logs , It is found that there are consumption logs that create corresponding tables .

command ：bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --from-beginning --topic example

• Insert another piece of data to see if there is a corresponding message log ：