Laravel配合Canal实现Elasticsearch与mysql的增量同步

此方法只能基础实现或者测试canal是否能够监听到不适用于生产环境，如使用守护进程canal与php消费者在生产环境意外重启一旦数据有变动php消费者没有启动canal会占用大量内存堆满数据，并且守护进程在意外重启后可能存在反复启动canal。如要使用canal应先推入队列再进行消费。但目前看来canal要与业务以及es、mysql在同一台服务器上运行是不可能的，不使用守护进程canal不会自动重启但会在重启后残留一个进程。目前canal放一个单独服务器了增量同步还是要找一些其他方法。现在去研究一下Apache SeaTunnel。

对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant

CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

下载canal(这里下载最新1.1.7)

wget https://github.com/alibaba/canal/releases/download/canal-1.0.17/canal.deployer-1.1.17.tar.gz

解压

mkdir /tmp/canal
tar zxvf canal.deployer-$version.tar.gz  -C /tmp/canal

修改配置

vi conf/example/instance.properties

#################################################
## mysql serverId , v1.0.26+ will autoGen
canal.instance.mysql.slaveId=2

# enable gtid use true/false
canal.instance.gtidon=false

# position info
canal.instance.master.address=你的服务器地址:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=

# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=

# table meta tsdb info
canal.instance.tsdb.enable=true
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
canal.instance.tsdb.dbUsername=你的数据库账号
canal.instance.tsdb.dbPassword=你的数据库密码

#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=

# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

# table regex
canal.instance.filter.regex=ity.orders
# table black regex
canal.instance.filter.black.regex=mysql\\.slave_.*
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch

# mq config
#canal.mq.servers = 127.0.0.1:6379 # Redis服务器地址
#canal.mq.canalGetTimeout = 100 # 拉取消息的超时时间（毫秒）
#canal.mq.flatMessage = true # 使用FlatMessage格式
#canal.instance.global.spring.xml = classpath:spring/default-instance.xml # 使用默认Spring配置
#canal.mq.topic=example
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,topic2:mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.enableDynamicQueuePartition=false
#canal.mq.partitionsNum=3
#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#
# multi stream for polardbx
canal.instance.multi.stream.on=false
#################################################

启动Canal服务

因为ES所用java是最新版本JDK所以有些参数弃用了，需要修改配置文件

替换 /tmp/canal/bin/startup.sh中的
JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"
为
JAVA_OPTS="$JAVA_OPTS -Xss1m -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"

进canal目录执行sh bin/startup.sh命令启动

在/tmp/canal下创建个example目录然后下载canal.example用于测试是否监控到binlog

wget https://github.com/alibaba/canal/releases/download/canal-1.0.17/canal.example-1.1.17.tar.gz

不用任何配置直接启动sh example/bin/startup.sh

启动后查看example.log即可

使用canal-php组建消费者

创建SyncOrdersToES命令

php artisan make:command SyncOrdersToES

编写链接以及监听后的操作

app\console\comands\SyncOrdersToES.php

<?php

namespace App\Console\Commands;

use Com\Alibaba\Otter\Canal\Protocol\EntryType;
use Com\Alibaba\Otter\Canal\Protocol\EventType;
use Com\Alibaba\Otter\Canal\Protocol\RowChange;
use Elastic\Elasticsearch\Client;
use Elastic\Elasticsearch\ClientBuilder;
use Illuminate\Console\Command;
use Predis\Client as RedisClient;
use xingwenge\canal_php\CanalClient;
use xingwenge\canal_php\CanalConnectorFactory;

class SyncOrdersToES extends Command
{
    protected $signature = 'sync:orders-to-es';
    protected $description = 'Sync orders from MySQL to Elasticsearch using Canal data from Redis.';

    protected $esClient;

    public function __construct(Client $esClient)
    {
        parent::__construct();
        $this->esClient = $esClient;

    }

    public function handle()
    {
        $connector = CanalConnectorFactory::createClient(CanalClient::TYPE_SOCKET_CLUE);
        $connector->connect('127.0.0.1', 11111); // Canal Server 的地址和端口
        $connector->subscribe('1001', 'example', '.*\\..*'); // 客户端ID, Canal 实例名称, 订阅的数据库和表

        while (true) {
            $message = $connector->get(100); // 从 Canal 获取数据变更
            if ($entries = $message->getEntries()) {
                foreach ($entries as $entry) {
                    if ($entry->getEntryType() == EntryType::ROWDATA) {
                        $rowChange = new RowChange();
                        $rowChange->mergeFromString($entry->getStoreValue());
                        $tableName = $entry->getHeader()->getTableName();
                        if ($tableName == 'orders') { // 确保是 orders 表的变更
                            foreach ($rowChange->getRowDatas() as $rowData) {
                                // 根据事件类型，决定是使用 getBeforeColumns 还是 getAfterColumns
                                $columns = ($rowChange->getEventType() == EventType::DELETE) ?
                                    $rowData->getBeforeColumns() :
                                    $rowData->getAfterColumns();

                                $change = [
                                    'type' => $rowChange->getEventType(),
                                    'data' => $this->parseColumns($columns),
                                ];
                                // 打印变动数据
                                $this->info("Detected change in orders: " . json_encode($change));
                                $this->syncChangeToElasticsearch($change, $this->esClient);
                            }
                        }
                    }
                }
            }
            sleep(1); // 简单的流控制
        }
    }

