文档中心

通过 RocketMQ 消费 Canal Proto 格式的订阅数据

最近更新时间：2023.10.08 11:45:20

首次发布时间：2023.09.01 10:59:59

数据库传输服务 DTS 的数据订阅服务支持使用 RocketMQ 客户端消费 Canal Proto 格式的订阅数据。本文以订阅云数据库 MySQL 版实例为例，介绍如何使用 Go 和 Java 语言消费 Canal Proto 格式的数据。

前提条件

已注册火山引擎账号并完成实名认证。详细操作，请参见如何进行账号注册和实名认证。
按需安装运行语言环境。

运行环境 说明

运行环境	说明
Go 语言	安装 Go，需使用 Go 1.13 或以上版本。说明您可以执行 `go -version` 查看 Go 的版本。
Java 语言	安装 Java，需使用 Java 1.8 或以上版本。您可以执行 `java --version` 查看 Java 版本。安装 Maven，需使用 Maven 3.8 或以上版本。在 `maven pom.xml` 文件中添加以下依赖。 `<dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java</artifactId> <version>3.16.0</version> </dependency> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.9.4</version> </dependency> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-acl</artifactId> <version>4.9.4</version> </dependency>`

Go 语言

安装 Go，需使用 Go 1.13 或以上版本。

说明

您可以执行 go -version 查看 Go 的版本。

Java 语言

安装 Java，需使用 Java 1.8 或以上版本。您可以执行 java --version 查看 Java 版本。
安装 Maven，需使用 Maven 3.8 或以上版本。

在 maven pom.xml 文件中添加以下依赖。

<dependency>
   <groupId>com.google.protobuf</groupId>
   <artifactId>protobuf-java</artifactId>
   <version>3.16.0</version>
</dependency>
<dependency>
   <groupId>org.apache.rocketmq</groupId>
   <artifactId>rocketmq-client</artifactId>
   <version>4.9.4</version>
</dependency>
<dependency>
   <groupId>org.apache.rocketmq</groupId>
   <artifactId>rocketmq-acl</artifactId>
   <version>4.9.4</version>
</dependency>

操作步骤

下载和编译 ProtoBuf

在运行对应语言的 demo 时，需要先根据以下操作步骤完成 Protocol Buffers（也称 ProtoBuf）文件的下载及编译：

说明

本文以火山引擎定义的 ProtoBuf 为例。

下载 ProtoBuf 文件。
将 ProtoBuf 文件编译成对应语言的代码。编译方法如下所示：
- Go 语言
  由于数据库传输服务的开发人员已经帮助您将 ProtoBuf 文件编译成 Go 语言的代码，您无需再编译。关于编译的更多信息，请参见 Protocol Buffer Basics: Go。
- Java 语言
  当您的语言是 Java 时，请执行以下命令将 ProtoBuf 文件编译成 Java 语言的代码，获取 canal.Canal 文件。关于编译的更多信息，请参见 Protocol Buffer Basics: Java。
```
protoc -I=. --java_out=. volc.proto
```

关联 RocketMQ 和订阅任务

本文以 macOS 操作系统为例，介绍如何关联 RocketMQ 和订阅任务。

登录 DTS 控制台，创建并配置数据订阅通道。详细信息，请参见订阅方案概览。
在目标数据订阅通道中新增消费组。详细信息，请参见新建消费组。

编辑 .zshrc 文件，配置以下环境变量信息，并完成认证，即可调用 SDK 来消费消息数据。

参数	说明	示例值
`GROUP`	消费组名称。	`285fef6b91754d0bbaab32e4976c****:test_dtssdk`
`ACCESSKEY`	RocketMQ 用户名。	`7ZrfJmOpVks5wvurbUgr****`
`SECRETKEY`	RocketMQ 用户密码。	`q70s5SrCgsePZPh3fb0d****`
`TOPIC`	目标 DTS 数据订阅通道的 Topic。	`d73e98e7fa9340faa3a0d4ccfa10****`
`NAMESRV_ADDR`	目标 DTS 数据订阅通道的私网地址。	`http://rocketmq-cndv83ef3******.rocketmq.ivolces.com:9876`

