canal来实现mysql的数据同步(2)-木庄网络博客

6.将数据库变更整理成自己的数据格式发送到kafka等消息队列，供消息队列的消费者进行消费。

工作原理

canal-php 是 Canal 的 php 客户端，它与 Canal 是采用的Socket来进行通信的，传输协议是TCP，交互协议采用的是 Google Protocol Buffer 3.0。

工作流程

使用组件安装，此处我有一个laravel框架，直接在laravel里面安装使用了，相关代码贴出

# 安装组件canal-php
 composer require xingwenge/canal_php
# 编写脚本监听
<?php
 
namespace App\\\\Console\\\\Commands;
use xingwenge\\\\canal_php\\\\CanalClient;
use xingwenge\\\\canal_php\\\\CanalConnectorFactory;
use xingwenge\\\\canal_php\\\\Fmt;
use Illuminate\\\\Console\\\\Command;
 
ini_set('display_errors', 'On');
error_reporting(E_ALL);
class CanalDemo extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'CanalDemo';
 
    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '测试canal';
 
    /**
     * Create a new command instance.
     *
     * @return void
     */
    public function __construct()
    {
        parent::__construct();
    }
 
    /**
     * Execute the console command.
     *
     * @return mixed
     */
    public function handle()
    {
        try {
            $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SOCKET_CLUE);
            # $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SWOOLE);
 
            $client->connect("172.17.0.5", 11111);//此处修改为自己的配置
            $client->checkValid();
            $client->subscribe("1001", "test", ".*\\\\\\\\..*");//对应启动容器时test的队列名
            while (true) {
                $message = $client->get(100);
                if ($entries = $message->getEntries()) {
                    foreach ($entries as $entry) {
                        Fmt::println($entry);
                    }
                }
                sleep(1);
            }
 
            $client->disConnect();
        } catch (\\\\Exception $e) {
            echo $e->getMessage(), PHP_EOL;
        }
 
    }
 
}复制代码