The documentation you are viewing is for Dapr v1.12 which is an older version of Dapr. For up-to-date documentation, see the latest version.

Pulsar

关于Pulsar pubsub组件的详细文档

配置

要设置 Apache Pulsar pubsub,请创建一个类型为 pubsub.pulsar的组件。 请参阅本指南,了解如何创建和应用 pubsub 配置。 For more information on Apache Pulsar read the docs

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: pulsar-pubsub
  namespace: default
spec:
  type: pubsub.pulsar
  version: v1
  metadata:
  - name: host
    value: "localhost:6650"
  - name: enableTLS
    value: "false"
  - name: tenant
    value: "public"
  - name: token
    value: "eyJrZXlJZCI6InB1bHNhci1wajU0cXd3ZHB6NGIiLCJhbGciOiJIUzI1NiJ9.eyJzd"
  - name: namespace
    value: "default"
  - name: persistent
    value: "true"
  - name: backOffPolicy
    value: "constant"
  - name: backOffMaxRetries
    value: "-1"
  - name: disableBatching
    value: "false"

元数据字段规范

字段 必填 详情 示例
host Pulsar broker. 地址, 默认值是 "localhost:6650" "localhost:6650" OR "http://pulsar-pj54qwwdpz4b-pulsar.ap-sg.public.pulsar.com:8080"
enableTLS 启用TLS 默认值为 "false" "true", "false"
token 启动身份验证 How to create pulsar token
tenant 实例中的主题租户。 租户对于 Pulsar 中的多租户至关重要,并且分布在集群中。 默认值: “public” "public"
namespace The administrative unit of the topic, which acts as a grouping mechanism for related topics. Default: "default" "默认值"
persistent N Pulsar supports two kind of topics: persistent and non-persistent. With persistent topics, all messages are durably persisted on disks (if the broker is not standalone, messages are durably persisted on multiple disks), whereas data for non-persistent topics is not persisted to storage disks. Note: the default retry behavior is to retry until it succeeds, so when you use a non-persistent theme, you can reduce or prohibit retries by defining backOffMaxRetries to 0. 默认: "true" "true", "false"
backOffPolicy N 重试策略,"constant"是一个总是返回相同的退避延迟的退避策略。 "exponential" 是一种退避策略,它使用呈指数级增长的随机化函数增加每次重试尝试的退避周期。 默认为 "constant" constantexponential
backOffDuration N The fixed interval only takes effect when the backOffPolicy is "constant". There are two valid formats, one is the fraction with a unit suffix format, and the other is the pure digital format that is processed as milliseconds. 有效的时间单位为"ns"、“us”(或"μs")、“ms”、“s”、“m”、“h”。 默认为 "5s" "5s""5000"
backOffInitialInterval N 重试时的回退初始间隔。 Only takes effect when the backOffPolicy is "exponential". There are two valid formats, one is the fraction with a unit suffix format, and the other is the pure digital format that is processed as milliseconds. 有效的时间单位为"ns"、“us”(或"μs")、“ms”、“s”、“m”、“h”。 默认值为 "500" "50"
backOffMaxInterval N 重试时的回退初始间隔。 Only takes effect when the backOffPolicy is "exponential". There are two valid formats, one is the fraction with a unit suffix format, and the other is the pure digital format that is processed as milliseconds. 有效的时间单位为"ns"、“us”(或"μs")、“ms”、“s”、“m”、“h”。 默认为 "60s" "60000"
backOffMaxRetries N 返回错误前重试处理消息的最大次数。 默认为 "0" 这意味着组件不会重试处理消息。 "-1" 将无限期重试,直到处理完消息或关闭应用程序。 任何正数都被视为最大重试计数。 "3"
backOffRandomizationFactor N 随机系数,介于 1 和 0 之间,包括 0 但不是 1。 随机间隔 = 重试间隔 * (1 ±backOffRandomizationFactor)。 默认值为 "0.5". "0.5"
backOffMultiplier N 策略的退避倍数。 通过将间隔乘以倍数来递增间隔。 默认值为 "1.5" "1.5"
backOffMaxElapsedTime 在 MaxElapsedTime 之后,ExponentialBackOff 返回 Stop。 There are two valid formats, one is the fraction with a unit suffix format, and the other is the pure digital format that is processed as milliseconds. 有效的时间单位为"ns"、“us”(或"μs")、“ms”、“s”、“m”、“h”。 默认为 "15m" "15m"
disableBatching disable batching. 默认值为 "false" "true", "false"

延迟队列

当调用 Pulsar 发布/订阅时,在请求 url 中使用 metadata 查询参数来提供一个可选的延迟队列时可行的。

可选参数的名称为 metadata.deliverAtmetadata.deliverAfter

  • deliverAt: 延迟消息以在指定的时间投递 (RFC3339 格式),例如 "2021-09-01T10:00:00Z"
  • deliverAfter:延迟消息在指定的时间后进行投递,例如 "4h5m3s"

示例:

curl -X POST http://localhost:3500/v1.0/publish/myPulsar/myTopic?metadata.deliverAt='2021-09-01T10:00:00Z' \
  -H "Content-Type: application/json" \
  -d '{
        "data": {
          "message": "Hi"
        }
      }'

或者

curl -X POST http://localhost:3500/v1.0/publish/myPulsar/myTopic?metadata.deliverAfter='4h5m3s' \
  -H "Content-Type: application/json" \
  -d '{
        "data": {
          "message": "Hi"
        }
      }'

创建 Pulsar 实例


docker run -it \
  -p 6650:6650 \
  -p 8080:8080 \
  --mount source=pulsardata,target=/pulsar/data \
  --mount source=pulsarconf,target=/pulsar/conf \
  apachepulsar/pulsar:2.5.1 \
  bin/pulsar standalone

请参考以下Helm chart文档。

相关链接