说明

关于以上表格的详细信息，请参见查看 RocketMQ 实例详情。

按需选择以下 demo 示例。

Java

package main
	
	import (
	   "context"
	   "datasubscription/proto/canal"
	   "fmt"
	   "os"
	   "strings"
	   "time"
	
	   "github.com/apache/rocketmq-client-go/v2"
	   "github.com/apache/rocketmq-client-go/v2/consumer"
	   "github.com/apache/rocketmq-client-go/v2/primitive"
	   "github.com/sirupsen/logrus"
	   protobuf "google.golang.org/protobuf/proto"
	)
	
	type Config struct {
	   accessKey   string
	   secretKey   string
	   topic       string
	   group       string
	   namesrvAddr string
	}
	
	var c Config
	
	func init() {
	   c.namesrvAddr = os.Getenv("NAMESRV_ADDR")
	   c.topic = os.Getenv("TOPIC")
	   c.group = os.Getenv("GROUP")
	   c.accessKey = os.Getenv("ACCESSKEY")
	   c.secretKey = os.Getenv("SECRETKEY")
	}
	
	func main() {
	   logrus.Infof("config: %+v", c)
	
	   cli, _ := rocketmq.NewPushConsumer(
	      consumer.WithGroupName(c.group),
	      consumer.WithNsResolver(primitive.NewPassthroughResolver(strings.Split(c.namesrvAddr, ","))),
	      consumer.WithConsumerModel(consumer.Clustering),
	      consumer.WithConsumerOrder(true),
	      consumer.WithCredentials(primitive.Credentials{
	         AccessKey: c.accessKey,
	         SecretKey: c.secretKey,
	      }),
	   )
	
	   err := cli.Subscribe(c.topic, consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
	      for _, msg := range msgs {
	         handleCanalMsg(msg)
	      }
	      return consumer.ConsumeSuccess, nil
	   })
	   if err != nil {
	      fmt.Println(err.Error())
	   }
	
	   err = cli.Start()
	   if err != nil {
	      fmt.Println(err.Error())
	      os.Exit(-1)
	   }
	
	   time.Sleep(time.Hour)
	   cli.Shutdown()
	}
	
	func handleCanalMsg(msg *primitive.MessageExt) {
	   entry := &canal.Entry{}
	   if err := protobuf.Unmarshal(msg.Body, entry); err != nil {
	      panic(err)
	   }
	
	   logrus.WithField("EntryType", entry.GetEntryType().String()).Info("get message")
	   if entry.GetEntryType() != canal.EntryType_TRANSACTIONBEGIN &&
	      entry.GetEntryType() != canal.EntryType_TRANSACTIONEND {
	      rowChange := &canal.RowChange{}
	      if err := protobuf.Unmarshal(entry.GetStoreValue(), rowChange); err != nil {
	         panic(err)
	      }
	
	      if rowChange.GetIsDdl() {
	         logrus.Infof("ddl: %s", rowChange.GetSql())
	      } else {
	         for _, row := range rowChange.GetRowDatas() {
	            var before, after []string
	            for _, col := range row.BeforeColumns {
	               before = append(before, fmt.Sprintf("%s[%s(%s)]", col.GetName(), col.GetValue(), col.GetMysqlType()))
	            }
	            for _, col := range row.AfterColumns {
	               after = append(after, fmt.Sprintf("%s[%s(%s)]", col.GetName(), col.GetValue(), col.GetMysqlType()))
	            }
	            logrus.WithField("after", after).WithField("before", before).Info("get row")
	         }
	      }
	
	      logrus.WithField("queueId", msg.Queue.QueueId).WithField("key", msg.GetKeys()).Info("fetch message")
	   }
	}