    protected function syncChangeToElasticsearch($change, $esClient)
    {
        $indexParams = [
            'index' => 'orders',
            'id'    => $change['data']['id'],
            'body' => [
                'id'               => $change['data']['id'],
                'order_no'         => $change['data']['order_no'],
                'open_status'      => $change['data']['open_status'],
                'open_message'     => $change['data']['open_message'],
                'id_card_name'     => $change['data']['id_card_name'],
                'id_card'          => $change['data']['id_card'],
                'phonenum'         => $change['data']['phonenum'],
                'contact'          => $change['data']['contact'],
                'mobile'           => $change['data']['mobile'],
                'express_address'  => $change['data']['express_address'],
                'express_no'       => $change['data']['express_no'],
                'express_name'     => $change['data']['express_name'],
                'idcard_front'     => $change['data']['idcard_front'],
                'idcard_back'      => $change['data']['idcard_back'],
                'user_with_idcard' => $change['data']['user_with_idcard'],
                'pay_num'          => $change['data']['pay_num'],
                'express_status'   => $change['data']['express_status'],
                'status'           => $change['data']['status'],
                'hkgj_package_id'  => $change['data']['hkgj_package_id'],
                'package_id'       => $change['data']['package_id'],
                'bus_order_id'     => $change['data']['bus_order_id'],
                'package_name'     => $change['data']['package_name'],
                'supplier_id'      => $change['data']['supplier_id'],
                'fxuser_id'        => $change['data']['fxuser_id'],
                'fx2_id'           => $change['data']['fx2_id'],
                'tenant_id'        => $change['data']['tenant_id'],
                'anchor_id'        => $change['data']['anchor_id'],
                'pay_id'           => $change['data']['pay_id'],
                'live_room_id'     => $change['data']['live_room_id'],
                'created_at'       => date('c', strtotime($change['data']['created_at'])),
                'updated_at'       => date('c', strtotime($change['data']['updated_at'])),
                'deleted_at'       => date('c', strtotime($change['data']['deleted_at'])),
                'api_order_no'     => $change['data']['api_order_no'],
                'api_lock'         => $change['data']['api_lock'],
                'api_valid'        => $change['data']['api_valid'],
                'api_sync'         => $change['data']['api_sync'],
                'hk_sync'          => $change['data']['hk_sync'],
                'remark'           => $change['data']['remark'],
                'cid'              => $change['data']['cid'],
                'open_id'          => $change['data']['open_id'],
                'channe_id'        => $change['data']['channe_id'],
                'admin_user_id'    => $change['data']['admin_user_id'],
                'pay_status'       => $change['data']['pay_status'],
                'refund_text'      => $change['data']['refund_text'],
                'sms_status'       => $change['data']['sms_status'],
                'intercept_status' => $change['data']['intercept_status'],
                'intercept_msg'    => $change['data']['intercept_msg'],
                'nick'             => $change['data']['nick'],
                'buyer_openid'     => $change['data']['buyer_openid'],
                'author_id'        => $change['data']['author_id'],
                'author_name'      => $change['data']['author_name'],
                'dy_pacage_code'   => $change['data']['dy_pacage_code'],
                'dy_product_id'    => $change['data']['dy_product_id'],
                // 添加更多字段...
            ],
        ];

        switch ($change['type']) {
            case EventType::INSERT:
            case EventType::UPDATE:
                // 插入或更新事件
                $response = $esClient->index($indexParams);
                break;
            case EventType::DELETE:
                // 删除事件
                $esClient->delete([
                    'index' => 'orders',
                    'id'    => $change['data']['id'], // 假设 'id' 是文档ID
                ]);
                break;
        }
    }
    protected function parseColumns($columns)
    {
        $parsedData = [];
        foreach ($columns as $column) {
            // 假设 Column 类有 getName() 和 getValue() 方法来获取列名和列值
            $parsedData[$column->getName()] = $column->getValue();
        }
        return $parsedData;
    }

}

启动同步命令php artisan sync:orders-to-es

Working

GitHub

artisan Canal CentOS Dcat-admin dcatadmin elasticsearch ES git GO JWT Laravel Laravel9.x Laravel注意事项 Linux mysql mysql优化 php php8(match) redis Tinker ubuntu wechat 业务与流程个人简介交互设计产品思维产品经理产品设计产品运营原型及需求文档原子锁宝塔导出心理学文件打包下载爬虫用户体验秒杀设计设计模式访问量量化交易阿德勒

Laravel配合Canal实现Elasticsearch与mysql的增量同步

Working

标签