php使用Canal监听msyql的过程详解

kka1220

这篇文章主要介绍了php使用Canal监听msyql的全过程,文中通过图文和代码示例讲解的非常详细,对大家的学习或工资有一定的帮助,需要的朋友可以参考下

−

目录

• 安装Java
• 设置mysql用户权限
• 修改mysql配置
• 安装canal
• 修改实例配置
• 启动和停止
• php测试
  

• canal需要java8
• 去官网下载java8
  

安装Java

1. #创建目录
   
2. mkdir -p /usr/local/java/
   
3. #解压到目录
   
4. tar zxvf jdk-8u411-linux-x64.tar.gz -C /usr/local/java/
   

复制代码

配置环境变量在 /etc/profile 最后加入

1. export JAVA_HOME=/usr/local/java/jdk1.8.0_411
   
2. export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
   
3. export  PATH=$PATH:$JAVA_HOME/bin
   

复制代码

使之生效

1. source /etc/profile

复制代码

查看是否安装成功

1. java -version
   

复制代码

设置mysql用户权限

1. #创建用户名和密码都为 canal 的用户
   
2. create user 'canal'@'%' identified by 'canal';
   
3. #授予该用户对所有数据库和表的查询、复制主节点数据的操作权限
   
4. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
   
5. FLUSH PRIVILEGES; #重新加载权限
   

复制代码

修改mysql配置

1. vim /etc/my.cnf

复制代码

修改部分内容如下

1. # 开启 binlog
   
2. log-bin=mysql-bin
   
3. #master端的ID号，不能和 canal 的 slaveId 重复；
   
4. server-id=1
   
5. #行级，记录每次操作后每行记录的变化。
   
6. binlog-format=row
   
7. #指定库，缩小监控的范围。
   
8. binlog-do-db=test
   

复制代码

查看是否开启主从

1. show master status;

复制代码

安装canal

下载最新canal

[size=1em]

1. 下载
   
2. wget https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz
   
3. #创建canal目录
   
4. mkdir -p /usr/local/canal/
   
5. #解压到canal目录
   
6. tar -zxvf canal.deployer-1.1.7.tar.gz  -C /usr/local/canal/
   

复制代码

查看canal主配置文件
[size=1em]

1. cat /usr/local/canal/conf/canal.properties
   

复制代码

[size=1em]

查看会看到暴露了三个端口和指定了一个实例

[size=1em]

1. canal.admin.port = 11110
   
2. canal.port = 11111
   
3. canal.metrics.pull.port = 11112
   
4. #多个实例使用逗号分隔: canal.destinations = example1,example2，
   
5. #这里对应/usr/local/canal/conf/example/
   
6. canal.destinations = example
   

复制代码

[size=1em]

canal.destinations：canal能可以收集多个MySQL数据库数据，每个MySQL数据库都有独立的配置文件控制。具体配置规则： conf/目录下，使用文件夹放置，文件夹名代表一个MySQL实例。canal.destinations用于配置需要监控数据的数据库。如果是多个，使用,隔开。
修改实例配置文件

修改实例配置

1. vim /usr/local/canal/conf/example/instance.properties
   

复制代码

在文件最后加入

1. #配置 slaveId ,不能等于 mysql 配置里的 server Id 即可
   
2. canal.instance.mysql.slaveId=10
   
3. #数据库连接
   
4. canal.instance.master.address=127.0.0.1:3306
   
5. #数据库账号密码
   
6. canal.instance.dbUsername=canal
   
7. canal.instance.dbPassword=canal
   
8. #代表数据库的编码方式
   
9. canal.instance.connectionCharset = UTF-8
   
10. #设置白名单，这里是监控test库下所有表
    
11. canal.instance.filter.regex=test\\..*
    
12. #设置黑名单，这里设置排除test库下以_noc结尾的表
    
13. canal.instance.filter.black.regex=test\\..*_noc
    
14. 
    

复制代码

这个正则表达式 test\.*_noc 的含义是：

1. test：精确匹配字符串 "test"。
   
2. \\.：匹配一个点（.），因为点在正则表达式中有特殊含义，所以需要使用 \\ 进行转义。
   