package com.bytedance.datasubscription; 
	
	import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
	import org.apache.rocketmq.client.consumer.listener.*;
	import org.apache.rocketmq.common.message.MessageExt;
	import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
	import org.apache.rocketmq.remoting.RPCHook;
	import org.apache.rocketmq.acl.common.AclClientRPCHook;
	import org.apache.rocketmq.acl.common.SessionCredentials;
	import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
	import com.google.protobuf.InvalidProtocolBufferException;
	import java.util.List;
	
	import canal.Canal;
	import static canal.Canal.EntryType.ROWDATA;
	
	
	public class App {
	      private static final String NAMESRV_ADDR = System.getenv("NAMESRV_ADDR");
	      private static final String TOPIC = System.getenv("TOPIC");
	      private static final String GROUP = System.getenv("GROUP");
	      private static final String ACL_ACCESS_KEY = System.getenv("ACCESSKEY");
	      private static final String ACL_SECRET_KEY = System.getenv("SECRETKEY");
	
	      public static void main(String[] args) {
	            System.out.printf("[CONFIG] namesrvAddr: %s, topic: %s, group: %s, accessKey: %s, secret: %s\n",
	                  NAMESRV_ADDR, TOPIC, GROUP, ACL_ACCESS_KEY, ACL_SECRET_KEY);
	
	            try {
	                DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP, getAclRPCHook(), new AllocateMessageQueueAveragely());
	                consumer.setNamesrvAddr(NAMESRV_ADDR);
	                consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
	                consumer.subscribe(TOPIC, "*");
	                consumer.registerMessageListener(new MessageListenerConcurrently() {
	                      @Override
	                      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
	                            for (MessageExt msg : msgs) {
	                                  try {
	                                     Canal.Entry entry = Canal.Entry.parseFrom(msg.getBody());
	                                     System.out.println(String.format("sourceType:%s\nentryType:%s\nexecuteTime:%s\nschemaName:%s" +
	                                                 "\ntableName:%s", entry.getHeader().getSourceType(), entry.getEntryType(),
	                                           entry.getHeader().getExecuteTime(), entry.getHeader().getSchemaName(),
	                                           entry.getHeader().getTableName()));
	                                     if (entry.getEntryType() ==  ROWDATA) {
	                                        Canal.RowChange event = Canal.RowChange.parseFrom(entry.getStoreValue());
	                                        System.out.println(event);
	                                     }
	                                    } catch (InvalidProtocolBufferException e) {
	                                          e.printStackTrace();
	                                    }
	                              }
	                              return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	                        }
	                  });
	
	                  consumer.start();
	                  System.out.println("Consumer Started.");
	            } catch (Exception ex) {
	                  System.out.println(ex.getMessage());
	                  System.exit(-1);
	            }
	      }
	
	      static RPCHook getAclRPCHook() {
	            return new AclClientRPCHook(new SessionCredentials(ACL_ACCESS_KEY, ACL_SECRET_KEY));
	      }
	}

运行测试

说明

本文以数据库 test，表格 demo 为例。
根据目标语言选择合适的 JSON 数据。

在源数据库中，执行以下命令创建一张名为 demo 的表。

CREATE TABLE demo (id_t INT);

预期输出：

Java

INFO[0009] get message                                   EntryType=ROWDATA
		INFO[0009] ddl: CREATE TABLE demo (id_t INT) 
		INFO[0009] fetch message                                 key=test.demo queueId=1

执行以下命令，在 demo 表中插入一条数据：

INSERT INTO demo (id_t) VALUES (1);

预期输出：

Java

INFO[0101] get message                                   EntryType=ROWDATA
	INFO[0101] get row                                       after="[id_t[1(int)]]" before="[]"
	INFO[0101] fetch message                                 key=test.demo queueId=1

sourceType:MYSQL
	entryType:ROWDATA
	executeTime:1693208039
	schemaName:test
	tableName:demo
	eventType: INSERT
	isDdl: false
	rowDatas {
	  afterColumns {
	    index: 1
	    name: "id_t"
	    updated: true
	    isNull: false
	    value: "1"
	    mysqlType: "int"
	  }
	}

前提条件

操作步骤

下载和编译 ProtoBuf

关联 RocketMQ 和订阅任务

运行测试