创建RocketMQ Sink Connector_云消息队列 Kafka 版(Kafka)-阿里云帮助中心

本文介绍如何在云消息队列 Kafka 版控制台创建消息流出任务，将云消息队列 Kafka 版的数据导出至云消息队列 RocketMQ 版。

前提条件

在创建任务面板，设置任务名称和描述，配置以下参数，单击保存。

任务创建

在Source（源）配置向导，选择数据提供方为消息队列 Kafka 版，设置以下参数，然后单击下一步。

参数	说明	示例
地域	选择云消息队列 Kafka 版源实例所在的地域。	华北2（北京）
kafka 实例	选择生产云消息队列 Kafka 版消息的源实例。	MQ_INST_115964845466****_ByBeUp3p
Topic	选择生产云消息队列 Kafka 版消息的Topic。	topic
Group ID	选择源实例的消费组名称。请使用独立的消费组来创建事件源，不要和已有的业务混用消费组，以免影响已有的消息收发。	GID_http_1
消费位点	选择开始消费消息的位点。	最新位点
网络配置	选择路由消息的网络类型。	默认网络
专有网络VPC	选择VPC ID。当网络配置设置为自建公网时需要设置此参数。	vpc-bp17fapfdj0dwzjkd****
交换机	选择vSwitch ID。当网络配置设置为自建公网时需要设置此参数。	vsw-bp1gbjhj53hdjdkg****
安全组	选择安全组。当网络配置设置为自建公网时需要设置此参数。	alikafka_pre-cn-7mz2****
批量推送条数	调用函数发送的最大批量消息条数，当积压的消息数量到达设定值时才会发送请求，取值范围为 [1,10000]。	100
批量推送间隔（单位：秒）	调用函数的间隔时间，系统每到间隔时间点会将消息聚合后发给函数计算，取值范围为[0,15]，单位为秒。0秒表示无等待时间，直接投递。	3

在Sink（目标）配置向导，选择服务类型为消息队列 RocketMQ 版，配置以下参数。

参数	说明	示例
版本	选择云消息队列 RocketMQ 版实例版本。	RocketMQ 4.x
实例ID	选择已创建的云消息队列 RocketMQ 版实例。	test
Topic	选择已创建的Topic。	test
消息体（body）	事件总线EventBridge通过JSONPath提取事件中的数据，将指定的事件内容路由到事件目标。	部分事件 `$.data.body`
自定义属性（Properties）	选择模板。您可以自定义一个模板，定义模板里需要的变量，事件总线EventBridge可以提取事件中的字段，按照模板定义的形式进行转换。说明如果需要全量传递源端的RocketMQ消息的属性，推荐使用示例中的配置。	变量： `{ "userProperties":"$.data.userProperties", "msgId":"$.data.systemProperties.UNIQ_KEY" }` 模板： `{ "EB_SYS_EMBED_OBJECT":"${userProperties}", "UNIQ_KEY":"${msgId}" }`
消息索引（Keys）	事件总线EventBridge通过JSONPath提取事件中的数据，将指定的事件内容路由到事件目标。	部分事件 `$.data.systemProperties.KEYS`
消息标签（Tags）	事件总线EventBridge通过JSONPath提取事件中的数据，将指定的事件内容路由到事件目标。	部分事件 `$.data.systemProperties.TAGS`

在任务列表页面，找到目标任务，在其右侧操作列，执行其他操作。