RocketMQ

组件格式

要设置 RocketMQ pub/sub，创建一个类型为 pubsub.rocketmq 的组件。请参阅 pub/sub broker 组件文件 了解 ConsumerID 是如何自动生成的。阅读 如何：发布和订阅指南 了解如何创建和应用 pub/sub 配置。

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: rocketmq-pubsub
spec:
  type: pubsub.rocketmq
  version: v1
  metadata:
    - name: instanceName
      value: dapr-rocketmq-test
    - name: consumerGroup
      value: dapr-rocketmq-test-g-c
    - name: producerGroup 
      value: dapr-rocketmq-test-g-p
    - name: consumerID
      value: channel1
    - name: nameSpace
      value: dapr-test
    - name: nameServer
      value: "127.0.0.1:9876,127.0.0.2:9876"
    - name: retries
      value: 3
    - name: consumerModel
      value: "clustering"
    - name: consumeOrderly
      value: false

规格元数据字段

出于向后兼容的原因，元数据中的以下值仍然支持，尽管不推荐使用。

设置 RocketMQ

请参阅 https://rocketmq.apache.org/docs/quick-start/ 以设置本地 RocketMQ 实例。

每次调用的元数据字段

分区键

在调用 RocketMQ pub/sub 时，可以通过在请求 URL 中使用 metadata 查询参数提供可选的分区键。

您需要在 metadata 中指定 rocketmq-tag 、"rocketmq-key" 、rocketmq-shardingkey 、rocketmq-queue

示例：

curl -X POST http://localhost:3500/v1.0/publish/myRocketMQ/myTopic?metadata.rocketmq-tag=?&metadata.rocketmq-key=?&metadata.rocketmq-shardingkey=key&metadata.rocketmq-queue=1 \
  -H "Content-Type: application/json" \
  -d '{
        "data": {
          "message": "Hi"
        }
      }'

队列选择器

RocketMQ 组件提供了五种队列选择器。RocketMQ 客户端提供以下队列选择器：

• HashQueueSelector
• RandomQueueSelector
• RoundRobinQueueSelector
• ManualQueueSelector

要了解有关这些 RocketMQ 客户端队列选择器的更多信息，请阅读 RocketMQ 文档。

Dapr RocketMQ 组件实现了以下队列选择器：

• DaprQueueSelector

本文重点介绍 DaprQueueSelector 的设计。

DaprQueueSelector

DaprQueueSelector 集成了三个队列选择器：

• HashQueueSelector
• RoundRobinQueueSelector
• ManualQueueSelector

DaprQueueSelector 从请求参数中获取队列 ID。您可以通过运行以下命令设置队列 ID：

http://localhost:3500/v1.0/publish/myRocketMQ/myTopic?metadata.rocketmq-queue=1

ManualQueueSelector 是通过上述方法实现的。

接下来，DaprQueueSelector 尝试：

• 获取 ShardingKey
• 哈希 ShardingKey 以确定队列 ID。

您可以通过以下方式设置 ShardingKey：

http://localhost:3500/v1.0/publish/myRocketMQ/myTopic?metadata.rocketmq-shardingkey=key

如果 ShardingKey 不存在，则使用 RoundRobin 算法确定队列 ID。

相关链接

• Dapr 组件的基本架构
• Pub/Sub 构建块
• 阅读本指南以获取有关配置 pub/sub 组件的说明