消息队列RocketMQ

RocketMQ连接器支持的信息如下。

类别	详情
支持类型	源表和结果表
运行模式	仅支持流模式
数据格式	CSV和二进制格式
特有监控指标	源表 numRecordsIn numRecordsInPerSecond numBytesIn numBytesInPerScond currentEmitEventTimeLag currentFetchEventTimeLag sourceIdleTime 结果表 numRecordsOut numRecordsOutPerSecond numBytesOut numBytesOutPerSecond currentSendTime 说明 指标的含义及如何查看监控指标，请参见查看监控指标。
API种类	Datastream和SQL
是否支持更新或删除结果表数据	不支持更新和删除结果表数据，只支持插入数据。

RocketMQ源表和结果表支持属性字段，具体如下。

• 源表属性字段
  说明 仅在VVR 3.0.1及以上版本支持获取以下RocketMQ属性字段。

  字段名	字段类型	说明
  topic	VARCHAR METADATA VIRTUAL	RocketMQ消息Topic。
  queue-id	INT METADATA VIRTUAL	RocketMQ消息队列ID。
  queue-offset	BIGINT METADATA VIRTUAL	RocketMQ消息队列的消费位点。
  msg-id	VARCHAR METADATA VIRTUAL	RocketMQ消息ID。
  store-timestamp	TIMESTAMP(3) METADATA VIRTUAL	RocketMQ消息存储时间。
  born-timestamp	TIMESTAMP(3) METADATA VIRTUAL	RocketMQ消息生成时间。
  keys	VARCHAR METADATA VIRTUAL	RocketMQ消息Keys。
  tags	VARCHAR METADATA VIRTUAL	RocketMQ消息Tags。

• 结果表属性字段
  说明 仅实时计算引擎VVR 4.0.0及以上版本支持以下RocketMQ属性字段。

  字段名	字段类型	说明
  keys	VARCHAR METADATA	RocketMQ消息Keys。
  tags	VARCHAR METADATA	RocketMQ消息Tags。

实时计算引擎VVR提供SourceFunction的实现类MetaQSourceFunction，用于读取RocketMQ；提供OutputFormat的实现类MetaQOutputFormat，用于写入RocketMQ。读取RocketMQ和写入RocketMQ的示例如下。

/**
 * A {@link DataStream} demo that illustrates how to consume messages from RocketMQ, convert
 * messages, then produce messages to RocketMQ.
 *
 * <pre>
 * Arguments
 * mqSourceTopic: The consumer topic of the RocketMQ source.
 * mqSourceConsumerGroup: The consumer group of the RocketMQ source.
 * mqSourceEndpoint: The endpoint address of consumer topic for the RocketMQ source.
 * mqSourceAccessId: The access id of consumer topic for the RocketMQ source.
 * mqSourceAccessKey: The access key of consumer topic for the RocketMQ source.
 * mqSourceInstanceId: The instance id of consumer topic for the RocketMQ source.
 * startMessageOffset: The starting offset of message consumption for the RocketMQ source.
 * startTime: The starting time of message consumption for the RocketMQ source.
 * mqSinkTopic: The producer topic of the RocketMQ sink.
 * mqSinkProducerGroup: The producer group of the RocketMQ sink.
 * mqSinkEndpoint: The endpoint address of producer topic for the RocketMQ sink.
 * mqSinkAccessId: The access id of producer topic for the RocketMQ sink.
 * mqSinkAccessKey: The access key of producer topic for the RocketMQ sink.
 * mqSinkInstanceId: The instance id of producer topic for the RocketMQ sink.
 * </pre>
 */
public class RocketMQDataStreamDemo {

    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMQDataStreamDemo.class);

    public static void main(String[] args) throws Exception {
        // Sets up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        final ParameterTool parameters = ParameterTool.fromArgs(args);
        // Creates and adds RocketMQ source.
        env.addSource(createRocketMQSource(parameters))
                // Converts message body to upper case.
                .map(RocketMQDataStreamDemo::convertMessages)
                // Creates and adds RocketMQ sink.
                .addSink(new OutputFormatSinkFunction<>(createRocketMQOutputFormat(parameters)))
                .name(RocketMQDataStreamDemo.class.getSimpleName());
        // Compiles and submits job.
        env.execute("RocketMQ connector end-to-end DataStream demo");
    }

    private static MetaQSourceFunction createRocketMQSource(ParameterTool parameters) {
        String sourceTopic = parameters.get("mqSourceTopic");
        String consumerGroup = parameters.get("mqSourceConsumerGroup");
        Properties mqProperties = createSourceMQProperties(parameters);
        int partitionCount = 0;
        MetaPullConsumer consumer = null;
        try {
            consumer = createConsumerInstance(sourceTopic, consumerGroup, mqProperties, null, -1);
            Set<MessageQueue> queues = consumer.fetchSubscribeMessageQueues(sourceTopic);
            partitionCount = queues == null ? 0 : queues.size();
        } catch (MQClientException e) {
            LOGGER.error(
                    "Fetches RocketMQ partition count for RocketMQ source exception [{}].",
                    e.getMessage());
        } finally {
            if (consumer != null) {
                try {
                    MetaQConnect.shutdownConsumer(consumer);
                } catch (Exception ignored) {
                }
            }
        }
        return new MetaQSourceFunction(
                sourceTopic,
                consumerGroup,
                null,
                null,
                100,
                partitionCount,
                Long.MAX_VALUE,
                Long.parseLong(parameters.get("startMessageOffset")),
                Long.parseLong(parameters.get("startTime")),
                mqProperties);
    }

    private static MetaQOutputFormat createRocketMQOutputFormat(ParameterTool parameters) {
        return new MetaQOutputFormat.Builder()
                .setTopicName(parameters.get("mqSinkTopic"))
                .setProducerGroup(parameters.get("mqSinkProducerGroup"))
                .setMqProperties(createSinkMQProperties(parameters))
                .build();
    }

    private static Properties createSourceMQProperties(ParameterTool parameters) {
        Properties properties = new Properties();
        properties.put(PROPERTY_ONS_CHANNEL, "ALIYUN");
        properties.put(NAMESRV_ADDR, parameters.get("mqSourceEndpoint"));
        properties.put(PROPERTY_ACCESSKEY, parameters.get("mqSourceAccessId"));
        properties.put(PROPERTY_SECRETKEY, parameters.get("mqSourceAccessKey"));
        properties.put(PROPERTY_ROCKET_AUTH_ENABLED, true);
        properties.put(PROPERTY_INSTANCE_ID, parameters.get("mqSourceInstanceId"));
        return properties;
    }

    private static Properties createSinkMQProperties(ParameterTool parameters) {
        Properties properties = new Properties();
        properties.put(PROPERTY_ONS_CHANNEL, "ALIYUN");
        properties.put(NAMESRV_ADDR, parameters.get("mqSinkEndpoint"));
        properties.put(PROPERTY_ACCESSKEY, parameters.get("mqSinkAccessId"));
        properties.put(PROPERTY_SECRETKEY, parameters.get("mqSinkAccessKey"));
        properties.put(PROPERTY_ROCKET_AUTH_ENABLED, true);
        properties.put(PROPERTY_INSTANCE_ID, parameters.get("mqSinkInstanceId"));
        return properties;
    }

    private static List<MessageExt> convertMessages(List<MessageExt> messages) {
        messages.forEach(
                message -> message.setBody(new String(message.getBody()).toUpperCase().getBytes()));
        return messages;
    }
}