一、介绍
RocketMQ 支持同一个 Topic 多分区读写，数据可以从多个分区读入，也可以写入到多个分区，以提供更高的吞吐量，减少数据倾斜和热点。
二、版本说明
| Flink 版本 | RocketMQ 4.x | RocketMQ 5.x |
| 1.16 | 支持 | 支持 |
| 1.18 | 支持 | 支持 |
| 1.20 | 支持 | 支持 |
三、使用范围
RocketMQ 支持用作数据源表（Source），也可以作为 Tuple 数据流的目的表（Sink）。
六、RocketMQ 5.x – 基础示例
CREATE TABLE rocketmq_source ( `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING) WITH ( 'connector' = 'rocketmq5', — 固定为 rocketmq5 'rocketmq.client.endpoints' = 'rmq-xxx.rocketmq.gz.qcloud.tencenttdmq.com:8080', — 必选，替换为你的 RocketMQ 5.x 接入点 'rocketmq.source.topic' = 'YourTopic', — 必选，替换为你要消费的 Topic 'rocketmq.source.group' = 'YourConsumerGroup', — 必选，替换为你的消费者分组 'rocketmq.source.startup.scan.mode' = 'earliest', 'rocketmq.client.accessKey' = 'YourAccessKey', — 可选，替换为你的 AccessKey 'rocketmq.client.secretKey' = 'YourSecretKey' — 可选，替换为你的 SecretKey);
七、RocketMQ 4.x – 基础示例
CREATE TABLE rocketmq_source ( `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING) WITH ( 'connector' = 'rocketmq4', — 固定为 rocketmq4 'rocketmq.client.endpoints' = 'http://rocketmq-xxx.rocketmq.gz.qcloud.tencenttdmq.com:9876', — 必选，替换为你的 RocketMQ 4.x 接入点 'rocketmq.source.topic' = 'YourTopic', — 必选，替换为你要消费的 Topic 'rocketmq.source.group' = 'YourConsumerGroup', — 必选，替换为你的消费者分组 'rocketmq.client.accessKey' = 'YourAccessKey', — 可选，替换为你的 AccessKey 'rocketmq.client.secretKey' = 'YourSecretKey' — 可选，替换为你的 SecretKey);
九、RocketMQ 5.x – 基础示例
CREATE TABLE rocketmq_sink ( `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING) WITH ( 'connector' = 'rocketmq5', — 固定为 rocketmq5 'rocketmq.sink.topic' = 'YourTopic', — 必选，替换为你要写入的 Topic 'rocketmq.client.endpoints' = 'rmq-xxx.rocketmq.gz.qcloud.tencenttdmq.com:8080', — 必选，替换为你的 RocketMQ 5.x 接入点 'rocketmq.client.accessKey' = 'YourAccessKey', — 可选，替换为你的 AccessKey 'rocketmq.client.secretKey' = 'YourSecretKey' — 可选，替换为你的 SecretKey);
十、RocketMQ 4.x – 基础示例
CREATE TABLE rocketmq_sink ( `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING) WITH ( 'connector' = 'rocketmq4', — 固定为 rocketmq4 'rocketmq.sink.topic' = 'YourTopic', — 必选，替换为你要写入的 Topic 'rocketmq.client.endpoints' = 'http://rocketmq-xxx.rocketmq.gz.qcloud.tencenttdmq.com:9876', — 必选，替换为你的 RocketMQ 4.x 接入点 'rocketmq.sink.group' = 'YourProducerGroup', — 必选，替换为你的生产者分组 'rocketmq.client.accessKey' = 'YourAccessKey', — 可选，替换为你的 AccessKey 'rocketmq.client.secretKey' = 'YourSecretKey' — 可选，替换为你的 SecretKey);
十二、通用客户端参数（rocketmq.client.*）
| 参数 | 必填 | 默认值 | 描述 |
| rocketmq.client.endpoints | 是 | 无 | RocketMQ 服务地址。RocketMQ 4.x 填 NameServer 地址（如 127.0.0.1:9876），RocketMQ 5.x 填 Proxy 接入点 |
| rocketmq.client.namespace | 否 | 无 | RocketMQ 实例命名空间（多租户场景使用） |
| rocketmq.client.channel | 否 | CLOUD | RocketMQ 接入通道。可选 LOCAL / CLOUD / ALIYUN |
| rocketmq.client.partition.discovery.interval.ms | 否 | 10000 | 从 NameServer / Proxy 拉取路由信息的周期（毫秒） |
| rocketmq.client.timeZone | 否 | 无 | 解析时间字符串所用的时区（如 Asia/Shanghai） |
| rocketmq.client.message.encoding | 否 | UTF-8 | 消息体的字符编码 |
| rocketmq.client.message.field.delimiter | 否 | \\u0001 | 字段分隔符（默认 ASCII 0x01） |
| rocketmq.client.message.line.delimiter | 否 | \\n | 消息内行分隔符 |
| rocketmq.client.message.length.check | 否 | NONE | 字段数不匹配时的处理策略，可选 NONE / SKIP / EXCEPTION / PAD |
| rocketmq.client.accessKey | 否 | 无 | ACL 访问凭据（AccessKey） |
| rocketmq.client.secretKey | 否 | 无 | ACL 访问凭据（SecretKey） |
十三、源表（Source）参数（rocketmq.source.*）
| 参数 | 必填 | 默认值 | 描述 |
| rocketmq.source.topic | 是 | 无 | 订阅的 Topic 名称，可使用 ; 分隔多个 Topic |
| rocketmq.source.group | 是 | 无 | 消费者分组名称（用于 Group 维度的消费位点管理） |
| rocketmq.source.filter.tag | 否 | * | 消息 Tag 过滤条件，RocketMQ 仅支持单 Tag 过滤 |
| rocketmq.source.filter.sql | 否 | 无 | 消息过滤表达式（与 Tag 二选一） |
| rocketmq.source.startup.scan.mode | 否 | latest | 启动消费位点模式：earliest / latest / timestamp / specific_offset / group_offsets |
| rocketmq.source.startup.offset.specific | 否 | -1 | 启动消费位点的具体偏移量。设置此参数即启用 specific_offset 模式，不要同时设置 startup.scan.mode |
| rocketmq.source.startup.offset.timestamp | 否 | -1 | 启动消费位点对应的时间戳（毫秒）。设置此参数即启用 timestamp 模式，不要同时设置 startup.scan.mode |
| rocketmq.source.startup.offset.date | 否 | 无 | 启动消费位点对应的日期字符串（yyyy-MM-dd HH:mm:ss），需配合 rocketmq.client.timeZone。设置此参数即启用 timestamp 模式，不要同时设置 startup.scan.mode |
| rocketmq.source.stop.offset.timestamp | 否 | 无 | 有限流场景下停止消费的时间戳；提供时 Source 进入有界模式 |
| rocketmq.source.allocate.strategy | 否 | hash | 队列分配策略，可选 hash/ broadcast/ average |
| rocketmq.source.pull.batch.size | 否 | 32 | 每次拉取消息的最大条数 |
| rocketmq.source.offset.commit.checkpoint | 否 | true | 是否在 Flink checkpoint 时 commit 位点 |
注意：说明：
注意：启动模式互斥说明：startup.scan.mode 与 startup.offset.specific / startup.offset.timestamp / startup.offset.date 互斥；同时设置会导致作业启动失败。
十四、结果表（Sink）参数（rocketmq.sink.*）
| 参数 | 必填 | 默认值 | 描述 |
| rocketmq.sink.topic | 是 | 无 | 写入的 Topic 名称 |
| rocketmq.sink.group | 否 | PID-flink-producer | 生产者分组 |
| rocketmq.sink.delivery.guarantee | 否 | AT_LEAST_ONCE | 暂时不支持设置 |
| rocketmq.sink.send.retry.times | 否 | 3 | 同步发送的重试次数 |
| rocketmq.sink.send.timeout | 否 | 5000 | 单次 send 调用超时（毫秒） |
| rocketmq.sink.tag | 否 | 无 | 消息 Tag |
十六、源表可读元数据
源表支持的元数据列（在 DDL 中需声明为 METADATA VIRTUAL）：
| 列名 | 数据类型 | 描述 |
| topic | STRING NOT NULL | 消息所属 Topic |
| ingestion-time | TIMESTAMP_LTZ(3) NOT NULL | 消息进入 Flink 引擎的时间 |
| event-time | TIMESTAMP_LTZ(3) NOT NULL | 消息生产时间（Born Timestamp） |
| queue-id | INT NOT NULL | 消息所在队列 ID |
| queue-offset | BIGINT NOT NULL | 消息的消费位点 |
| msg-id | STRING NOT NULL | 消息唯一 ID |
| keys | STRING NOT NULL | 消息的 Keys |
| tags | STRING NOT NULL | 消息的 Tags |
| properties | MAP<STRING, STRING> NOT NULL | 消息自定义属性 |
十七、结果表可写元数据
结果表支持的元数据列：
| 列名 | 数据类型 | 描述 |
| keys | STRING | 消息 Keys，写入 Message 的 Keys 字段 |
| tags | STRING | 消息 Tags，写入 Message 的 Tags 字段 |
十九、SQL：从 RocketMQ 读 + 写入 RocketMQ
— Source：定义 RocketMQ 源表CREATE TABLE rocketmq_source ( `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING) WITH ( 'connector' = 'rocketmq5', — 固定为 rocketmq5 'rocketmq.client.endpoints' = 'rmq-xxx.rocketmq.gz.qcloud.tencenttdmq.com:8080', — 必选，替换为你的 RocketMQ 5.x 接入点 'rocketmq.source.topic' = 'YourTopic', — 必选，替换为你要消费的 Topic 'rocketmq.source.group' = 'YourConsumerGroup', — 必选，替换为你的消费者分组 'rocketmq.source.startup.scan.mode' = 'earliest', — 可选，启动消费位点模式：earliest / latest 'rocketmq.client.accessKey' = 'YourAccessKey', — 可选，替换为你的 AccessKey 'rocketmq.client.secretKey' = 'YourSecretKey' — 可选，替换为你的 SecretKey);– Sink：定义 RocketMQ 结果表CREATE TABLE rocketmq_sink ( `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING) WITH ( 'connector' = 'rocketmq5', — 固定为 rocketmq5 'rocketmq.sink.topic' = 'YourTopic', — 必选，替换为你要写入的 Topic 'rocketmq.client.endpoints' = 'rmq-xxx.rocketmq.gz.qcloud.tencenttdmq.com:8080', — 必选，替换为你的 RocketMQ 5.x 接入点 'rocketmq.client.accessKey' = 'YourAccessKey', — 可选，替换为你的 AccessKey 'rocketmq.client.secretKey' = 'YourSecretKey' — 可选，替换为你的 SecretKey);INSERT INTO rocketmq_sink SELECT * FROM rocketmq_source;
20、二进制格式示例
当 RocketMQ 消息体是 Protobuf / Thrift / Avro / 自定义二进制协议等无法按字符切分的内容时，DDL 中只声明一个 VARBINARY 字段即可。连接器会把整条消息的字节数组原样写入该字段，不做切分；后续可在 Flink 作业里用 UDF 自行解析。
CREATE TABLE rocketmq_source ( msg_body VARBINARY) WITH ( 'connector' = 'rocketmq5', — 必选，固定为 rocketmq5 'rocketmq.client.endpoints' = 'YourEndpoint', — 必选，替换为你的 RocketMQ 5.x 接入点 'rocketmq.source.topic' = 'YourTopic', — 必选，替换为你要消费的 Topic 'rocketmq.source.group' = 'YourConsumerGroup', — 必选，替换为你的消费者分组 'rocketmq.client.accessKey' = 'YourAccessKey', — 可选，替换为你的 AccessKey 'rocketmq.client.secretKey' = 'YourSecretKey' — 可选，替换为你的 SecretKey);
二进制模式下需要注意：
·整张表只能声明一个数据列且类型必须为 VARBINARY；
·字段分隔符（rocketmq.client.message.field.delimiter）和字符编码（rocketmq.client.message.encoding）在二进制模式下不生效。
·若需要在二进制数据中同时保留消息属性（如 keys / tags / properties），可以在 DDL 中追加元数据列（参考「元数据列」章节），二者在解析时互不干扰。
21、类型映射
连接器对常见 Flink 类型与消息体字段的映射关系：
| Flink 字段类型 | RocketMQ 字段类型 |
| BOOLEAN | STRING |
| VARBINARY | VARCHAR |
| TINYINT / SMALLINT / INT / BIGINT | INTEGER / BIGINT |
| FLOAT / DOUBLE | DOUBLE |
| DECIMAL | DECIMAL |