ApsaraMQ for RocketMQ:Canal を使用して MySQL の変更データを ApsaraMQ for RocketMQ にストリーミングする

バージョンの互換性

バイナリロギングの有効化

ApsaraDB RDS for MySQL

バイナリロギングはデフォルトで有効になっており、ご利用の Alibaba Cloud アカウントには必要なダンプ権限が自動的に付与されています。テストデータベースの作成 に進んでください。

セルフマネージド MySQL

1. MySQL の設定ファイル (my.cnf) を編集して、ROW フォーマットのバイナリロギングを有効にします。

       [mysqld]
       log-bin=mysql-bin   # バイナリロギングを有効化
       binlog-format=ROW   # Canal に必須のフォーマット
       server_id=1         # Canal のセカンダリノード ID とは異なる ID にする必要があります

2. Canal 専用の MySQL ユーザーを作成し、レプリケーション権限を付与します。

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

テストデータベースの作成

CREATE DATABASE canal DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;

テストテーブルの作成

canal データベースに students テーブルを作成します。

CREATE TABLE students (
    id INT AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL,
    age INT,
    gender VARCHAR(10),
    PRIMARY KEY (id)
);

JDK のインストール

sudo yum install java-1.8.0-openjdk

Canal Deployer のダウンロードと展開

# Canal Deployer 1.1.6 をダウンロード
sudo wget https://github.com/alibaba/canal/releases/download/canal-1.1.6/canal.deployer-1.1.6.tar.gz

# インストールディレクトリを作成して展開
sudo mkdir -p /usr/local/canal-server
sudo tar -zxvf canal.deployer-1.1.6.tar.gz -C /usr/local/canal-server

RocketMQ 向け Canal の設定

Canal の動作は 2 つの設定ファイルで制御されます：グローバル設定用の canal.properties と、インスタンスごとの設定用の instance.properties です。

sudo vi /usr/local/canal-server/conf/canal.properties

# 出力先：変更イベントを RocketMQ に送信
canal.serverMode = rocketMQ

# Alibaba Cloud 認証
canal.aliyun.accessKey = <your-access-key-id>
canal.aliyun.secretKey = <your-access-key-secret>

# アクセスチャネル：Alibaba Cloud マネージドの RocketMQ の場合は "cloud" に設定
canal.mq.accessChannel = cloud

# メッセージフォーマット：ApsaraMQ for RocketMQ の場合は false にする必要があります
canal.mq.flatMessage = false

# RocketMQ プロデューサー設定
rocketmq.producer.group = canal_test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = <your-rocketmq-endpoint>
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag =

sudo vi /usr/local/canal-server/conf/example/instance.properties

# MySQL 接続
canal.instance.master.address = <your-mysql-endpoint>:3306
canal.instance.dbUsername = <your-db-username>
canal.instance.dbPassword = <your-db-password>

# テーブルフィルター："canal" スキーマ内のすべてのテーブルをキャプチャ
canal.instance.filter.regex = canal\\..*

# ターゲット RocketMQ トピック
canal.mq.topic = canal_topic

Canal Deployer の起動

/usr/local/canal-server/bin/startup.sh

起動の確認

Canal サーバーログをチェックして、正常に起動したことを確認します。

sudo cat /usr/local/canal-server/logs/canal/canal.log

起動に成功すると、以下のような出力が生成されます。

2024-07-15 17:24:12.154 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## set default uncaught exception handler
2024-07-15 17:24:12.202 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## load canal configurations
2024-07-15 17:24:12.497 [main] INFO  c.a.o.c.c.rocketmq.producer.CanalRocketMQProducer - ##Start RocketMQ producer##
2024-07-15 17:24:12.799 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## start the canal server.
2024-07-15 17:24:12.984 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[172.17.XX.XX:11111]
2024-07-15 17:24:16.208 [main] INFO  com.alibaba.otter.canal.deployer.CanalStarter - ## the canal server is running now ......

最後の the canal server is running now という行は、Canal が正常に起動し、RocketMQ に接続されたことを示します。

次に、インスタンスログをチェックして、Canal が MySQL に接続し、モニタリングを開始したことを確認します。

sudo cat /usr/local/canal-server/logs/example/example.log

接続が成功したことを示す以下の主要な兆候を探してください。

2024-07-15 18:22:15.667 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
2024-07-15 18:22:15.699 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^canal\..*$
2024-07-15 18:22:15.699 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$

インスタンスログにエラーが表示される場合は、以下の一般的な原因を確認してください。

テストデータの挿入

students テーブルに行を挿入して、変更イベントをトリガーします。

INSERT INTO `students` (`name`, `age`, `gender`) VALUES ('Tome', 18, 'male');

ApsaraMQ for RocketMQ でのメッセージの確認

1. ApsaraMQ for RocketMQ コンソールにログインします。

2. 左側のナビゲーションウィンドウで、[インスタンス] をクリックします。

3. [インスタンス] ページで、設定したインスタンスの名前をクリックします。

4. [インスタンス詳細] ページの左側のナビゲーションウィンドウで、[メッセージクエリ] をクリックします。

5. クエリ結果に Canal からのメッセージが表示されることを確認します。

メッセージが表示されない場合は、Canal がバイナリログイベントを処理するまで数秒待ってから、クエリを再試行してください。30 秒経ってもメッセージが表示されない場合は、/usr/local/canal-server/logs/example/example.log にある Canal インスタンスログでエラーを確認してください。

canal.properties パラメーター

instance.properties パラメーター