3. *：匹配零个或多个任意字符。
   
4. _noc：精确匹配字符串 "_noc"。
   
5. 因此，这个正则表达式可以用来匹配所有以 test. 开头且以 _noc 结尾的字符串。在 Canal 的配置中使用这个正则表达式，就可以实现排除以 test. 开头且以 _noc 结尾的数据库和表，而监控其他数据库和表。
   

复制代码

比如我们要查询 test库下users表id为3的用户信息，sql语句可以这么写
SELECT * FROM test.users WHERE id = 3;
这里的test.users就是正则要匹配的地方

[size=1em]

1. 如果系统是1个 cpu，需要将 canal.instance.parser.parallel 设置为 false

复制代码

[size=1em]启动和停止

#启动

1. /usr/local/canal/bin/startup.sh
   

复制代码

1. /usr/local/canal/bin/startup.sh
   

复制代码

#停止

1. /usr/local/canal/bin/stop.sh

复制代码

查看是否启动成功

1. ps -ef | grep canal

复制代码

php测试

使用 canal-php.

1. composer require xingwenge/canal_php

复制代码

新建index.php文件

1. <?php
   
2. require __DIR__.'/vendor/autoload.php';
   
3. use xingwenge\canal_php\CanalClient;
   
4. use xingwenge\canal_php\CanalConnectorFactory;
   
5. use xingwenge\canal_php\Fmt;
   
6. try {
   
7.     $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SOCKET_CLUE);
   
8.     # $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SWOOLE);
   
9.     $client->connect("127.0.0.1", 11111);
   
10.     $client->checkValid();
    
11.     //设置过滤tes库t下的所有表
    
12.     $client->subscribe("1001", "example", "test.*");
    
13.     while (true) {
    
14.         $message = $client->get(100);
    
15.         if ($entries = $message->getEntries()) {
    
16.             foreach ($entries as $entry) {
    
17.                 Fmt::println($entry);
    
18.             }
    
19.         }
    
20.         sleep(1);
    
21.     }
    
22.     $client->disConnect();
    
23. } catch (\Exception $e) {
    
24.     echo $e->getMessage(), PHP_EOL;
    
25. }
    
26. 
    

复制代码

运行php文件

1. php index.php
   

复制代码

eventType：1、是新增行，2、修改行，3、删除行、4、新增表

也可以直接将数据写入rabbitmq

修改conf/canal.properties

[size=1em]

1. vim conf/canal.properties
   

复制代码

[size=1em]修改RabbitMQ配置如下
[size=1em]

1. rabbitmq.host = 127.0.0.1
   
2. rabbitmq.virtual.host = /
   
3. #交换机名称
   
4. rabbitmq.exchange =exchange.canal
   
5. #队列名称
   
6. rabbitmq.queue = canal_queue
   
7. #路由键名
   
8. rabbitmq.routingKey = canal-routing-key
   
9. #账号密码
   
10. rabbitmq.username =admin
    
11. rabbitmq.password =admin
    
12. #路由模式，这里是严格模式
    
13. rabbitmq.deliveryMode =direct
    

复制代码

[size=1em]

修改实例配置conf/example/instance.properties

[size=1em]

1. vim conf/example/instance.properties
   

复制代码

[size=1em]修改部分内容如下
[size=1em]

1. #rabbitmq的路由键配置
   
2. #这里与canal.properties里的rabbitmq.routingKey一样
   
3. canal.mq.topic=canal-routing-key
   
4. canal.mq.partition=0
   
5. canal.instance.multi.stream.on=false
   
6. 
   
7. #rabbitmq 的 routing key
   
8. #dynamic topic route by schema or table regex
   
9. #canal.mq.dynamicTopic=mytest1.user,topic2:mytest2\\..*,.*\\..*
   
10. #hash partition config
    
11. #canal.mq.enableDynamicQueuePartition=false
    
12. #canal.mq.partitionsNum=3
    
13. #canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
    
14. #canal.mq.partitionHash=test.table:id^name,.*\\..*
    
15. 
    

复制代码