发布订阅代理组件规范 与 Dapr 接口的受支持的发布订阅代理
下表列出了 Dapr 发布订阅构建块支持的发布和订阅代理。了解如何为 Dapr 发布和订阅设置不同的代理。
发布订阅组件重试与入站弹性 每个发布订阅组件都有自己的内置重试行为，这是消息代理解决方案特有的，与 Dapr 无关。在显式应用
Dapr 弹性策略 之前，请确保您了解所使用的发布订阅组件的隐式重试策略。Dapr 弹性不会覆盖这些内置重试，而是对其进行增强，这可能会导致消息的重复聚类。
Table headers to note:
Header Description Example Status Component certification statusAlpha Beta Stable Component version The version of the component v1 Since runtime version The version of the Dapr runtime when the component status was set or updated 1.11
Generic Amazon Web Services (AWS) Component Status Component version Since runtime version AWS SNS/SQS Stable v1 1.10
Google Cloud Platform (GCP) Component Status Component version Since runtime version GCP Pub/Sub Stable v1 1.11
Microsoft Azure 1 - Apache Kafka Apache Kafka 发布订阅组件的详细文档
组件格式 要设置 Apache Kafka 发布订阅，请创建类型为 pubsub.kafka 的组件。请参阅发布订阅代理组件文件 以了解 ConsumerID 是如何自动生成的。请阅读操作指南：发布与订阅指南 以了解如何创建和应用发布订阅配置。
所有组件元数据字段值都可以携带模板化元数据值 ，这些值在 Dapr 边车启动时解析。
例如，您可以选择使用 {namespace} 作为 consumerGroup，以支持在不同的命名空间中使用相同的 appId 和相同的主题，如本文 中所述。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "{namespace}"
- name : consumerID # 可选。如果未提供，运行时将创建一个。
value : "channel1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "password"
- name : saslUsername # 如果 authType 为 `password` 则必需。
value : "adminuser"
- name : saslPassword # 如果 authType 为 `password` 则必需。
secretKeyRef :
name : kafka-secrets
key : saslPasswordSecret
- name : saslMechanism
value : "SHA-512"
- name : maxMessageBytes # 可选。
value : 1024
- name : consumeRetryInterval # 可选。
value : 200ms
- name : heartbeatInterval # 可选。
value : 5s
- name : sessionTimeout # 可选。
value : 15s
- name : version # 可选。
value : 2.0.0
- name : disableTls # 可选。禁用 TLS。这对生产环境不安全！！您应该阅读 `Mutual TLS` 部分以了解如何使用 TLS。
value : "true"
- name : consumerFetchMin # 可选。高级设置。单个请求中获取的最小消息字节数 - 代理将等待直到至少有这么多数据可用。
value : 1
- name : consumerFetchDefault # 可选。高级设置。每个请求从代理获取的默认消息字节数。
value : 2097152
- name : channelBufferSize # 可选。高级设置。内部和外部通道中要缓冲的事件数量。
value : 512
- name : consumerGroupRebalanceStrategy # 可选。高级设置。用于消费者组重新平衡的策略。
value : sticky
- name : schemaRegistryURL # 可选。使用 Schema Registry Avro 序列化/反序列化时。Schema Registry URL。
value : http://localhost:8081
- name : schemaRegistryAPIKey # 可选。使用 Schema Registry Avro 序列化/反序列化时。Schema Registry API 密钥。
value : XYAXXAZ
- name : schemaRegistryAPISecret # 可选。使用 Schema Registry Avro 序列化/反序列化时。Schema Registry 凭证 API 密钥。
value : "ABCDEFGMEADFF"
- name : schemaCachingEnabled # 可选。使用 Schema Registry Avro 序列化/反序列化时。启用 schema 缓存。
value : true
- name : schemaLatestVersionCacheTTL # 可选。使用 Schema Registry Avro 序列化/反序列化时。使用最新可用 schema 发布消息时的 schema 缓存 TTL。
value : 5m
- name : useAvroJson # 可选。启用 Avro JSON schema 进行序列化，而不是默认的 Standard JSON。仅当订阅使用 valueSchemaType=Avro 时适用
value : "true"
- name : escapeHeaders # 可选。
value : false
有关使用 secretKeyRef 的详细信息，请参阅如何在组件中引用密钥 指南。
规格元数据字段 字段 必需 详细信息 示例 brokers Y 以逗号分隔的 Kafka 代理列表。 "localhost:9092,dapr-kafka.myapp.svc.cluster.local:9093"consumerGroup N 要监听的 Kafka 消费者组。发布到主题的每条记录都会传递给订阅该主题的每个消费者组中的一个消费者。如果提供了 consumerGroup 的值，任何 consumerID 的值都将被忽略 - 取而代之的是，将设置消费者组和随机唯一标识符的组合作为 consumerID。 "group1"consumerID N 消费者 ID（消费者标签）将一个或多个消费者组织成一个组。具有相同消费者 ID 的消费者作为一个虚拟消费者工作；例如，消息只由组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时会将其设置为 Dapr 应用程序 ID（appID）值。如果提供了 consumerGroup 的值，任何 consumerID 的值都将被忽略 - 取而代之的是，将设置消费者组和随机唯一标识符的组合作为 consumerID。 可以设置为字符串值（如上例中的 "channel1"）或字符串格式的值（如 "{podName}" 等）。查看您可以在组件元数据中使用的所有模板标签。 clientID N 用户提供的字符串，随每个请求发送到 Kafka 代理，用于日志记录、调试和审计目的。默认为 Kubernetes 模式的 "namespace.appID" 或自托管模式的 "appID"。 "my-namespace.my-dapr-app", "my-dapr-app"authRequired N 已弃用 使用 SASL 与 Kafka 代理进行身份验证。"true", "false"authType Y 配置或禁用身份验证。支持的值：none、password、mtls、oidc、oidc_private_key_jwt 或 awsiam "password", "none"saslUsername N 用于身份验证的 SASL 用户名。仅当 authType 设置为 "password" 时才需要。 "adminuser"saslPassword N 用于身份验证的 SASL 密码。可以是 secretKeyRef 以使用密钥引用 。仅当 authType 设置为 "password" 时才需要。 "", "KeFg23!"saslMechanism N 您希望使用的 SASL 身份验证机制。仅当 authType 设置为 "password" 时才需要。默认为 PLAINTEXT "SHA-512", "SHA-256", "PLAINTEXT"initialOffset N 如果之前未提交偏移量，则使用的初始偏移量。应该是 “newest” 或 “oldest”。默认为 “newest”。 "oldest"maxMessageBytes N 单个 Kafka 消息允许的最大字节大小。默认为 1024。 2048consumeRetryInterval N 尝试使用主题时重试之间的间隔。将没有后缀的数字视为毫秒。默认为 100ms。 200msconsumeRetryEnabled N 通过设置 "false" 来禁用消费重试 "true", "false"version N Kafka 集群版本。默认为 2.0.0。请注意，如果您使用的是 Azure EventHubs with Kafka，则必须将其设置为 1.0.0。 0.10.2.0caCert N 证书颁发机构证书，使用 TLS 所需。可以是 secretKeyRef 以使用密钥引用 "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"clientCert N 客户端证书，authType mtls 所需。可以是 secretKeyRef 以使用密钥引用 "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"clientKey N 客户端密钥，authType mtls 所需。可以是 secretKeyRef 以使用密钥引用 "-----BEGIN RSA PRIVATE KEY-----\n<base64-encoded PKCS8>\n-----END RSA PRIVATE KEY-----"skipVerify N 跳过 TLS 验证，不建议在生产环境中使用。默认为 "false" "true", "false"disableTls N 禁用传输安全性的 TLS。要禁用，您需要将值设置为 "true"。不建议在生产环境中使用。默认为 "false"。 "true", "false"oidcTokenEndpoint N OAuth2 身份提供者访问令牌端点的完整 URL。当 authType 设置为 oidc 时需要 “https://identity.example.com/v1/token" oidcClientID N 在身份提供者中配置的 OAuth2 客户端 ID。当 authType 设置为 oidc 时需要 dapr-kafkaoidcClientSecret N 在身份提供者中配置的 OAuth2 客户端密钥。当 authType 设置为 oidc 时需要 "KeFg23!"oidcScopes N 使用访问令牌请求的 OAuth2/OIDC 范围的逗号分隔列表。当 authType 设置为 oidc 或 oidc_private_key_jwt 时建议使用。默认为 "openid" "openid,kafka-prod"oidcClientAssertionCert N 用于身份验证的 OAuth2 客户端断言证书。当 authType 设置为 oidc_private_key_jwt 时需要。可以是 secretKeyRef 以使用密钥引用 "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"oidcClientAssertionKey N 用于身份验证的 OAuth2 客户端断言密钥。当 authType 设置为 oidc_private_key_jwt 时需要。可以是 secretKeyRef 以使用密钥引用 "-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----"oidcResource N 使用访问令牌请求的 OAuth2 资源。当 authType 设置为 oidc_private_key_jwt 时建议使用。 "api://kafka"oidcAudience N 使用访问令牌请求的 OAuth2 受众。当 authType 设置为 oidc_private_key_jwt 时建议使用。 "http://<idp-host>/realms/local"oidcKid N 使用访问令牌请求的 OAuth2 密钥 ID（kid）。当 authType 设置为 oidc_private_key_jwt 时建议使用。 "1234567890"oidcExtensions N 包含使用访问令牌请求的 OAuth2/OIDC 扩展的 JSON 编码字典的字符串 {"cluster":"kafka","poolid":"kafkapool"}awsRegion N 这保持与现有字段的向后兼容性。它将从 Dapr 1.17 开始弃用。请改用 ‘region’。部署 Kafka 集群的 AWS 区域。当 authType 设置为 awsiam 时需要 us-west-1awsAccessKey N 这保持与现有字段的向后兼容性。它将从 Dapr 1.17 开始弃用。请改用 ‘accessKey’。与 IAM 账户关联的 AWS 访问密钥。 "accessKey"awsSecretKey N 这保持与现有字段的向后兼容性。它将从 Dapr 1.17 开始弃用。请改用 ‘secretKey’。与访问密钥关联的密钥。 "secretKey"awsSessionToken N 这保持与现有字段的向后兼容性。它将从 Dapr 1.17 开始弃用。请改用 ‘sessionToken’。要使用的 AWS 会话令牌。仅当您使用临时安全凭证时才需要会话令牌。 "sessionToken"awsIamRoleArn N 这保持与现有字段的向后兼容性。它将从 Dapr 1.17 开始弃用。请改用 ‘assumeRoleArn’。有权访问 AWS Managed Streaming for Apache Kafka (MSK) 的 IAM 角色。这是使用 AWS 凭证对 MSK 进行身份验证的另一种选择。 "arn:aws:iam::123456789:role/mskRole"awsStsSessionName N 这保持与现有字段的向后兼容性。它将从 Dapr 1.17 开始弃用。请改用 ‘sessionName’。表示承担角色的会话名称。 "DaprDefaultSession"schemaRegistryURL N 使用 Schema Registry Avro 序列化/反序列化时需要。Schema Registry URL。 http://localhost:8081schemaRegistryAPIKey N 使用 Schema Registry Avro 序列化/反序列化时。Schema Registry 凭证 API 密钥。 XYAXXAZschemaRegistryAPISecret N 使用 Schema Registry Avro 序列化/反序列化时。Schema Registry 凭证 API 密钥。 ABCDEFGMEADFFschemaCachingEnabled N 使用 Schema Registry Avro 序列化/反序列化时。启用 schema 缓存。默认为 true trueschemaLatestVersionCacheTTL N 使用 Schema Registry Avro 序列化/反序列化时。使用最新可用 schema 发布消息时的 schema 缓存 TTL。默认为 5 分钟 5museAvroJson N 启用 Avro JSON schema 进行序列化，而不是默认的 Standard JSON。仅当订阅使用 valueSchemaType=Avro 时适用。默认为 "false" "true"clientConnectionTopicMetadataRefreshInterval N 客户端连接的主题元数据与代理刷新的间隔，以 Go 持续时间表示。默认为 9m。 "4m"clientConnectionKeepAliveInterval N 客户端连接在与代理保持活动状态的最大时间（以 Go 持续时间表示），然后关闭连接。零值（默认）表示无限期保持活动。 "4m"consumerFetchMin N 单个请求中获取的最小消息字节数 - 代理将等待直到至少有这么多数据可用。默认为 1，因为 0 会导致在没有可用消息时消费者空转。等效于 JVM 的 fetch.min.bytes。 "2"consumerFetchDefault N 每个请求从代理获取的默认消息字节数。默认为 "1048576" 字节。 "2097152"channelBufferSize N 在内部和外部通道中缓冲的事件数量。这允许生产者和消费者在用户代码工作时在后台继续处理某些消息，从而大大提高吞吐量。默认为 256。 "512"heartbeatInterval N 向消费者协调器发送心跳之间的间隔。该值最多应设置为 sessionTimeout 值的 1/3。默认为 “3s”。 "5s"sessionTimeout N 使用 Kafka 的组管理功能时用于检测客户端故障的超时时间。如果代理在此会话超时到期前未能收到消费者的任何心跳，则消费者将被移除并启动重新平衡。默认为 “10s”。 "20s"consumerGroupRebalanceStrategy N 用于消费者组重新平衡的策略。支持的值：range、sticky、roundrobin。默认为 range "sticky"escapeHeaders N 启用消费者接收的消息头值的 URL 转义。允许接收通常在 HTTP 头中不允许的特殊字符的内容。默认为 false。 trueexcludeHeaderMetaRegex N 一个正则表达式，用于在消费消息时排除将某些键从头转换为元数据，以及在发布消息时从元数据转换为头。此功能避免了主题消费者的意外下游副作用。 ‘"^valueSchemaType$”’
上述的 secretKeyRef 引用了 kubernetes secrets store 来访问 tls 信息。访问这里 以了解更多有关如何配置密钥存储组件的信息。
注意 当使用 Azure EventHubs with Kafka 时，元数据 version 必须设置为 1.0.0。
身份验证 Kafka 支持多种身份验证方案，Dapr 支持其中几种：SASL 密码、mTLS、OIDC/OAuth2。随着新增的身份验证方法，authRequired 字段已从 v1.6 版本开始弃用，而应使用 authType 字段。如果 authRequired 设置为 true，Dapr 将尝试根据 saslPassword 的值正确配置 authType。authType 的有效值为：
nonepasswordcertificatemtlsoidcoidc_private_key_jwtawsiam
注意 authType 仅用于身份验证。授权仍在 Kafka 中配置，但 awsiam 除外，它还可以驱动在 AWS IAM 中配置的授权决策。无身份验证 将 authType 设置为 none 将禁用任何身份验证。这在生产环境中不 推荐。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub-noauth
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "group1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "none"
- name : maxMessageBytes # 可选。
value : 1024
- name : consumeRetryInterval # 可选。
value : 200ms
- name : heartbeatInterval # 可选。
value : 5s
- name : sessionTimeout # 可选。
value : 15s
- name : version # 可选。
value : 0.10.2.0
- name : disableTls
value : "true"
SASL 密码 将 authType 设置为 password 可启用 SASL 身份验证。这需要设置 saslUsername 和 saslPassword 字段。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub-sasl
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "group1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "password"
- name : saslUsername # 如果 authType 为 `password` 则必需。
value : "adminuser"
- name : saslPassword # 如果 authType 为 `password` 则必需。
secretKeyRef :
name : kafka-secrets
key : saslPasswordSecret
- name : saslMechanism
value : "SHA-512"
- name : maxMessageBytes # 可选。
value : 1024
- name : consumeRetryInterval # 可选。
value : 200ms
- name : heartbeatInterval # 可选。
value : 5s
- name : sessionTimeout # 可选。
value : 15s
- name : version # 可选。
value : 0.10.2.0
- name : caCert
secretKeyRef :
name : kafka-tls
key : caCert
双向 TLS 将 authType 设置为 mtls 使用 x509 客户端证书（clientCert 字段）和密钥（clientKey 字段）进行身份验证。请注意，mTLS 作为一种身份验证机制与使用 TLS 通过加密保护传输层是不同的。mTLS 需要 TLS 传输（意味着 disableTls 必须为 false），但保护传输层不需要使用 mTLS。有关配置底层 TLS 传输的信息，请参阅使用 TLS 通信 。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub-mtls
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "group1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "mtls"
- name : caCert
secretKeyRef :
name : kafka-tls
key : caCert
- name : clientCert
secretKeyRef :
name : kafka-tls
key : clientCert
- name : clientKey
secretKeyRef :
name : kafka-tls
key : clientKey
- name : maxMessageBytes # 可选。
value : 1024
- name : consumeRetryInterval # 可选。
value : 200ms
- name : heartbeatInterval # 可选。
value : 5s
- name : sessionTimeout # 可选。
value : 15s
- name : version # 可选。
value : 0.10.2.0
OAuth2 或 OpenID Connect 将 authType 设置为 oidc 可通过 OAUTHBEARER 机制启用 SASL 身份验证。这支持从外部 OAuth2 或 OIDC 身份提供者指定不记名令牌。目前，仅支持 client_credentials 授权。
将 oidcTokenEndpoint 配置为身份提供者访问令牌端点的完整 URL。
将 oidcClientID 和 oidcClientSecret 设置为在身份提供者中配置的客户端凭证。
如果在组件配置中指定了 caCert，该证书将附加到系统 CA 信任中以验证身份提供者证书。同样，如果在组件配置中指定了 skipVerify，则在访问身份提供者时也将跳过验证。
默认情况下，为令牌请求的唯一范围是 openid；强烈建议通过 oidcScopes 以逗号分隔的列表指定其他范围，并由 Kafka 代理进行验证。如果不使用其他范围来缩小访问令牌的有效性，受损的 Kafka 代理可能会重放令牌以 Dapr clientID 的身份访问其他服务。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "group1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "oidc"
- name : oidcTokenEndpoint # 如果 authType 为 `oidc` 则必需。
value : "https://identity.example.com/v1/token"
- name : oidcClientID # 如果 authType 为 `oidc` 则必需。
value : "dapr-myapp"
- name : oidcClientSecret # 如果 authType 为 `oidc` 则必需。
secretKeyRef :
name : kafka-secrets
key : oidcClientSecret
- name : oidcScopes # 如果 authType 为 `oidc` 则建议使用。
value : "openid,kafka-dev"
- name : caCert # 也应用于验证 OIDC 提供者证书
secretKeyRef :
name : kafka-tls
key : caCert
- name : maxMessageBytes # 可选。
value : 1024
- name : consumeRetryInterval # 可选。
value : 200ms
- name : heartbeatInterval # 可选。
value : 5s
- name : sessionTimeout # 可选。
value : 15s
- name : version # 可选。
value : 0.10.2.0
OAuth2 私钥 JWT 将 authType 设置为 oidc_private_key_jwt 可通过 OAUTHBEARER 机制启用 SASL 身份验证。这支持从外部 OAuth2 或 OIDC 身份提供者指定私钥 JWT。目前，仅支持 client_credentials 授权。
将 oidcTokenEndpoint 配置为身份提供者访问令牌端点的完整 URL。
将 oidcClientID 设置为客户端 ID，oidcClientAssertionCert 设置为客户端断言证书，oidcClientAssertionKey 设置为在身份提供者中配置的客户端断言密钥。
如果在组件配置中指定了 caCert，该证书将附加到系统 CA 信任中以验证身份提供者证书。同样，如果在组件配置中指定了 skipVerify，则在访问身份提供者时也将跳过验证。
默认情况下，为令牌请求的唯一范围是 openid；强烈建议通过 oidcScopes 以逗号分隔的列表指定其他范围，并由 Kafka 代理进行验证。如果不使用其他范围来缩小访问令牌的有效性，受损的 Kafka 代理可能会重放令牌以 Dapr clientID 的身份访问其他服务。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "group1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "oidc_private_key_jwt"
- name : oidcTokenEndpoint # 如果 authType 为 `oidc_private_key_jwt` 则必需。
value : "https://identity.example.com/v1/token"
- name : oidcClientID # 如果 authType 为 `oidc_private_key_jwt` 则必需。
value : "dapr-myapp"
- name : oidcClientAssertionCert # 如果 authType 为 `oidc_private_key_jwt` 则必需。
secretKeyRef :
name : kafka-tls
key : oidcClientAssertionCert
- name : oidcClientAssertionKey # 如果 authType 为 `oidc_private_key_jwt` 则必需。
secretKeyRef :
name : kafka-tls
key : oidcClientAssertionKey
- name : oidcScopes # 如果 authType 为 `oidc_private_key_jwt` 则建议使用。
value : "openid,kafka-dev"
- name : oidcResource # 可选。
value : "api://kafka"
- name : oidcAudience # 可选。
value : "http://<idp-host>/realms/local"
- name : oidcKid # 可选。
value : "1234567890"
- name : caCert # 可选。
secretKeyRef :
name : kafka-tls
key : caCert
- name : maxMessageBytes # 可选。
value : 1024
- name : consumeRetryInterval # 可选。
value : 200ms
- name : heartbeatInterval # 可选。
value : 5s
- name : sessionTimeout # 可选。
value : 15s
- name : version # 可选。
value : 0.10.2.0
AWS IAM 支持使用 AWS IAM 与 MSK 进行身份验证。将 authType 设置为 awsiam 使用 AWS SDK 生成身份验证令牌进行身份验证。
注意 唯一必需的元数据字段是 region。如果未提供 accessKey 和 secretKey，您可以使用 AWS IAM 服务账户角色对 Kafka 集群进行无密码身份验证。apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub-awsiam
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "group1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "awsiam"
- name : region # 必需。
value : "us-west-1"
- name : accessKey # 可选。
value : <AWS_ACCESS_KEY>
- name : secretKey # 可选。
value : <AWS_SECRET_KEY>
- name : sessionToken # 可选。
value : <AWS_SESSION_KEY>
- name : assumeRoleArn # 可选。
value : "arn:aws:iam::123456789:role/mskRole"
- name : sessionName # 可选。
value : "DaprDefaultSession"
使用 TLS 通信 默认情况下，启用 TLS 以保护到 Kafka 的传输层。要禁用 TLS，请将 disableTls 设置为 true。启用 TLS 后，您可以使用 skipVerify 禁用验证（在生产环境中不 推荐）并使用 caCert 指定受信任的 TLS 证书颁发机构（CA）来控制服务器证书验证。如果未指定 caCert，将使用系统 CA 信任。要配置 mTLS 身份验证，请参阅_身份验证_下的部分。
以下是配置为使用传输层 TLS 的 Kafka 发布订阅组件的示例：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "group1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "certificate"
- name : consumeRetryInterval # 可选。
value : 200ms
- name : heartbeatInterval # 可选。
value : 5s
- name : sessionTimeout # 可选。
value : 15s
- name : version # 可选。
value : 0.10.2.0
- name : maxMessageBytes # 可选。
value : 1024
- name : caCert # 证书颁发机构证书。
secretKeyRef :
name : kafka-tls
key : caCert
auth :
secretStore : <SECRET_STORE_NAME>
从多个主题消费 当使用单个发布/订阅组件从多个主题消费时，无法保证消费者组中的消费者如何在主题分区之间进行平衡。
例如，假设您订阅了两个主题，每个主题有 10 个分区，并且您有 20 个服务副本从这两个主题消费。无法保证 10 个会被分配给第一个主题，10 个会被分配给第二个主题。相反，分区可能会不均匀地划分，第一个主题分配超过 10 个，其余的分配给第二个主题。
这可能导致监听第一个主题的消费者空闲，而第二个主题的消费者过度扩展，反之亦然。当使用自动伸缩器（如 HPA 或 KEDA）时，也会观察到相同的行为。
如果您遇到此特定问题，建议您为主题配置单个发布/订阅组件，并为每个组件唯一定义消费者组。这可以确保您的服务的所有副本都完全分配给唯一的消费者组，其中每个消费者组针对一个特定主题。
例如，您可以使用以下配置定义两个 Dapr 组件：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub-topic-one
spec :
type : pubsub.kafka
version : v1
metadata :
- name : consumerGroup
value : "{appID}-topic-one"
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub-topic-two
spec :
type : pubsub.kafka
version : v1
metadata :
- name : consumerGroup
value : "{appID}-topic-two"
发送和接收多条消息 Apache Kafka 组件支持使用批量发布/订阅 API 在单个操作中发送和接收多条消息。
配置批量订阅 订阅主题时，您可以配置 bulkSubscribe 选项。有关更多详细信息，请参阅批量订阅消息 。了解更多有关批量订阅 API 的信息。
Apache Kafka 支持以下批量元数据选项：
配置 默认值 maxAwaitDurationMs10000 (10s)maxMessagesCount80
每次调用的元数据字段 分区键 调用 Kafka 发布/订阅时，可以通过在请求 URL 中使用 metadata 查询参数来提供可选的分区键。
参数名称可以是 partitionKey 或 __key
示例：
curl -X POST http://localhost:3500/v1.0/publish/myKafka/myTopic?metadata.partitionKey= key1 \
"Content-Type: application/json" \
'{
"data": {
"message": "Hi"
}
}'
消息头 所有其他元数据键/值对（不是 partitionKey 或 __key）都作为 Kafka 消息的头设置。以下是设置消息的 correlationId 的示例。
curl -X POST http://localhost:3500/v1.0/publish/myKafka/myTopic?metadata.correlationId= myCorrelationID& metadata.partitionKey= key1 \
"Content-Type: application/json" \
'{
"data": {
"message": "Hi"
}
}'
消费者端接收的 Kafka 发布订阅特殊消息头 消费消息时，特殊消息元数据会作为头自动传递。这些包括：
__key：消息键（如果有）__topic：消息的主题__partition：消息的分区号__offset：消息在分区中的偏移量__timestamp：消息的时间戳您可以在消费者端点中按如下方式访问它们：
from fastapi import APIRouter , Body , Response , status
import json
import sys
app = FastAPI ()
router = APIRouter ()
@router.get ( '/dapr/subscribe' )
def subscribe ():
subscriptions = [{ 'pubsubname' : 'pubsub' ,
'topic' : 'my-topic' ,
'route' : 'my_topic_subscriber' ,
}]
return subscriptions
@router.post ( '/my_topic_subscriber' )
def my_topic_subscriber (
key : Annotated [ str , Header ( alias = "__key" )],
offset : Annotated [ int , Header ( alias = "__offset" )],
event_data = Body ()):
print ( f "key= { key } - offset= { offset } - data= { event_data } " , flush = True )
return Response ( status_code = status . HTTP_200_OK )
app . include_router ( router )
接收带有特殊字符的消息头 消费者应用程序可能需要接收包含特殊字符的消息头，这可能导致 HTTP 协议验证错误。
HTTP 头值必须遵循规范，因此不允许某些字符。了解更多有关协议的信息 。
在这种情况下，您可以启用 escapeHeaders 配置设置，该设置使用 URL 转义在消费者端对头值进行编码。
注意 使用此设置时，接收到的消息头是 URL 转义的，您需要 URL “un-escape”（转义）它以获取原始值。将 escapeHeaders 设置为 true 以进行 URL 转义。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub-escape-headers
spec :
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : consumerGroup # 可选。用于输入绑定。
value : "group1"
- name : clientID # 可选。被 Kafka 代理用作客户端追踪 ID。
value : "my-dapr-app-id"
- name : authType # 必需。
value : "none"
- name : escapeHeaders
value : "true"
Avro Schema Registry 序列化/反序列化 您可以配置发布/订阅以使用 Avro 二进制序列化 发布或消费编码的数据，利用 Apache Schema Registry （例如，Confluent Schema Registry 、Apicurio ）。
配置 重要 目前，仅支持消息值序列化/反序列化。由于不支持 Cloud Events，因此在发布 Avro 消息时必须传递 rawPayload=true 元数据。
请注意，消费者不应设置 rawPayload=true，因为消息值将被包装到 CloudEvent 中并进行 base64 编码。将 rawPayload 保留为默认值（即 false）会将 Avro 解码的消息作为 JSON 负载发送到应用程序。
当将 useAvroJson 组件元数据设置为 true 时，入站/出站 Avro 二进制会被转换为/从 Avro JSON 编码。
当需要准确的类型映射时，这可能更可取。
默认是标准 JSON，通常更容易绑定到应用程序中的本机类型。
配置 Kafka 发布/订阅组件元数据时，您必须定义：
Schema registry URL API 密钥/密钥（如果适用） Schema 主题使用标准命名约定从主题名称自动派生。例如，对于名为 my-topic 的主题，schema 主题将是 my-topic-value。
在服务内与消息负载交互时，它采用 JSON 格式。负载在 Dapr 组件内透明地序列化/反序列化。
日期/日期时间字段必须作为其 Epoch Unix 时间戳 等效值传递（而不是典型的 Iso8601）。例如：
2024-01-10T04:36:05.986Z 应作为 1704861365986 传递（自 1970 年 1 月 1 日以来的毫秒数）2024-01-10 应作为 19732 传递（自 1970 年 1 月 1 日以来的天数）发布 Avro 消息 为了向 Kafka 发布/订阅组件指示消息应使用 Avro 序列化，必须将 valueSchemaType 元数据设置为 Avro。
curl -X "POST" http://localhost:3500/v1.0/publish/pubsub/my-topic?metadata.rawPayload= true& metadata.valueSchemaType= Avro -H "Content-Type: application/json" -d '{"order_number": "345", "created_date": 1704861365986}'
from dapr.clients import DaprClient
with DaprClient () as d :
req_data = {
'order_number' : '345' ,
'created_date' : 1704861365986
}
# Create a typed message with content type and body
resp = d . publish_event (
pubsub_name = 'pubsub' ,
topic_name = 'my-topic' ,
data = json . dumps ( req_data ),
publish_metadata = { 'rawPayload' : 'true' , 'valueSchemaType' : 'Avro' }
)
# Print the request
print ( req_data , flush = True )
订阅 Avro 主题 为了向 Kafka 发布/订阅组件指示消息应使用 Avro 反序列化，必须在订阅元数据中将 valueSchemaType 元数据设置为 Avro。
from fastapi import APIRouter , Body , Response , status
import json
import sys
app = FastAPI ()
router = APIRouter ()
@router.get ( '/dapr/subscribe' )
def subscribe ():
subscriptions = [{ 'pubsubname' : 'pubsub' ,
'topic' : 'my-topic' ,
'route' : 'my_topic_subscriber' ,
'metadata' : {
'valueSchemaType' : 'Avro' ,
} }]
return subscriptions
@router.post ( '/my_topic_subscriber' )
def my_topic_subscriber ( event_data = Body ()):
print ( event_data , flush = True )
return Response ( status_code = status . HTTP_200_OK )
app . include_router ( router )
发布需要自定义元数据的消息时避免下游副作用 Dapr 允许通过设置自定义发布元数据来自定义发布行为。
例如，要以 avro 格式发布，需要设置 valueSchemaType=Avro 元数据。
然而，默认情况下，这些元数据项会被转换为 Kafka 头并随消息一起发布。这种默认行为对于在发布者/消费者链中转发追踪头非常有用。
但在某些情况下，它会产生不必要的副作用。
假设您使用 Dapr 使用上述头消费 Avro 消息。如果此消息无法成功消费并配置为发送到死信主题，则在发布到死信主题时会自动转发 valueSchemaType=Avro，需要设置与此主题关联的 schema。在许多情况下，最好仅以 JSON 格式发布死信消息，因为无法遵守确定的 schema。
要避免此行为，可以配置 kafka-pubsub 组件以排除某些元数据键被转换为/从头。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kafka-pubsub-exclude-metadata
type : pubsub.kafka
version : v1
metadata :
- name : brokers # 必需。Kafka 代理连接设置
value : "dapr-kafka.myapp.svc.cluster.local:9092"
- name : authType # 必需。
value : "none"
- name : excludeMetaHeaderRegex
value : "^valueSchemaType$" # 可选。排除 `valueSchemaType` 头被发布到头并转换为元数据
覆盖默认消费者组重新平衡 在 Kafka 中，重新平衡策略确定如何在消费者组内将分区分配给消费者。默认策略是 “range”，但也可以使用 “roundrobin” 和 “sticky”。
Range：
分区根据其字典顺序分配给消费者。
如果您有三个分区（0、1、2）和两个消费者（A、B），消费者 A 可能会获得分区 0 和 1，而消费者 B 获得分区 2。RoundRobin：
分区以循环方式分配给消费者。
在上面的相同示例中，消费者 A 可能会获得分区 0 和 2，而消费者 B 获得分区 1。Sticky：
此策略旨在尽可能保留以前的分配，同时仍保持平衡的分配。
如果消费者离开或加入组，只会重新分配受影响的分区，从而最大限度地减少中断。选择策略： Range：
简单易懂且易于实现，但如果分区大小差异很大，可能会导致分布不均匀。RoundRobin：
在许多情况下提供了良好的平衡，但如果消息键分布不均匀，可能不是最佳选择。Sticky：
通常首选，因为它能够在重新平衡期间最大限度地减少中断，尤其是在处理大量分区或频繁的消费者组更改时。创建 Kafka 实例 您可以使用这个 Docker 镜像在本地运行 Kafka。
要在不使用 Docker 的情况下运行，请参阅这里 的入门指南。
要在 Kubernetes 上运行 Kafka，您可以使用任何 Kafka 运算符，例如 Strimzi 。
相关链接 2 - AWS SNS/SQS AWS SNS/SQS 发布订阅组件的详细文档
组件格式 要设置 AWS SNS/SQS 发布订阅，请创建类型为 pubsub.aws.snssqs 的组件。
默认情况下，AWS SNS/SQS 组件：
生成 SNS 主题 预配 SQS 队列 配置队列到主题的订阅 注意 如果你只有发布者而没有订阅者，则仅创建 SNS 主题。
但是，如果你有订阅者，则会生成 SNS、SQS 及其动态或静态订阅。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : snssqs-pubsub
spec :
type : pubsub.aws.snssqs
version : v1
metadata :
- name : accessKey
value : "AKIAIOSFODNN7EXAMPLE"
- name : secretKey
value : "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
- name : region
value : "us-east-1"
# - name: consumerID # 可选。如果未提供，运行时将创建一个。
# value: "channel1"
# - name: endpoint # 可选。
# value: "http://localhost:4566"
# - name: sessionToken # 可选（如果使用 AssignedRole 则为必需；例如，临时 accessKey 和 secretKey）
# value: "TOKEN"
# - name: messageVisibilityTimeout # 可选
# value: 10
# - name: messageRetryLimit # 可选
# value: 10
# - name: messageReceiveLimit # 可选
# value: 10
# - name: sqsDeadLettersQueueName # 可选
# - value: "myapp-dlq"
# - name: messageWaitTimeSeconds # 可选
# value: 1
# - name: messageMaxNumber # 可选
# value: 10
# - name: fifo # 可选
# value: "true"
# - name: fifoMessageGroupID # 可选
# value: "app1-mgi"
# - name: disableEntityManagement # 可选
# value: "false"
# - name: disableDeleteOnRetryLimit # 可选
# value: "false"
# - name: assetsManagementTimeoutSeconds # 可选
# value: 5
# - name: concurrencyMode # 可选
# value: "single"
# - name: concurrencyLimit # 可选
# value: "0"
规范元数据字段 字段 必填 详情 示例 accessKey Y 具有对 SNS 和 SQS 适当权限的 AWS 账户/角色的 ID（见下文） "AKIAIOSFODNN7EXAMPLE"secretKey Y AWS 用户/角色的密钥。如果使用 AssumeRole 访问，你还需要提供 sessionToken "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"region Y SNS/SQS 资源所在或将要创建的 AWS 区域。有关有效区域，请参阅此页面 。确保 SNS 和 SQS 在该区域可用 "us-east-1"consumerID N 消费者 ID（消费者标签）将一个或多个消费者组织到一个组中。具有相同消费者 ID 的消费者作为一个虚拟消费者工作；例如，一条消息仅由组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时会将其设置为 Dapr 应用程序 ID（appID）的值。请参阅发布订阅代理组件文件 以了解 ConsumerID 如何自动生成。 可以设置为字符串值（如上面示例中的 "channel1"）或字符串格式值（如 "{podName}" 等）。查看你可以在组件元数据中使用的所有模板标签。 endpoint N 组件要使用的 AWS 端点。仅用于本地开发，例如使用 localstack 。当针对生产环境 AWS 运行时，不需要 endpoint "http://localhost:4566"sessionToken N 要使用的 AWS 会话令牌。仅当你使用临时安全凭证时才需要会话令牌 "TOKEN"messageReceiveLimit N 在处理消息失败后，接收消息的次数，达到该次数后，会将该消息从队列中移除。如果指定了 sqsDeadLettersQueueName，messageReceiveLimit 是在处理消息失败后接收消息的次数，达到该次数后，会将该消息移动到 SQS 死信队列。默认值：10 10sqsDeadLettersQueueName N 此应用程序的死信队列名称 "myapp-dlq"messageVisibilityTimeout N 消息发送给订阅者后，从接收请求中隐藏的时间（秒）。默认值：10 10messageRetryLimit N 在处理消息失败后，从队列中移除该消息之前，重新发送该消息的次数。默认值：10 10messageWaitTimeSeconds N 调用在返回之前等待消息到达队列的持续时间（秒）。如果有消息可用，则调用会早于 messageWaitTimeSeconds 返回。如果没有可用的消息且等待时间到期，则调用成功返回并返回空消息列表。默认值：1 1messageMaxNumber N 一次从队列接收的最大消息数。默认值：10，最大值：10 10fifo N 使用 SQS FIFO 队列提供消息排序和去重。默认值："false"。有关 SQS FIFO 的更多详情 "true", "false"fifoMessageGroupID N 如果启用了 fifo，指示 Dapr 为发布订阅部署使用自定义消息组 ID 。这不是必需的，因为 Dapr 会为每个生产者创建自定义消息组 ID，从而确保每个 Dapr 生产者的消息排序。默认值："" "app1-mgi"disableEntityManagement N 当设置为 true 时，SNS 主题、SQS 队列以及 SNS 的 SQS 订阅不会自动创建。默认值："false" "true", "false"disableDeleteOnRetryLimit N 当设置为 true 时，在重试和处理消息失败 messageRetryLimit 次后，重置消息可见性超时，以便其他消费者可以尝试处理，而不是从 SQS 中删除该消息（默认行为）。默认值："false" "true", "false"assetsManagementTimeoutSeconds N AWS 资源管理操作的超时时间（秒），超时后将取消操作。资源管理操作是在 STS、SNS 和 SQS 上执行的任何操作，除了实现默认 Dapr 组件重试行为的消息发布和消费操作。该值可以设置为任何非负浮点数/整数。默认值：5 0.5, 10concurrencyMode N 当从 SQS 批量接收消息时，按顺序（一次"单条"消息）或并发（“并行”）调用订阅者。默认值："parallel" "single", "parallel"concurrencyLimit N 定义处理消息的最大并发工作线程数。当 concurrencyMode 设置为 "single" 时，将忽略此值。要避免限制并发工作线程数，请将其设置为 0。默认值：0 100
附加信息 符合 AWS 规范 Dapr 创建的 SNS 主题和 SQS 队列名称符合 AWS 规范 。默认情况下，Dapr 根据消费者的 app-id 创建 SQS 队列名称，因此 Dapr 可能会执行名称标准化以符合 AWS 规范。
SNS/SQS 组件行为 当发布订阅 SNS/SQS 组件预配 SNS 主题时，SQS 队列和订阅在组件代表消息生产者（未部署订阅者应用程序）运行的情况下的行为，与存在订阅者应用程序（未部署发布者）的情况下的行为不同。
由于 SNS 在仅发布者设置中没有 SQS 订阅的工作方式，SQS 队列和订阅的行为类似于依赖于监听主题消息的订阅者的"经典"发布订阅系统。如果没有这些订阅者，消息：
无法继续传递并会被有效丢弃 对于未来的订阅者不可用（订阅者最终订阅时不会重放消息） SQS FIFO 根据 AWS 规范，使用 SQS FIFO（fifo 元数据字段设置为 "true"）提供消息排序和去重，但会降低 SQS 处理吞吐量，以及其他限制。
指定 fifoMessageGroupID 将所使用的 FIFO 队列的并发消费者数量限制为仅一个，但保证应用程序 Dapr 边车发布的消息的全局排序。请参阅此 AWS 博客文章 以更好地了解消息组 ID 和 FIFO 队列的主题。
为避免丢失传递给消费者的消息顺序，SQS 组件的 FIFO 配置需要将 concurrencyMode 元数据字段设置为 "single"。
默认并行 concurrencyMode 自 v1.8.0 起，该组件支持 "parallel" concurrencyMode 作为其默认模式。在以前的版本中，组件的默认行为是一次调用订阅者一条消息并等待其响应。
SQS 死信队列 在使用 SQS 死信队列配置发布订阅组件时，元数据字段 messageReceiveLimit 和 sqsDeadLettersQueueName 必须同时设置为某个值。对于 messageReceiveLimit，该值必须大于 0，并且 sqsDeadLettersQueueName 不能为空字符串。
重要 当在 EKS (AWS Kubernetes) 节点/Pod 上运行 Dapr 边车（daprd）与你的应用程序，并且已附加定义对 AWS 资源访问权限的 IAM 策略时，你不得 在组件规范定义中提供 AWS access-key、secret-key 和令牌。SNS/SQS 与 Dapr 的争用 从根本上说，SNS 通过创建对这些主题的 SQS 订阅，将来自多个发布者主题的消息聚合到单个 SQS 队列中。作为订阅者，SNS/SQS 发布订阅组件从该唯一的 SQS 队列消费消息。
然而，像任何 SQS 消费者一样，该组件无法有选择地检索发布到其特定订阅的 SNS 主题的消息。这可能导致组件接收来自没有关联处理程序的主题的消息。通常，这发生在：
组件初始化： 如果基础设施订阅在组件订阅处理程序之前准备就绪，或关闭： 如果组件处理程序在基础设施订阅之前被移除。由于此问题会影响多个 SNS 主题的任何 SQS 消费者，因此该组件无法防止消费来自缺少处理程序的主题的消息。当发生这种情况时，组件会记录一条错误，指示此类消息被错误地检索。
在这些情况下，未处理的消息将在每次拉取后在其接收计数 递减的情况下重新出现在 SQS 中。因此，存在未处理的消息可能超过其 messageReceiveLimit 并丢失的风险。
重要 在将 SNS/SQS 与 Dapr 一起使用时，请考虑潜在的争用情况，并适当配置 messageReceiveLimit。强烈建议通过设置 sqsDeadLettersQueueName 来使用 SQS 死信队列，以防止消息丢失。创建 SNS/SQS 实例 对于本地开发，localstack 项目 用于集成 AWS SNS/SQS。按照这些说明 运行 localstack。
要使用 Docker 在本地从命令行运行 localstack，请应用以下命令：
docker run --rm -it -p 4566:4566 -p 4571:4571 -e SERVICES = "sts,sns,sqs" -e AWS_DEFAULT_REGION = "us-east-1" localstack/localstack
为了将 localstack 与你的发布订阅绑定一起使用，你需要在组件元数据中提供 endpoint 配置。当针对生产环境 AWS 运行时，不需要 endpoint。
有关身份验证相关属性的信息，请参阅向 AWS 进行身份验证 。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : snssqs-pubsub
spec :
type : pubsub.aws.snssqs
version : v1
metadata :
- name : accessKey
value : "anyString"
- name : secretKey
value : "anyString"
- name : endpoint
value : http://localhost:4566
# 如果提供给了 localstack，则使用 us-east-1 或任何其他区域，如 "AWS_DEFAULT_REGION" 环境变量所定义
- name : region
value : us-east-1
要在 Kubernetes 上运行 localstack，你可以应用以下配置。然后可以通过 DNS 名称 http://localstack.default.svc.cluster.local:4566 访问 Localstack（假设此配置应用于默认命名空间），该名称应作为 endpoint 使用。
apiVersion : apps/v1
kind : Deployment
metadata :
name : localstack
spec :
# 使用选择器，我们将暴露正在运行的部署
# Kubernetes 知道给定的服务属于某个部署
selector :
matchLabels :
app : localstack
replicas : 1
template :
metadata :
labels :
app : localstack
spec :
containers :
- name : localstack
image : localstack/localstack:latest
ports :
# 暴露边缘端点
- containerPort : 4566
---
kind : Service
apiVersion : v1
metadata :
name : localstack
labels :
app : localstack
spec :
selector :
app : localstack
ports :
- protocol : TCP
port : 4566
targetPort : 4566
type : LoadBalancer
为了在 AWS 中运行，请创建或分配具有对 SNS 和 SQS 服务权限的 IAM 用户，并使用类似以下的策略：
{
"Version" : "2012-10-17" ,
"Statement" : [
{
"Sid" : "YOUR_POLICY_NAME" ,
"Effect" : "Allow" ,
"Action" : [
"sns:CreateTopic" ,
"sns:GetTopicAttributes" ,
"sns:ListSubscriptionsByTopic" ,
"sns:Publish" ,
"sns:Subscribe" ,
"sns:TagResource" ,
"sqs:ChangeMessageVisibility" ,
"sqs:CreateQueue" ,
"sqs:DeleteMessage" ,
"sqs:GetQueueAttributes" ,
"sqs:GetQueueUrl" ,
"sqs:ReceiveMessage" ,
"sqs:SetQueueAttributes" ,
"sqs:TagQueue"
],
"Resource" : [
"arn:aws:sns:AWS_REGION:AWS_ACCOUNT_ID:*" ,
"arn:aws:sqs:AWS_REGION:AWS_ACCOUNT_ID:*"
]
}
]
}
使用 Kubernetes 密钥和 secretKeyRef 将 AWS 账户 ID 和 AWS 账户密钥插入组件元数据中的 accessKey 和 secretKey。
或者，假设你想使用自己选择的工具（例如 Terraform）预配 SNS 和 SQS 资源，同时防止 Dapr 动态执行此操作。你需要启用 disableEntityManagement 并为你的使用 Dapr 的应用程序分配一个 IAM 角色，并使用类似以下的策略：
{
"Version" : "2012-10-17" ,
"Statement" : [
{
"Sid" : "YOUR_POLICY_NAME" ,
"Effect" : "Allow" ,
"Action" : [
"sqs:DeleteMessage" ,
"sqs:ReceiveMessage" ,
"sqs:ChangeMessageVisibility" ,
"sqs:GetQueueUrl" ,
"sqs:GetQueueAttributes" ,
"sns:Publish" ,
"sns:ListSubscriptionsByTopic" ,
"sns:GetTopicAttributes"
],
"Resource" : [
"arn:aws:sns:AWS_REGION:AWS_ACCOUNT_ID:APP_TOPIC_NAME" ,
"arn:aws:sqs:AWS_REGION:AWS_ACCOUNT_ID:APP_ID"
]
}
]
}
在上面的示例中，你在 EKS 集群上运行应用程序，并启用了动态资源创建（默认的 Dapr 行为）。
相关链接 3 - Azure Event Hubs Azure Event Hubs 发布订阅组件的详细文档
组件格式 要设置 Azure Event Hubs 发布订阅，请创建类型为 pubsub.azure.eventhubs 的组件。请参阅发布订阅代理组件文件 以了解 ConsumerID 是如何自动生成的。阅读操作指南：发布和订阅 以了解如何创建和应用发布订阅配置。
除了下面显示的配置元数据字段外，Azure Event Hubs 还支持 Azure 身份验证 机制。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : eventhubs-pubsub
spec :
type : pubsub.azure.eventhubs
version : v1
metadata :
# connectionString 或 eventHubNamespace 是必需的
# 当*不*使用 Microsoft Entra ID 时使用 connectionString
- name : connectionString
value : "Endpoint=sb://{EventHubNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key};EntityPath={EventHub}"
# 使用 Microsoft Entra ID 时使用 eventHubNamespace
- name : eventHubNamespace
value : "namespace"
- name : consumerID # 可选。如果未提供，运行时将创建一个。
value : "channel1"
- name : enableEntityManagement
value : "false"
- name : enableInOrderMessageDelivery
value : "false"
# 仅当 enableEntityManagement 设置为 true 时才需要以下四个属性
- name : resourceGroupName
value : "test-rg"
- name : subscriptionID
value : "Azure 订阅 ID 的值"
- name : partitionCount
value : "1"
- name : messageRetentionInDays
value : "3"
# 检查点存储属性
- name : storageAccountName
value : "myeventhubstorage"
- name : storageAccountKey
value : "112233445566778899"
- name : storageContainerName
value : "myeventhubstoragecontainer"
# 传递 storageAccountKey 的替代方案
- name : storageConnectionString
value : "DefaultEndpointsProtocol=https;AccountName=<account>;AccountKey=<account-key>"
警告 上述示例将密钥作为纯字符串使用。建议按照
此处 的说明使用密钥存储来管理密钥。
规范元数据字段 字段 必填 详情 示例 connectionStringY* Event Hub 或 Event Hub 命名空间的连接字符串。eventHubNamespace 字段互斥。Microsoft Entra ID 身份验证 时必需。 "Endpoint=sb://{EventHubNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key};EntityPath={EventHub}" 或 "Endpoint=sb://{EventHubNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key}"eventHubNamespaceY* Event Hub 命名空间名称。connectionString 字段互斥。Microsoft Entra ID 身份验证 时必需。 "namespace"consumerIDN 消费者 ID（消费者标签）将一个或多个消费者组织成一个组。具有相同消费者 ID 的消费者作为一个虚拟消费者工作；例如，一条消息只由组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时会将其设置为 Dapr 应用程序 ID（appID）的值。 可以设置为字符串值（如上例中的 "channel1"）或字符串格式的值（如 "{podName}" 等）。查看您可以在组件元数据中使用的所有模板标签。 enableEntityManagementN 布尔值，允许管理 EventHub 命名空间和存储账户。默认值：false "true", "false"enableInOrderMessageDeliveryN 输入/输出 布尔值，允许按照消息发布的顺序传递消息。这假设在发布或发布时设置了 partitionKey 以确保跨分区的顺序。默认值：false storageAccountNameY 用于检查点存储的存储账户名称。 "myeventhubstorage"storageAccountKeyY* 检查点存储账户的存储账户密钥。 "112233445566778899"storageConnectionStringY* 检查点存储的连接字符串，指定 storageAccountKey 的替代方案 "DefaultEndpointsProtocol=https;AccountName=myeventhubstorage;AccountKey=<account-key>"storageContainerNameY 存储账户名称的存储容器名称。 "myeventhubstoragecontainer"resourceGroupNameN Event Hub 命名空间所属的资源组名称。在启用实体管理时必需 "test-rg"subscriptionIDN Azure 订阅 ID 值。在启用实体管理时必需 "azure subscription id"partitionCountN 新 Event Hub 命名空间的分区数量。仅在启用实体管理时使用。默认值："1" "2"messageRetentionInDaysN 在新创建的 Event Hub 命名空间中保留消息的天数。仅在启用实体管理时使用。默认值："1" "90"
Microsoft Entra ID 身份验证 Azure Event Hubs 发布订阅组件支持使用所有 Microsoft Entra ID 机制进行身份验证。有关更多信息以及根据所选的 Microsoft Entra ID 身份验证机制需要提供的组件元数据字段，请参阅Azure 身份验证文档 。
配置示例 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : eventhubs-pubsub
spec :
type : pubsub.azure.eventhubs
version : v1
metadata :
# 使用的 Azure 身份验证
- name : azureTenantId
value : "***"
- name : azureClientId
value : "***"
- name : azureClientSecret
value : "***"
- name : eventHubNamespace
value : "namespace"
- name : enableEntityManagement
value : "false"
# 仅当 enableEntityManagement 设置为 true 时才需要以下四个属性
- name : resourceGroupName
value : "test-rg"
- name : subscriptionID
value : "Azure 订阅 ID 的值"
- name : partitionCount
value : "1"
- name : messageRetentionInDays
# 检查点存储属性
# 在此情况下，我们也使用 Microsoft Entra ID 访问存储账户
- name : storageAccountName
value : "myeventhubstorage"
- name : storageContainerName
value : "myeventhubstoragecontainer"
发送和接收多条消息 Azure Event Hubs 支持使用批量发布订阅 API 在单个操作中发送和接收多条消息。
配置批量发布 要设置批量发布操作的元数据，请在 HTTP 请求或 gRPC 元数据上设置查询参数，如 API 参考中所述 。
元数据 默认值 metadata.maxBulkPubBytes1000000
配置批量订阅 订阅主题时，您可以配置 bulkSubscribe 选项。有关更多详细信息，请参阅批量订阅消息 并了解批量订阅 API 。
配置 默认值 maxMessagesCount100maxAwaitDurationMs10000
配置检查点频率 订阅主题时，您可以在 HTTP 或 gRPC 订阅请求中设置元数据 来配置分区中的检查点频率。此元数据在分区事件序列中达到配置数量的事件后启用检查点。通过将频率设置为 0 来禁用检查点。
了解有关检查点的更多信息 。
元数据 默认值 metadata.checkPointFrequencyPerPartition1
以下示例显示了使用 checkPointFrequencyPerPartition 元数据的声明式订阅 的示例订阅文件。同样，您也可以在程序化订阅 中传递元数据。
apiVersion : dapr.io/v2alpha1
kind : Subscription
metadata :
name : order-pub-sub
spec :
topic : orders
routes :
default : /checkout
pubsubname : order-pub-sub
metadata :
checkPointFrequencyPerPartition : 1
scopes :
orderprocessing
checkout
注意 使用 BulkSubscribe 订阅主题时，您将检查点配置为在指定数量的_批次_之后发生，而不是事件，其中_批次_是指在单个请求中接收到的事件集合。创建 Azure Event Hub 按照文档 中的说明设置 Azure Event Hubs。
由于此组件使用 Azure Storage 作为检查点存储，您还需要一个 Azure Storage 账户 。按照文档 中的说明管理存储账户访问密钥。
请参阅文档 了解如何获取 Event Hubs 连接字符串（注意这不是 Event Hubs 命名空间的连接字符串）。
为每个订阅者创建消费者组 对于每个想要订阅事件的 Dapr 应用程序，请创建一个以 Dapr 应用程序 ID 命名的 Event Hubs 消费者组。例如，一个在 Kubernetes 上运行且 dapr.io/app-id: "myapp" 的 Dapr 应用程序将需要一个名为 myapp 的 Event Hubs 消费者组。
注意：Dapr 将消费者组的名称传递给 Event Hub，因此不需要在元数据中提供。
实体管理 当在元数据中启用实体管理时，只要应用程序具有操作 Event Hub 命名空间的正确角色和权限，Dapr 就可以自动为您创建 Event Hub 和消费者组。
Event Hub 名称是发布或订阅的传入请求中的 topic 字段，而消费者组名称是订阅给定 Event Hub 的 Dapr 应用程序的名称。例如，一个在 Kubernetes 上运行且名称为 dapr.io/app-id: "myapp" 的 Dapr 应用程序需要一个名为 myapp 的 Event Hubs 消费者组。
实体管理仅在使用 Microsoft Entra ID 身份验证 时才可用，而不使用连接字符串。
Dapr 将消费者组的名称传递给 Event Hub，因此不需要在元数据中提供。
接收自定义属性 默认情况下，Dapr 不会转发自定义属性 。但是，通过将订阅元数据 requireAllProperties 设置为 "true"，您可以接收自定义属性作为 HTTP 头。
apiVersion : dapr.io/v2alpha1
kind : Subscription
metadata :
name : order-pub-sub
spec :
topic : orders
routes :
default : /checkout
pubsubname : order-pub-sub
metadata :
requireAllProperties : "true"
可以使用 Dapr SDK 实现相同的效果：
[Topic("order-pub-sub", "orders")]
[TopicMetadata("requireAllProperties", "true")]
[HttpPost("checkout")]
public ActionResult Checkout ( Order order , [ FromHeader ] int priority )
{
return Ok ();
}
订阅 Azure IoT Hub 事件 Azure IoT Hub 提供了一个与 Event Hubs 兼容的端点 ，因此 Azure Event Hubs 发布订阅组件也可用于订阅 Azure IoT Hub 事件。
由 Azure IoT Hub 设备创建的设备到云事件将包含额外的 IoT Hub 系统属性 ，Dapr 的 Azure Event Hubs 发布订阅组件将把以下内容作为响应元数据的一部分返回：
系统属性名称 描述和路由查询关键字 iothub-connection-auth-generation-id发送消息的设备的 connectionDeviceGenerationId 。请参阅 IoT Hub 设备身份属性 。 iothub-connection-auth-method用于对发送消息的设备进行身份验证的 connectionAuthMethod 。 iothub-connection-device-id发送消息的设备的 deviceId 。请参阅 IoT Hub 设备身份属性 。 iothub-connection-module-id发送消息的设备的 moduleId 。请参阅 IoT Hub 设备身份属性 。 iothub-enqueuedtimeIoT Hub 接收到设备到云消息的 RFC3339 格式的 enqueuedTime 。 message-id用户可设置的 AMQP messageId 。
例如，传递的 HTTP 订阅消息的头将包含：
{
'user-agent' : 'fasthttp' ,
'host' : '127.0.0.1:3000' ,
'content-type' : 'application/json' ,
'content-length' : '120' ,
'iothub-connection-device-id' : 'my-test-device' ,
'iothub-connection-auth-generation-id' : '637618061680407492' ,
'iothub-connection-auth-method' : '{"scope":"module","type":"sas","issuer":"iothub","acceptingIpFilterRule":null}' ,
'iothub-connection-module-id' : 'my-test-module-a' ,
'iothub-enqueuedtime' : '2021-07-13T22:08:09Z' ,
'message-id' : 'my-custom-message-id' ,
'x-opt-sequence-number' : '35' ,
'x-opt-enqueued-time' : '2021-07-13T22:08:09Z' ,
'x-opt-offset' : '21560' ,
'traceparent' : '00-4655608164bc48b985b42d39865f3834-ed6cf3697c86e7bd-01'
}
相关链接 4 - Azure Service Bus 队列 Azure Service Bus 队列发布订阅组件的详细文档
组件格式 要设置 Azure Service Bus 队列发布订阅，需创建类型为 pubsub.azure.servicebus.queues 的组件。请参阅 发布订阅代理组件文件 以了解 ConsumerID 是如何自动生成的。阅读操作指南：发布和订阅 了解如何创建和应用发布订阅配置。
此组件使用 Azure Service Bus 上的队列；有关主题和队列 之间差异的说明，请参阅官方文档。
若要使用主题，请参阅 Azure Service Bus Topics 发布订阅组件 。
连接字符串身份验证 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : servicebus-pubsub
spec :
type : pubsub.azure.servicebus.queues
version : v1
metadata :
# 未使用 Microsoft Entra ID 身份验证时必需
- name : connectionString
value : "Endpoint=sb://{ServiceBusNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key};EntityPath={ServiceBus}"
# - name: consumerID # 可选
# value: channel1
# - name: timeoutInSec # 可选
# value: 60
# - name: handlerTimeoutInSec # 可选
# value: 60
# - name: disableEntityManagement # 可选
# value: "false"
# - name: maxDeliveryCount # 可选
# value: 3
# - name: lockDurationInSec # 可选
# value: 60
# - name: lockRenewalInSec # 可选
# value: 20
# - name: maxActiveMessages # 可选
# value: 10000
# - name: maxConcurrentHandlers # 可选
# value: 10
# - name: defaultMessageTimeToLiveInSec # 可选
# value: 10
# - name: autoDeleteOnIdleInSec # 可选
# value: 3600
# - name: minConnectionRecoveryInSec # 可选
# value: 2
# - name: maxConnectionRecoveryInSec # 可选
# value: 300
# - name: maxRetriableErrorsPerSec # 可选
# value: 10
# - name: publishMaxRetries # 可选
# value: 5
# - name: publishInitialRetryIntervalInMs # 可选
# value: 500
警告 上述示例将密钥作为纯字符串使用。建议按照
此处 的描述使用密钥存储来管理密钥。
规范元数据字段 字段 必填 详情 示例 connectionStringY Service Bus 的共享访问策略连接字符串。除非使用 Microsoft Entra ID 身份验证，否则为必填。 见上方示例 consumerIDN 消费者 ID（consumer tag）将一个或多个消费者组织成一个组。具有相同消费者 ID 的消费者作为一个虚拟消费者工作；例如，一条消息仅由组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时会将其设置为 Dapr 应用程序 ID（appID）的值。 可以设置为字符串值（如上方示例中的 "channel1"）或字符串格式的值（如 "{podName}" 等）。请参阅您可以在组件元数据中使用的所有模板标签。 namespaceNameN 用于设置 Service Bus 命名空间地址的参数，为完全限定域名。如果使用 Microsoft Entra ID 身份验证，则为必填。 "namespace.servicebus.windows.net"timeoutInSecN 发送消息和管理操作的超时时间。默认值：60 30handlerTimeoutInSecN 调用应用程序处理器的超时时间。默认值：60 30lockRenewalInSecN 定义缓冲消息锁的续订频率。默认值：20。 20maxActiveMessagesN 定义一次处理或缓冲的最大消息数。此值应至少与最大并发处理器数一样大。默认值：1000 2000maxConcurrentHandlersN 定义最大并发消息处理器数。默认值：0（无限制） 10disableEntityManagementN 当设置为 true 时，队列和订阅不会自动创建。默认值："false" "true", "false"defaultMessageTimeToLiveInSecN 默认消息存活时间，以秒为单位。仅在创建订阅时使用。 10autoDeleteOnIdleInSecN 在自动删除空闲订阅之前等待的时间（秒）。仅在创建订阅时使用。必须为 300 秒或更长。默认值：0（禁用） 3600maxDeliveryCountN 定义服务器尝试传递消息的次数。仅在创建订阅时使用。默认值由服务器设置。 10lockDurationInSecN 定义消息在过期前被锁定的时长（秒）。仅在创建订阅时使用。默认值由服务器设置。 30minConnectionRecoveryInSecN 在连接失败后重新连接到 Azure Service Bus 之前等待的最小间隔（秒）。默认值：2 5maxConnectionRecoveryInSecN 在连接失败后重新连接到 Azure Service Bus 之前等待的最大间隔（秒）。每次尝试后，组件会在最小值和最大值之间等待一个随机秒数，每次都在增加。默认值：300（5 分钟） 600maxRetriableErrorsPerSecN 每秒处理的最大可重试错误数。如果消息因可重试错误而处理失败，组件会在开始处理另一条消息之前添加延迟，以避免立即重新处理失败的消息。默认值：10 10publishMaxRetriesN 当 Azure Service Bus 响应"太忙"以限制消息时的最大重试次数。默认值：5 5publishInitialRetryIntervalInMsN 当 Azure Service Bus 限制消息时初始指数退避的时间（毫秒）。默认值：500 500
Microsoft Entra ID 身份验证 Azure Service Bus 队列发布订阅组件支持使用所有 Microsoft Entra ID 机制进行身份验证，包括托管标识。有关更多信息以及根据选择的 Microsoft Entra ID 身份验证机制需要提供的相关组件元数据字段，请参阅向 Azure 进行身份验证的文档 。
示例配置 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : servicebus-pubsub
spec :
type : pubsub.azure.servicebus.queues
version : v1
metadata :
- name : namespaceName
# 使用 Azure 身份验证时必需。
# 必须是完全限定域名
value : "servicebusnamespace.servicebus.windows.net"
- name : azureTenantId
value : "***"
- name : azureClientId
value : "***"
- name : azureClientSecret
value : "***"
消息元数据 Azure Service Bus 消息使用其他上下文元数据扩展了 Dapr 消息格式。一些元数据字段由 Azure Service Bus 本身设置（只读），其他字段可以在发布消息时由客户端设置。
发送带有元数据的消息 要在发送消息时设置 Azure Service Bus 元数据，请在 HTTP 请求上设置查询参数或 gRPC 元数据，如此处 所述。
metadata.MessageIdmetadata.CorrelationIdmetadata.SessionIdmetadata.Labelmetadata.ReplyTometadata.PartitionKeymetadata.Tometadata.ContentTypemetadata.ScheduledEnqueueTimeUtcmetadata.ReplyToSessionId注意 metadata.MessageId 属性不会设置 Dapr 返回的 Cloud Event 的 id 属性，应单独处理。metadata.ScheduledEnqueueTimeUtc 属性支持 RFC1123 和 RFC3339 时间戳格式。接收带有元数据的消息 当 Dapr 调用您的应用程序时，它使用 HTTP 标头或 gRPC 元数据将 Azure Service Bus 消息元数据附加到请求中。
除了上面列出的可设置元数据 之外，您还可以访问以下只读消息元数据。
metadata.DeliveryCountmetadata.LockedUntilUtcmetadata.LockTokenmetadata.EnqueuedTimeUtcmetadata.SequenceNumber要了解这些元数据属性的用途的更多详细信息，请参阅官方 Azure Service Bus 文档 。
此外，原始 Azure Service Bus 消息的所有 ApplicationProperties 条目都会作为 metadata.<application property's name> 附加。
注意 所有时间都由服务器填充，不会针对时钟偏差进行调整。发送和接收多条消息 Azure Service Bus 支持使用批量发布订阅 API 在单个操作中发送和接收多条消息。
配置批量发布 要设置批量发布操作的元数据，请在 HTTP 请求上设置查询参数或 gRPC 元数据，如此处 所述
元数据 默认值 metadata.maxBulkPubBytes131072 (128 KiB)
配置批量订阅 订阅主题时，您可以配置 bulkSubscribe 选项。有关更多详细信息，请参阅批量订阅消息 。了解有关批量订阅 API 的更多信息。
为队列创建 Azure Service Bus 代理 按照此处 的说明设置 Azure Service Bus 队列。
注意 您的队列必须与您使用 Dapr 发布到的主题同名。例如，如果您在主题 "orders" 上发布到发布订阅 "myPubsub"，您的队列必须命名为 "orders"。
如果您使用共享访问策略连接到队列，该策略必须能够"管理"队列。要使用死信队列，该策略必须位于包含主队列和死信队列的 Service Bus 命名空间上。重试策略和死信队列 默认情况下，Azure Service Bus 队列有一个死信队列。消息会重试 maxDeliveryCount 给定的次数。默认的 maxDeliveryCount 值默认为 10，但可以设置为最高 2000。这些重试发生得非常快，如果没有返回成功，消息将被放入死信队列。
Dapr 发布订阅提供了自己的死信队列概念，允许您控制重试策略并通过 Dapr 订阅死信队列。
在 Azure Service Bus 命名空间中设置一个单独的队列作为死信队列，以及一个定义如何重试的弹性策略。 订阅主题以获取失败的消息并处理它们。 例如，在订阅中设置死信队列 orders-dlq 和弹性策略可以让您订阅主题 orders-dlq 来处理失败的消息。
有关设置死信队列的更多详细信息，请参阅死信文章 。
相关链接 5 - Azure Service Bus 主题 Azure Service Bus 主题发布订阅组件的详细文档
组件格式 要设置 Azure Service Bus 主题发布订阅，请创建类型为 pubsub.azure.servicebus.topics 的组件。请参阅发布订阅代理组件文件 以了解 ConsumerID 是如何自动生成的。阅读如何：发布和订阅指南 以了解如何创建和应用发布订阅配置。
此组件使用 Azure Service Bus 上的主题；有关主题和队列 之间差异，请参阅官方文档。Azure Service Bus 队列发布订阅组件 。
连接字符串身份验证 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : servicebus-pubsub
spec :
type : pubsub.azure.servicebus.topics
version : v1
metadata :
# 不使用 Microsoft Entra ID 身份验证时必需
- name : connectionString
value : "Endpoint=sb://{ServiceBusNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key};EntityPath={ServiceBus}"
# - name: consumerID # 可选：默认为应用自身的 ID
# value: channel1
# - name: timeoutInSec # 可选
# value: 60
# - name: handlerTimeoutInSec # 可选
# value: 60
# - name: disableEntityManagement # 可选
# value: "false"
# - name: maxDeliveryCount # 可选
# value: 3
# - name: lockDurationInSec # 可选
# value: 60
# - name: lockRenewalInSec # 可选
# value: 20
# - name: maxActiveMessages # 可选
# value: 10000
# - name: maxConcurrentHandlers # 可选
# value: 10
# - name: defaultMessageTimeToLiveInSec # 可选
# value: 10
# - name: autoDeleteOnIdleInSec # 可选
# value: 3600
# - name: minConnectionRecoveryInSec # 可选
# value: 2
# - name: maxConnectionRecoveryInSec # 可选
# value: 300
# - name: maxRetriableErrorsPerSec # 可选
# value: 10
# - name: publishMaxRetries # 可选
# value: 5
# - name: publishInitialRetryIntervalInMs # 可选
# value: 500
注意： 上述设置在使用此组件的所有主题之间共享。
警告 上述示例使用纯文本字符串作为密钥。建议按照
此处 的描述使用密钥存储来管理密钥。
规范元数据字段 字段 必填 详情 示例 connectionStringY Service Bus 的共享访问策略连接字符串。除非使用 Microsoft Entra ID 身份验证，否则必需。 参见上面的示例 namespaceNameN 用于设置 Service Bus 命名空间地址的参数，为完全限定域名。如果使用 Microsoft Entra ID 身份验证，则为必需。 "namespace.servicebus.windows.net"consumerIDN 消费者 ID（消费者标签）将一个或多个消费者组织成一个组。具有相同消费者 ID 的消费者作为一个虚拟消费者工作；例如，消息只由组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时将其设置为 Dapr 应用程序 ID（appID）值。(appID) 值。 可以设置为字符串值（例如上面示例中的 "channel1"）或字符串格式值（例如 "{podName}" 等）。请参阅您可以在组件元数据中使用的所有模板标签。 timeoutInSecN 发送消息和管理操作的超时时间。默认值：60 30handlerTimeoutInSecN 调用应用程序处理程序的超时时间。默认值：60 30lockRenewalInSecN 定义缓冲消息锁的续订频率。默认值：20。 20maxActiveMessagesN 定义一次处理或缓冲的最大消息数。此值应至少等于最大并发处理程序数。默认值：1000 2000maxConcurrentHandlersN 定义并发消息处理程序的最大数量。默认值：0（无限制） 10disableEntityManagementN 当设置为 true 时，队列和订阅不会自动创建。默认值："false" "true", "false"defaultMessageTimeToLiveInSecN 默认消息生存时间（秒）。仅在订阅创建期间使用。 10autoDeleteOnIdleInSecN 自动删除空闲订阅前等待的时间（秒）。仅在订阅创建期间使用。必须为 300 秒或更长。默认值：0（禁用） 3600maxDeliveryCountN 定义服务器尝试传递消息的次数。仅在订阅创建期间使用。默认值由服务器设置。 10lockDurationInSecN 定义消息在过期前被锁定的秒数。仅在订阅创建期间使用。默认值由服务器设置。 30minConnectionRecoveryInSecN 在连接失败后尝试重新连接到 Azure Service Bus 之前等待的最小间隔（秒）。默认值：2 5maxConnectionRecoveryInSecN 在连接失败后尝试重新连接到 Azure Service Bus 之前等待的最大间隔（秒）。在每次尝试之后，组件在最小值和最大值之间等待一个随机秒数，每次都会增加。默认值：300（5 分钟） 600maxRetriableErrorsPerSecN 每秒处理的可重试错误的最大数量。如果消息因可重试错误而处理失败，组件会在开始处理另一条消息之前添加延迟，以避免立即重新处理失败的消息。默认值：10 10publishMaxRetriesN 当 Azure Service Bus 响应"太忙"以限制消息时的最大重试次数。默认值：5 5publishInitialRetryIntervalInMsN 当 Azure Service Bus 限制消息时，初始指数退避的时间（毫秒）。默认值：500 500
Microsoft Entra ID 身份验证 Azure Service Bus 主题发布订阅组件支持使用所有 Microsoft Entra ID 机制进行身份验证，包括托管标识。有关详细信息以及根据选择的 Microsoft Entra ID 身份验证机制需要提供的相关组件元数据字段，请参阅Azure 身份验证文档 。
配置示例 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : servicebus-pubsub
spec :
type : pubsub.azure.servicebus.topics
version : v1
metadata :
- name : namespaceName
# 使用 Azure 身份验证时必需。
# 必须是完全限定域名
value : "servicebusnamespace.servicebus.windows.net"
- name : azureTenantId
value : "***"
- name : azureClientId
value : "***"
- name : azureClientSecret
value : "***"
消息元数据 Azure Service Bus 消息使用额外的上下文元数据扩展了 Dapr 消息格式。一些元数据字段由 Azure Service Bus 本身设置（只读），其他字段可以在发布消息时由客户端设置。
发送带有元数据的消息 要在发送消息时设置 Azure Service Bus 元数据，请按照此处 的文档在 HTTP 请求上设置查询参数或在 gRPC 元数据中设置。
metadata.MessageIdmetadata.CorrelationIdmetadata.SessionIdmetadata.Labelmetadata.ReplyTometadata.PartitionKeymetadata.Tometadata.ContentTypemetadata.ScheduledEnqueueTimeUtcmetadata.ReplyToSessionId注意： metadata.MessageId 属性不会设置 Dapr 返回的 cloud event 的 id 属性，应单独处理。
注意： 如果未设置 metadata.SessionId 属性但主题需要会话，则将使用空的会话 ID。
注意： metadata.ScheduledEnqueueTimeUtc 属性支持 RFC1123 和 RFC3339 时间戳格式。
接收带有元数据的消息 当 Dapr 调用您的应用程序时，它会使用 HTTP 头或 gRPC 元数据将 Azure Service Bus 消息元数据附加到请求中。除了上面列出的可设置元数据 之外，您还可以访问以下只读消息元数据。
metadata.DeliveryCountmetadata.LockedUntilUtcmetadata.LockTokenmetadata.EnqueuedTimeUtcmetadata.SequenceNumber要了解有关这些元数据属性用途的更多详细信息，请参阅官方 Azure Service Bus 文档 。
此外，原始 Azure Service Bus 消息的所有 ApplicationProperties 条目都将作为 metadata.<application property's name> 附加。
注意：所有时间都由服务器填充，不会针对时钟偏差进行调整。
订阅启用会话的主题 要订阅已启用会话 的主题，您可以在订阅元数据中提供以下属性。
requireSessions（默认：false）sessionIdleTimeoutInSec（默认：60）maxConcurrentSessions（默认：8）为主题创建 Azure Service Bus 代理 按照此处 的说明设置 Azure Service Bus 主题。
相关链接 6 - GCP GCP Pub/Sub 组件的详细文档
创建 Dapr 组件 要设置 GCP 发布订阅，需创建类型为 pubsub.gcp.pubsub 的组件。请参阅发布订阅代理组件文件 以了解 ConsumerID 如何自动生成。阅读操作指南：发布和订阅指南 了解如何创建和应用发布订阅配置。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : gcp-pubsub
spec :
type : pubsub.gcp.pubsub
version : v1
metadata :
- name : type
value : service_account
- name : projectId
value : <PROJECT_ID> # replace
- name : endpoint # Optional.
value : "http://localhost:8085"
- name : consumerID # Optional - defaults to the app's own ID
value : <CONSUMER_ID>
- name : identityProjectId
value : <IDENTITY_PROJECT_ID> # replace
- name : privateKeyId
value : <PRIVATE_KEY_ID> #replace
- name : clientEmail
value : <CLIENT_EMAIL> #replace
- name : clientId
value : <CLIENT_ID> # replace
- name : authUri
value : https://accounts.google.com/o/oauth2/auth
- name : tokenUri
value : https://oauth2.googleapis.com/token
- name : authProviderX509CertUrl
value : https://www.googleapis.com/oauth2/v1/certs
- name : clientX509CertUrl
value : https://www.googleapis.com/robot/v1/metadata/x509/<PROJECT_NAME>.iam.gserviceaccount.com #replace PROJECT_NAME
- name : privateKey
value : <PRIVATE_KEY> # replace x509 cert
- name : disableEntityManagement
value : "false"
- name : enableMessageOrdering
value : "false"
- name : orderingKey # Optional
value : <ORDERING_KEY>
- name : maxReconnectionAttempts # Optional
value : 30
- name : connectionRecoveryInSec # Optional
value : 2
- name : deadLetterTopic # Optional
value : <EXISTING_PUBSUB_TOPIC>
- name : maxDeliveryAttempts # Optional
value : 5
- name : maxOutstandingMessages # Optional
value : 1000
- name : maxOutstandingBytes # Optional
value : 1000000000
- name : maxConcurrentConnections # Optional
value : 10
警告 上述示例将密钥作为纯字符串使用。建议使用密钥存储，如
此处 所述。
规范元数据字段 字段 必填 详情 示例 projectId Y GCP 项目 ID myproject-123endpoint N 组件使用的 GCP 端点。仅用于本地开发（例如）与 GCP Pub/Sub 模拟器 配合使用。当针对 GCP 生产 API 运行时，不需要 endpoint。 "http://localhost:8085"consumerIDN Consumer ID 将一个或多个消费者组织成一个组。具有相同 consumer ID 的消费者作为一个虚拟消费者工作；例如，一条消息仅被组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时会将其设置为 Dapr 应用程序 ID (appID) 值。consumerID 与请求的一部分提供的 topic 一起用于构建发布订阅订阅 ID 可以设置为字符串值（如 "channel1"）或字符串格式值（如 "{podName}" 等）。请参阅您可以在组件元数据中使用的所有模板标签。 identityProjectId N 如果 GCP 发布订阅项目与身份项目不同，请使用此属性指定身份项目 "myproject-123"privateKeyId N 如果使用显式凭据，此字段应包含服务账户 json 文档中的 private_key_id 字段 "my-private-key"privateKey N 如果使用显式凭据，此字段应包含服务账户 json 中的 private_key 字段 -----BEGIN PRIVATE KEY-----MIIBVgIBADANBgkqhkiG9w0BclientEmail N 如果使用显式凭据，此字段应包含服务账户 json 中的 client_email 字段 "myservice@myproject-123.iam.gserviceaccount.com"clientId N 如果使用显式凭据，此字段应包含服务账户 json 中的 client_id 字段 106234234234authUri N 如果使用显式凭据，此字段应包含服务账户 json 中的 auth_uri 字段 https://accounts.google.com/o/oauth2/authtokenUri N 如果使用显式凭据，此字段应包含服务账户 json 中的 token_uri 字段 https://oauth2.googleapis.com/tokenauthProviderX509CertUrl N 如果使用显式凭据，此字段应包含服务账户 json 中的 auth_provider_x509_cert_url 字段 https://www.googleapis.com/oauth2/v1/certsclientX509CertUrl N 如果使用显式凭据，此字段应包含服务账户 json 中的 client_x509_cert_url 字段 https://www.googleapis.com/robot/v1/metadata/x509/myserviceaccount%40myproject.iam.gserviceaccount.comdisableEntityManagement N 当设置为 "true" 时，主题和订阅不会自动创建。默认值："false" "true", "false"enableMessageOrdering N 当设置为 "true" 时，订阅的消息将按顺序接收，具体取决于发布和权限配置。 "true", "false"orderingKey N 请求中提供的键。当 enableMessageOrdering 设置为 true 时使用，用于根据该键对消息进行排序。 “my-orderingkey” maxReconnectionAttempts N 定义最大重连尝试次数。默认值：30 30connectionRecoveryInSec N 连接恢复尝试之间等待的秒数。默认值：2 2deadLetterTopic N GCP Pub/Sub 主题的名称。使用此组件前，此主题必须 存在。 "myapp-dlq"maxDeliveryAttempts N 尝试传递消息的最大次数。如果指定了 deadLetterTopic，maxDeliveryAttempts 是消息处理失败的最大尝试次数。达到该次数后，消息将被移动到死信主题。默认值：5 5type N 已弃用 GCP 凭据类型。仅支持 service_account。默认值为 service_accountservice_accountmaxOutstandingMessages N 给定流式拉取 连接可以拥有的最大未完成消息数。默认值：1000 50maxOutstandingBytes N 给定流式拉取 连接可以拥有的最大未完成字节数。默认值：1000000000 1000000000maxConcurrentConnections N 要维护的最大并发流式拉取 连接数。默认值：10 2ackDeadline N 消息确认持续时间截止时间。默认值：20s 1m
警告 如果 enableMessageOrdering 设置为 “true”，则服务账户需要 roles/viewer 或 roles/pubsub.viewer 角色，以便在消息中未嵌入顺序令牌的情况下保证顺序。如果未授予此角色，或由于任何其他原因对 Subscription.Config() 的调用失败，按嵌入顺序令牌排序仍将正常工作。GCP 凭据 由于 GCP Pub/Sub 组件使用 GCP Go 客户端库，默认情况下它使用应用程序默认凭据 进行身份验证。这在使用客户端库向 GCP Cloud 服务进行身份验证 指南中有进一步说明。
创建 GCP Pub/Sub 对于本地开发，使用 GCP Pub/Sub 模拟器 来测试 GCP Pub/Sub 组件。按照这些说明 运行 GCP Pub/Sub 模拟器。
要使用 Docker 在本地运行 GCP Pub/Sub 模拟器，请使用以下 docker-compose.yaml：
version : '3'
services :
pubsub :
image : gcr.io/google.com/cloudsdktool/cloud-sdk:422.0.0-emulators
ports :
- "8085:8085"
container_name : gcp-pubsub
entrypoint : gcloud beta emulators pubsub start --project local-test-prj --host-port 0.0.0.0:8085
为了将 GCP Pub/Sub 模拟器与您的发布订阅绑定一起使用，您需要在组件元数据中提供 endpoint 配置。当针对 GCP 生产 API 运行时，不需要 endpoint。
projectId 属性必须与 docker-compose.yaml 或 Docker 命令中使用的 --project 匹配。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : gcp-pubsub
spec :
type : pubsub.gcp.pubsub
version : v1
metadata :
- name : projectId
value : "local-test-prj"
- name : consumerID
value : "testConsumer"
- name : endpoint
value : "localhost:8085"
您可以使用"显式"或"隐式"凭据来配置对 GCP 发布订阅实例的访问。如果使用显式凭据，大多数字段都是必填的。隐式凭据依赖于在 Kubernetes 服务账户（KSA）下运行的 dapr，该账户映射到具有访问发布订阅所需权限的 Google 服务账户（GSA）。在隐式模式下，只需要 projectId 属性，所有其他属性都是可选的。
按照此处 的说明设置 Google Cloud Pub/Sub 系统。
相关链接 7 - In-memory In Memory 发布订阅组件的详细文档
内存中的发布订阅组件在单个 Dapr 边车内运行。这主要用于开发目的。状态不会在多个边车之间复制，当 Dapr 边车重启时，状态会丢失。
组件格式 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : pubsub
spec :
type : pubsub.in-memory
version : v1
metadata : []
注意：内存中不需要任何特定的元数据即可使组件工作，但是 spec.metadata 是必填字段。
相关链接 8 - JetStream NATS JetStream 组件的详细文档
组件格式 要设置 JetStream 发布订阅，请创建类型为 pubsub.jetstream 的组件。请参阅发布订阅代理组件文件 以了解 ConsumerID 是如何自动生成的。阅读如何：发布和订阅指南 以了解如何创建和应用发布订阅配置。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : jetstream-pubsub
spec :
type : pubsub.jetstream
version : v1
metadata :
- name : natsURL
value : "nats://localhost:4222"
- name : jwt # 可选。用于去中心化 JWT 认证。
value : "eyJhbGciOiJ...6yJV_adQssw5c"
- name : seedKey # 可选。用于去中心化 JWT 认证。
value : "SUACS34K232O...5Z3POU7BNIL4Y"
- name : tls_client_cert # 可选。用于 TLS 客户端认证。
value : "/path/to/tls.crt"
- name : tls_client_key # 可选。用于 TLS 客户端认证。
value : "/path/to/tls.key"
- name : token # 可选。用于基于 Token 的认证。
value : "my-token"
- name : name
value : "my-conn-name"
- name : streamName
value : "my-stream"
- name : durableName
value : "my-durable-subscription"
- name : queueGroupName
value : "my-queue-group"
- name : startSequence
value : 1
- name : startTime # Unix 格式
value : 1630349391
- name : flowControl
value : false
- name : ackWait
value : 10s
- name : maxDeliver
value : 5
- name : backOff
value : "50ms, 1s, 10s"
- name : maxAckPending
value : 5000
- name : replicas
value : 1
- name : memoryStorage
value : false
- name : rateLimit
value : 1024
- name : heartbeat
value : 15s
- name : ackPolicy
value : explicit
- name : deliverPolicy
value : all
- name : domain
value : hub
- name : apiPrefix
value : PREFIX
规范元数据字段 字段 必填 详情 示例 natsURL Y NATS 服务器地址 URL "nats://localhost:4222"jwt N NATS 去中心化认证 JWT "eyJhbGciOiJ...6yJV_adQssw5c"seedKey N NATS 去中心化认证 Seed Key "SUACS34K232O...5Z3POU7BNIL4Y"tls_client_cert N NATS TLS 客户端认证证书 "/path/to/tls.crt"tls_client_key N NATS TLS 客户端认证密钥 "/path/to/tls.key"token N [NATS 基于 Token 的认证] "my-token"name N NATS 连接名称 "my-conn-name"streamName N 要绑定的 JetStream Stream 名称 "my-stream"durableName N Durable name "my-durable"queueGroupName N 队列组名称 "my-queue"startSequence N Start Sequence 1startTime N Unix 格式的[开始时间] 1630349391flowControl N [流控] trueackWait N Ack Wait 10smaxDeliver N [最大投递次数] 15backOff N [退避策略] "50ms, 1s, 5s, 10s"maxAckPending N [最大待确认消息数] 5000replicas N [副本数] 3memoryStorage N [内存存储] falserateLimit N [速率限制] 1024heartbeat N [心跳] 10sackPolicy N [确认策略] explicitdeliverPolicy N 可选值：all、last、new、sequence、time alldomain N [JetStream Leafondes] HUBapiPrefix N [JetStream Leafnodes] PREFIX
创建 NATS 服务器 你可以使用 Docker 在本地运行启用了 JetStream 的 NATS 服务器：
docker run -d -p 4222:4222 nats:latest -js
然后你可以使用客户端端口与服务器交互：localhost:4222。
在 Kubernetes 上安装 NATS JetStream，使用 helm ：
helm repo add nats https://nats-io.github.io/k8s/helm/charts/
helm install --set nats.jetstream.enabled= true my-nats nats/nats
这将在 default 命名空间中安装单个 NATS 服务器。要查找与 NATS 交互的服务，请使用：
有关 helm chart 设置的更多信息，请参阅 Helm chart 文档 。
创建 JetStream 为特定的主题创建 NATS JetStream 是必不可少的。例如，对于本地运行的 NATS 服务器，使用：
nats -s localhost:4222 stream add myStream --subjects mySubject
示例：竞争消费者模式 假设你希望每条消息仅由一个具有相同 app-id 的应用程序或 Pod 处理。通常，consumerID 元数据规范可以帮助你定义竞争消费者。
由于 NATS JetStream 不支持 consumerID，你需要指定 durableName 和 queueGroupName 来实现竞争消费者模式。例如：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : pubsub
spec :
type : pubsub.jetstream
version : v1
metadata :
- name : name
value : "my-conn-name"
- name : streamName
value : "my-stream"
- name : durableName
value : "my-durable-subscription"
- name : queueGroupName
value : "my-queue-group"
相关链接 9 - KubeMQ KubeMQ 发布订阅组件的详细文档
组件格式 要设置 KubeMQ 发布订阅，需创建类型为 pubsub.kubemq 的组件。参阅发布订阅代理组件文件 以了解如何自动生成 ConsumerID。阅读操作指南：发布与订阅 以了解如何创建和应用发布订阅配置。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : kubemq-pubsub
spec :
type : pubsub.kubemq
version : v1
metadata :
- name : address
value : localhost:50000
- name : store
value : false
- name : consumerID
value : channel1
规范元数据字段 字段 必填 详情 示例 address Y KubeMQ 服务器的地址 "localhost:50000"store N 发布订阅类型，true：发布订阅持久化（EventsStore），false：发布订阅内存（Events） true 或 false（默认为 false）consumerID N 消费者 ID（消费者标签）将一个或多个消费者组织成一个组。具有相同消费者 ID 的消费者作为一个虚拟消费者工作；例如，一条消息仅由该组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时将其设置为 Dapr 应用程序 ID（appID）值。 可设置为字符串值（例如上面示例中的 "channel1"）或字符串格式值（例如 "{podName}" 等）。查看您可在组件元数据中使用的所有模板标签。 clientID N 客户端 ID 连接的名称 sub-client-12345authToken N 用于连接的身份验证 JWT 令牌 查看 KubeMQ Authentication ew...group N 用于负载均衡的订阅者组 g1disableReDelivery N 设置在来自应用程序发生错误的情况下是否应重新传递消息 true 或 false（默认为 false）
创建 KubeMQ 代理 获取 KubeMQ 密钥 。等待带有您的密钥的电子邮件确认 您可以使用 Docker 运行 KubeMQ 代理：
docker run -d -p 8080:8080 -p 50000:50000 -p 9090:9090 -e KUBEMQ_TOKEN = <your-key> kubemq/kubemq
然后，您可以使用客户端端口与服务器交互：localhost:50000
获取 KubeMQ 密钥 。等待带有您的密钥的电子邮件确认 然后运行以下 kubectl 命令：
kubectl apply -f https://deploy.kubemq.io/init
kubectl apply -f https://deploy.kubemq.io/key/<your-key>
安装 KubeMQ CLI 前往 KubeMQ CLI 并下载最新版本的 CLI。
浏览 KubeMQ 仪表板 安装了 KubeMQCTL 后，运行以下命令：
或者，安装了 kubectl 后，运行 port-forward 命令：
kubectl port-forward svc/kubemq-cluster-api -n kubemq 8080:8080
KubeMQ 文档 访问 KubeMQ 文档 以获取更多信息。
相关链接 10 - MQTT MQTT 发布订阅组件的详细文档
组件格式 要设置 MQTT 发布订阅，请创建类型为 pubsub.mqtt 的组件。请参阅 发布订阅代理组件文件 以了解如何自动生成 ConsumerID。阅读 操作指南：发布和订阅 了解如何创建和应用发布订阅配置。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : mqtt-pubsub
spec :
type : pubsub.mqtt
version : v1
metadata :
- name : url
value : "tcp://[username][:password]@host.domain[:port]"
- name : qos
value : 1
- name : retain
value : "false"
- name : cleanSession
value : "false"
- name : consumerID
value : "channel1"
警告 上述示例将密钥作为纯字符串使用。建议按照
此处 的描述使用密钥存储来管理密钥。
规范元数据字段 字段 必填 详情 示例 url 是 MQTT 代理的地址。可以是 secretKeyRef 以使用密钥引用。tcp://ssl:// "tcp://[username][:password]@host.domain[:port]"consumerID 否 用于连接到 MQTT 代理的客户端 ID（用于消费者连接）。默认为 Dapr 应用 ID。producerID，则会在此值后追加 -consumer 作为消费者连接的 ID 可以设置为字符串值（如上例中的 "channel1"）或字符串格式的值（如 "{podName}" 等）。请参阅您可以在组件元数据中使用的所有模板标签。 producerID 否 用于连接到 MQTT 代理的客户端 ID（用于生产者连接）。默认为 {consumerID}-producer。 "myMqttProducerApp"qos 否 指示消息的服务质量级别（QoS）（更多信息 ）。默认为 1。 0、1、2retain 否 定义代理是否将消息保存为指定主题的最后一个已知良好值。默认为 "false"。 "true"、"false"cleanSession 否 如果为 "true"，则在发送到 MQTT 代理的连接消息中设置 clean_session 标志（更多信息 ）。默认为 "false"。 "true"、"false"caCert 使用 TLS 时必填 用于验证服务器 TLS 证书的 PEM 格式证书颁发机构（CA）证书。 "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"clientCert 使用 TLS 时必填 PEM 格式的 TLS 客户端证书。必须与 clientKey 一起使用。 "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"clientKey 使用 TLS 时必填 PEM 格式的 TLS 客户端密钥。必须与 clientCert 一起使用。可以是 secretKeyRef 以使用密钥引用。 "-----BEGIN RSA PRIVATE KEY-----\n<base64-encoded PKCS8>\n-----END RSA PRIVATE KEY-----"
启用消息传递重试 MQTT 发布订阅组件没有内置重试策略支持。这意味着边车只会向服务发送一次消息。如果服务将消息标记为未处理，消息将不会被确认回代理。只有当代理重新发送消息时，才会重试。
要使 Dapr 使用更复杂的重试策略，您可以将重试弹性策略 应用于 MQTT 发布订阅组件。
这两种重试方式之间存在关键区别：
未确认消息的重新传递完全依赖于代理。Dapr 不保证这一点。某些代理如 emqx 、vernemq 等支持它，但这不是 MQTT3 规范 的一部分。
使用重试弹性策略 会使同一个 Dapr 边车重试重新传递消息。因此是同一个 Dapr 边车和同一个应用程序接收同一条消息。
使用 TLS 进行通信 要配置使用 TLS 进行通信，请确保 MQTT 代理（例如 mosquitto）配置为支持证书，并在组件配置中提供 caCert、clientCert、clientKey 元数据。例如：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : mqtt-pubsub
spec :
type : pubsub.mqtt
version : v1
metadata :
- name : url
value : "ssl://host.domain[:port]"
- name : qos
value : 1
- name : retain
value : "false"
- name : cleanSession
value : "false"
- name : caCert
value : ${{ myLoadedCACert }}
- name : clientCert
value : ${{ myLoadedClientCert }}
- name : clientKey
secretKeyRef :
name : myMqttClientKey
key : myMqttClientKey
auth :
secretStore : <SECRET_STORE_NAME>
请注意，虽然 caCert 和 clientCert 值可能不是密钥，但为了方便起见，也可以从 Dapr 密钥存储中引用它们。
消费共享主题 在消费共享主题时，每个消费者必须具有唯一标识符。默认情况下，应用程序 ID 用于唯一标识每个消费者和发布者。在自托管模式下，每次调用 dapr run 时使用不同的应用程序 ID 足以让它们从同一个共享主题消费。然而，在 Kubernetes 上，应用程序 pod 的多个实例将共享同一个应用程序 ID，从而阻止所有实例消费同一个主题。为了克服这一点，请使用 {uuid} 标签配置组件的 consumerID 元数据，这将在启动时为每个实例分配一个随机生成的 consumerID 值。例如：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : mqtt-pubsub
spec :
type : pubsub.mqtt
version : v1
metadata :
- name : consumerID
value : "{uuid}"
- name : url
value : "tcp://admin:public@localhost:1883"
- name : qos
value : 1
- name : retain
value : "false"
- name : cleanSession
value : "true"
警告 上述示例将密钥作为纯字符串使用。建议按照
此处 的描述使用密钥存储来管理密钥。
请注意，在这种情况下，每次 Dapr 重启时消费者 ID 的值都是随机的，因此我们也将 cleanSession 设置为 true。
创建 MQTT 代理 您可以使用 Docker 在本地运行 MQTT 代理 ：
docker run -d -p 1883:1883 -p 9001:9001 --name mqtt eclipse-mosquitto:1.6
然后您可以使用客户端端口与服务器交互：mqtt://localhost:1883
您可以使用以下 yaml 在 kubernetes 中运行 MQTT 代理：
apiVersion : apps/v1
kind : Deployment
metadata :
name : mqtt-broker
labels :
app-name : mqtt-broker
spec :
replicas : 1
selector :
matchLabels :
app-name : mqtt-broker
template :
metadata :
labels :
app-name : mqtt-broker
spec :
containers :
- name : mqtt
image : eclipse-mosquitto:1.6
imagePullPolicy : IfNotPresent
ports :
- name : default
containerPort : 1883
protocol : TCP
- name : websocket
containerPort : 9001
protocol : TCP
---
apiVersion : v1
kind : Service
metadata :
name : mqtt-broker
labels :
app-name : mqtt-broker
spec :
type : ClusterIP
selector :
app-name : mqtt-broker
ports :
- port : 1883
targetPort : default
name : default
protocol : TCP
- port : 9001
targetPort : websocket
name : websocket
protocol : TCP
然后您可以使用客户端端口与服务器交互：tcp://mqtt-broker.default.svc.cluster.local:1883
相关链接 11 - MQTT3 MQTT3 发布订阅组件的详细文档
组件格式 要设置 MQTT3 发布订阅，请创建类型为 pubsub.mqtt3 的组件。请参阅 发布订阅代理组件文件 以了解 ConsumerID 是如何自动生成的。阅读 操作指南：发布和订阅指南 了解如何创建和应用发布订阅配置。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : mqtt-pubsub
spec :
type : pubsub.mqtt3
version : v1
metadata :
- name : url
value : "tcp://[username][:password]@host.domain[:port]"
# Optional
- name : retain
value : "false"
- name : cleanSession
value : "false"
- name : qos
value : "1"
- name : consumerID
value : "channel1"
Warning 上述示例将密钥作为纯字符串使用。建议按照
此处 所述使用密钥存储。
规格元数据字段 字段 必填 详情 示例 urlY MQTT 代理的地址。可以是 secretKeyRef 以使用密钥引用。tcp://ssl:// "tcp://[username][:password]@host.domain[:port]"consumerIDN 用于连接到 MQTT 代理的客户端 ID。默认为 Dapr 应用 ID。 可以设置为字符串值（例如上面示例中的 "channel1"）或字符串格式值（例如 "{podName}" 等）。查看您可以在组件元数据中使用的所有模板标签。 retainN 定义消息是否由代理保存为指定主题的最后一个已知良好值。默认为 "false"。 "true", "false"cleanSessionN 如果为 "true"，则设置到 MQTT 代理的连接消息中的 clean_session 标志（更多信息 ）。默认为 "false"。 "true", "false"caCert使用 TLS 时必填 用于验证服务器 TLS 证书的 PEM 格式证书颁发机构 (CA) 证书。 参见下面的示例 clientCert使用 TLS 时必填 PEM 格式的 TLS 客户端证书。必须与 clientKey 一起使用。 参见下面的示例 clientKey使用 TLS 时必填 PEM 格式的 TLS 客户端密钥。必须与 clientCert 一起使用。可以是 secretKeyRef 以使用密钥引用。 参见下面的示例 qosN 指示消息的服务质量级别 (QoS)（更多信息 ）。默认为 1。 0, 1, 2
使用 TLS 进行通信 要配置使用 TLS 的通信，请确保 MQTT 代理（例如 emqx）配置为支持证书，并在组件配置中提供 caCert、clientCert、clientKey 元数据。例如：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : mqtt-pubsub
spec :
type : pubsub.mqtt3
version : v1
metadata :
- name : url
value : "ssl://host.domain[:port]"
# TLS configuration
- name : caCert
value : |
-----BEGIN CERTIFICATE-----
...
-----END CERTIFICATE-----
- name : clientCert
value : |
-----BEGIN CERTIFICATE-----
...
-----END CERTIFICATE-----
- name : clientKey
secretKeyRef :
name : myMqttClientKey
key : myMqttClientKey
# Optional
- name : retain
value : "false"
- name : cleanSession
value : "false"
- name : qos
value : 1
请注意，虽然 caCert 和 clientCert 值可能不是密钥，但为了方便起见，也可以从 Dapr 密钥存储中引用它们。
消费共享主题 在消费共享主题时，每个消费者必须具有唯一的标识符。默认情况下，应用 ID 用于唯一标识每个消费者和发布者。在自托管模式下，为每个 dapr run 调用使用不同的应用 ID 就足以使它们从同一个共享主题消费。然而，在 Kubernetes 上，应用 Pod 的多个实例将共享同一个应用 ID，从而阻止所有实例消费同一个主题。为了克服这一点，请使用 {uuid} 标签（这将在每个实例启动时为其生成一个随机生成的值）或 {podName}（这将使用 Kubernetes 上的 Pod 名称）配置组件的 consumerID 元数据。例如：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : mqtt-pubsub
spec :
type : pubsub.mqtt3
version : v1
metadata :
- name : consumerID
value : "{uuid}"
- name : cleanSession
value : "true"
- name : url
value : "tcp://admin:public@localhost:1883"
- name : qos
value : 1
- name : retain
value : "false"
Warning 上述示例将密钥作为纯字符串使用。建议按照
此处 所述使用密钥存储。
请注意，在这种情况下，消费者 ID 的值在每次 Dapr 重启时都是随机的，因此您还应该将 cleanSession 设置为 "true"。
建议将 StatefulSets 与共享订阅一起使用。
创建 MQTT3 代理 您可以使用 Docker 在本地运行 MQTT 代理（如 emqx） ：
docker run -d -p 1883:1883 --name mqtt emqx:latest
然后您可以使用客户端端口与服务器交互：tcp://localhost:1883
您可以使用以下 yaml 在 kubernetes 中运行 MQTT3 代理：
apiVersion : apps/v1
kind : Deployment
metadata :
name : mqtt-broker
labels :
app-name : mqtt-broker
spec :
replicas : 1
selector :
matchLabels :
app-name : mqtt-broker
template :
metadata :
labels :
app-name : mqtt-broker
spec :
containers :
- name : mqtt
image : emqx:latest
imagePullPolicy : IfNotPresent
ports :
- name : default
containerPort : 1883
protocol : TCP
---
apiVersion : v1
kind : Service
metadata :
name : mqtt-broker
labels :
app-name : mqtt-broker
spec :
type : ClusterIP
selector :
app-name : mqtt-broker
ports :
- port : 1883
targetPort : default
name : default
protocol : TCP
然后您可以使用客户端端口与服务器交互：tcp://mqtt-broker.default.svc.cluster.local:1883
相关链接 12 - Pulsar Pulsar 发布订阅组件的详细文档
组件格式 要设置 Apache Pulsar 发布订阅，需创建一个类型为 pubsub.pulsar 的组件。请参阅 发布订阅代理组件文件 以了解 ConsumerID 是如何自动生成的。阅读 操作指南：发布和订阅指南 以了解如何创建和应用发布订阅配置。
关于 Apache Pulsar 的更多信息，请阅读官方文档 。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : pulsar-pubsub
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 : consumerID
value : "channel1"
- name : namespace
value : "default"
- name : persistent
value : "true"
- name : disableBatching
value : "false"
- name : receiverQueueSize
value : "1000"
- name : <topic-name>.jsonschema # 为配置的主题设置 json schema 验证
value : |
{
"type": "record",
"name": "Example",
"namespace": "test",
"fields": [
{"name": "ID","type": "int"},
{"name": "Name","type": "string"}
]
}
- name : <topic-name>.avroschema # 为配置的主题设置 avro schema 验证
value : |
{
"type": "record",
"name": "Example",
"namespace": "test",
"fields": [
{"name": "ID","type": "int"},
{"name": "Name","type": "string"}
]
}
Warning 上面的示例将密钥作为纯字符串使用。建议使用
密钥存储来管理密钥 。此组件支持将
token 参数和任何其他敏感参数及数据存储为 Kubernetes Secrets。
规范元数据字段 Field Required Details Example host Y Pulsar broker 的地址。默认为 "localhost:6650" "localhost:6650" OR "http://pulsar-pj54qwwdpz4b-pulsar.ap-sg.public.pulsar.com:8080"enableTLS N 启用 TLS。 默认值："false" "true", "false"tenant N 实例内的主题租户。租户对于 Pulsar 中的多租户至关重要，并跨集群分布。 默认值："public" "public"consumerID N 用于设置订阅名称或消费者 ID。 可以设置为字符串值（如上面示例中的 "channel1"）或字符串格式值（如 "{podName}" 等）。查看您可以在组件元数据中使用的所有模板标签。 namespace N 主题的管理单元，充当相关主题的分组机制。 默认值："default" "default"persistent N Pulsar 支持两种主题：持久化 和非持久化 。对于持久化主题，所有消息都会持久化保存在磁盘上（如果 broker 不是独立的，消息会持久化保存在多个磁盘上），而非持久化主题的数据则不会持久化到存储磁盘。 "true", "false"disableBatching N 禁用批处理。启用批处理时，默认批处理延迟设置为 10 毫秒，默认批处理大小为 1000 条消息，设置 disableBatching: true 将使生产者单独发送消息。 默认值："false" "true", "false"receiverQueueSize N 设置消费者接收队列的大小。控制在 Dapr 显式调用读取消息之前，消费者可以累积多少消息。 默认值："1000" "1000"batchingMaxPublishDelay N batchingMaxPublishDelay 设置发送消息将被批处理的时间段（如果启用了批处理消息）。如果设置为非零值，消息将被排队，直到达到此时间间隔或 batchingMaxMessages（见下文）或 batchingMaxSize（见下文）。有两种有效格式，一种是带单位后缀的分数格式，另一种是作为毫秒处理的纯数字格式。有效的时间单位为 “ns”、“us”（或 “µs”）、“ms”、“s”、“m”、“h”。 默认值："10ms" "10ms", "10"batchingMaxMessages N batchingMaxMessages 设置批处理中允许的最大消息数。如果设置为大于 1 的值，消息将被排队，直到达到此阈值或 batchingMaxSize（见下文）已达到或批处理间隔已过去。 默认值："1000" "1000"batchingMaxSize N batchingMaxSize 设置批处理中允许的最大字节数。如果设置为大于 1 的值，消息将被排队，直到达到此阈值或 batchingMaxMessages（见上文）已达到或批处理间隔已过去。 默认值："128KB" "131072".jsonschema N 为配置的主题强制执行 JSON schema 验证。 .avroschema N 为配置的主题强制执行 Avro schema 验证。 publicKey N 用于发布者和消费者加密的公钥。值可以是以下两种选项之一：本地 PEM 证书的文件路径，或证书数据字符串值 privateKey N 用于消费者加密的私钥。值可以是以下两种选项之一：本地 PEM 证书的文件路径，或证书数据字符串值 keys N 包含 Pulsar 会话密钥 名称的逗号分隔字符串。与 publicKey 结合使用，用于发布者加密 processMode N 启用一次处理多条消息。 默认值："async" "async", "sync"subscribeType N Pulsar 支持四种订阅类型 。 默认值："shared" "shared", "exclusive", "failover", "key_shared"subscribeInitialPosition N 订阅位置是开始消费时光标设置的初始位置。 默认值："latest" "latest", "earliest"subscribeMode N 订阅模式指示光标持久化，持久化订阅保留消息并持久化当前位置。 默认值："durable" "durable", "non_durable"partitionKey N 设置用于路由策略的消息键。 默认值："" maxConcurrentHandlersN 定义并发消息处理程序的最大数量。 默认值：100 10replicateSubscriptionState N 启用跨地域复制的 Pulsar 集群的订阅状态复制。 默认值："false" "true", "false"
使用 Token 进行身份验证 要使用静态 JWT token 向 pulsar 进行身份验证，您可以使用以下元数据字段：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "pulsar.example.com:6650"
- name : token
secretKeyRef :
name : pulsar
key : token
使用 OIDC 进行身份验证 自 v3.0 起，Pulsar 支持 OIDC 身份验证 。
要启用 OIDC 身份验证，您需要向组件规范提供以下 OAuth2 参数。
OAuth2 身份验证不能与 token 身份验证结合使用。
建议您使用密钥引用来存储客户端密钥。
pulsar OAuth2 身份验证器并不专门符合 OIDC，因此您有责任确保字段符合要求。例如，颁发者 URL 必须使用 https 协议，请求的范围包括 openid 等。
如果省略 oauth2TokenCAPEM 字段，则在使用 https 连接到 OAuth2 颁发者时将使用系统的证书池。
注意： 元数据值会覆盖文件值。
Field Required Details Example oauth2CredentialsFile N 包含 client_id、client_secret、issuer_url 的 JSON 文件。使用此项 或 下面的单独字段。 "/path/to/credentials.json"oauth2TokenURL N 从中请求 OIDC client_credentials 令牌的 URL。如果不使用 oauth2CredentialsFile 则必需。 "https://oauth.example.com/token"oauth2ClientID N OIDC 客户端 ID。如果不使用 oauth2CredentialsFile 则必需。 "my-client-id"oauth2ClientSecret N OIDC 客户端密钥。如果使用 oauth2ClientID（而非 oauth2ClientSecretPath）则需要。 "my-client-secret"oauth2ClientSecretPath N 包含客户端密钥的纯文本文件。需要 oauth2ClientID 和 oauth2TokenURL。 "/path/to/client_secret.txt"oauth2TokenCAPEM N 用于连接到 OAuth2 颁发者的 CA PEM 证书束。如果未定义，将使用系统的证书池。 "---BEGIN CERTIFICATE---\n...\n---END CERTIFICATE---"oauth2Audiences N 请求的受众的逗号分隔列表。不能为空。 "my-audience-1,my-audience-2"oauth2Scopes N 请求的范围的逗号分隔列表。不能为空。 "openid,profile,email"
直接使用元数据字段 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "pulsar.example.com:6650"
- name : oauth2TokenURL
value : https://oauth.example.com/o/oauth2/token
- name : oauth2TokenCAPEM
value : "---BEGIN CERTIFICATE---\n...\n---END CERTIFICATE---"
- name : oauth2ClientID
value : my-client-id
- name : oauth2ClientSecret
secretKeyRef :
name : pulsar-oauth2
key : my-client-secret
- name : oauth2Audiences
value : "my.pulsar.example.com,another.pulsar.example.com"
- name : oauth2Scopes
value : "openid,profile,email"
- name : oauth2ClientSecretPath
value : "/path/to/oauth2/client_secret.json"
使用 JSON 凭据文件 您可以将凭据存储为具有以下格式的 JSON 文件：
{
"client_id" : "my-client-id" ,
"client_secret" : "my-client-secret" ,
"issuer_url" : "https://oauth.example.com/o/oauth2/token"
}
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "pulsar.example.com:6650"
- name : oauth2CredentialsFile
value : "/path/to/oauth2/credentials.json"
- name : oauth2TokenCAPEM
value : "---BEGIN CERTIFICATE---\n...\n---END CERTIFICATE---"
- name : oauth2Audiences
value : "my.pulsar.example.com,another.pulsar.example.com"
- name : oauth2Scopes
value : "openid,profile,email"
使用纯文本密钥文件 您可以将客户端密钥仅存储在纯文本文件中：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "pulsar.example.com:6650"
- name : oauth2TokenURL
value : https://oauth.example.com/o/oauth2/token
- name : oauth2ClientID
value : my-client-id
- name : oauth2ClientSecretPath
value : "/path/to/oauth2/client_secret.txt"
- name : oauth2TokenCAPEM
value : "---BEGIN CERTIFICATE---\n...\n---END CERTIFICATE---"
- name : oauth2Audiences
value : "my.pulsar.example.com,another.pulsar.example.com"
- name : oauth2Scopes
value : "openid,profile,email"
使用 JSON 凭据文件 您可以将凭据存储为具有以下格式的 JSON 文件：
{
"client_id" : "my-client-id" ,
"client_secret" : "my-client-secret" ,
"issuer_url" : "https://oauth.example.com/o/oauth2/token"
}
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "pulsar.example.com:6650"
- name : oauth2CredentialsFile
value : "/path/to/oauth2/credentials.json"
- name : oauth2TokenCAPEM
value : "---BEGIN CERTIFICATE---\n...\n---END CERTIFICATE---"
- name : oauth2Audiences
value : "my.pulsar.example.com,another.pulsar.example.com"
- name : oauth2Scopes
value : "openid,profile,email"
使用纯文本密钥文件 您可以将客户端密钥仅存储在纯文本文件中：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "pulsar.example.com:6650"
- name : oauth2TokenURL
value : https://oauth.example.com/o/oauth2/token
- name : oauth2ClientID
value : my-client-id
- name : oauth2ClientSecretPath
value : "/path/to/oauth2/client_secret.txt"
- name : oauth2TokenCAPEM
value : "---BEGIN CERTIFICATE---\n...\n---END CERTIFICATE---"
- name : oauth2Audiences
value : "my.pulsar.example.com,another.pulsar.example.com"
- name : oauth2Scopes
value : "openid,profile,email"
启用消息传递重试 Pulsar 发布订阅组件没有内置对重试策略的支持。这意味着边车仅向服务发送一次消息，并且在发生故障时不会重试。要使 Dapr 使用更复杂的重试策略，您可以将 重试弹性策略 应用于 Pulsar 发布订阅组件。请注意，将是同一个 Dapr 边车向同一应用实例重新传递消息，而不是其他实例。
延迟队列 调用 Pulsar 发布订阅时，可以通过在请求 url 中使用 metadata 查询参数来提供可选的延迟队列。
这些可选参数名称是 metadata.deliverAt 或 metadata.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' \
"Content-Type: application/json" \
'{
"data": {
"message": "Hi"
}
}'
或
curl -X POST http://localhost:3500/v1.0/publish/myPulsar/myTopic?metadata.deliverAfter= '4h5m3s' \
"Content-Type: application/json" \
'{
"data": {
"message": "Hi"
}
}'
启用消息压缩 消息压缩可以减小消息大小，代价是在发布期间稍微增加 CPU 使用率。压缩在生产者级别应用。
Compression Type Description none无压缩（默认） lz4LZ4 压缩 - 快速压缩/解压 zlibZLib 压缩 - 平衡的压缩比 zstdZSTD 压缩 - 高压缩比
Compression Level Description default所选类型的默认压缩级别 faster优先考虑速度而非压缩比 better优先考虑压缩比而非速度
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "localhost:6650"
- name : compressionType
value : lz4
- name : compressionLevel
value : faster
注意： 元数据键 compressionType 和 compressionLevel 区分大小写，必须完全按照所示方式指定。压缩在发布消息时应用；消费者无论设置如何都会自动解压。
端到端加密 Dapr 支持设置公钥和私钥对以启用 Pulsar 的端到端加密功能 。
从文件证书启用发布者加密 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "localhost:6650"
- name : publicKey
value : ./public.key
- name : keys
value : myapp.key
从文件证书启用消费者加密 apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "localhost:6650"
- name : publicKey
value : ./public.key
- name : privateKey
value : ./private.key
从值启用发布者加密 注意：建议从密钥引用公钥 。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "localhost:6650"
- name : publicKey
value : "-----BEGIN PUBLIC KEY-----\nMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA1KDAM4L8RtJ+nLaXBrBh\nzVpvTemsKVZoAct8A+ShepOHT9lgHOCGLFGWNla6K6j+b3AV/P/fAAhwj82vwTDd\nruXSflvSdmYeFAw3Ypphc1A5oM53wSRWhg63potBNWqdDzj8ApYgqjpmjYSQdL5/\na3golb36GYFrY0MLFTv7wZ87pmMIPsOgGIcPbCHker2fRZ34WXYLb1hkeUpwx4eK\njpwcg35gccvR6o/UhbKAuc60V1J9Wof2sNgtlRaQej45wnpjWYzZrIyk5qUbn0Qi\nCdpIrXvYtANq0Id6gP8zJvUEdPIgNuYxEmVCl9jI+8eGI6peD0qIt8U80hf9axhJ\n3QIDAQAB\n-----END PUBLIC KEY-----\n"
- name : keys
value : myapp.key
从值启用消费者加密 注意：建议从密钥引用公钥和私钥 。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : messagebus
spec :
type : pubsub.pulsar
version : v1
metadata :
- name : host
value : "localhost:6650"
- name : publicKey
value : "-----BEGIN PUBLIC KEY-----\nMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA1KDAM4L8RtJ+nLaXBrBh\nzVpvTemsKVZoAct8A+ShepOHT9lgHOCGLFGWNla6K6j+b3AV/P/fAAhwj82vwTDd\nruXSflvSdmYeFAw3Ypphc1A5oM53wSRWhg63potBNWqdDzj8ApYgqjpmjYSQdL5/\na3golb36GYFrY0MLFTv7wZ87pmMIPsOgGIcPbCHker2fRZ34WXYLb1hkeUpwx4eK\njpwcg35gccvR6o/UhbKAuc60V1J9Wof2sNgtlRaQej45wnpjWYzZrIyk5qUbn0Qi\nCdpIrXvYtANq0Id6gP8zJvUEdPIgNuYxEmVCl9jI+8eGI6peD0qIt8U80hf9axhJ\n3QIDAQAB\n-----END PUBLIC KEY-----\n"
- name : privateKey
value : "-----BEGIN RSA PRIVATE KEY-----\nMIIEpAIBAAKCAQEA1KDAM4L8RtJ+nLaXBrBhzVpvTemsKVZoAct8A+ShepOHT9lg\nHOCGLFGWNla6K6j+b3AV/P/fAAhwj82vwTDdruXSflvSdmYeFAw3Ypphc1A5oM53\nwSRWhg63potBNWqdDzj8ApYgqjpmjYSQdL5/a3golb36GYFrY0MLFTv7wZ87pmMI\nPsOgGIcPbCHker2fRZ34WXYLb1hkeUpwx4eKjpwcg35gccvR6o/UhbKAuc60V1J9\nWof2sNgtlRaQej45wnpjWYzZrIyk5qUbn0QiCdpIrXvYtANq0Id6gP8zJvUEdPIg\nNuYxEmVCl9jI+8eGI6peD0qIt8U80hf9axhJ3QIDAQABAoIBAQCKuHnM4ac/eXM7\nQPDVX1vfgyHc3hgBPCtNCHnXfGFRvFBqavKGxIElBvGOcBS0CWQ+Rg1Ca5kMx3TQ\njSweSYhH5A7pe3Sa5FK5V6MGxJvRhMSkQi/lJZUBjzaIBJA9jln7pXzdHx8ekE16\nBMPONr6g2dr4nuI9o67xKrtfViwRDGaG6eh7jIMlEqMMc6WqyhvI67rlVDSTHFKX\njlMcozJ3IT8BtTzKg2Tpy7ReVuJEpehum8yn1ZVdAnotBDJxI07DC1cbOP4M2fHM\ngfgPYWmchauZuTeTFu4hrlY5jg0/WLs6by8r/81+vX3QTNvejX9UdTHMSIfQdX82\nAfkCKUVhAoGBAOvGv+YXeTlPRcYC642x5iOyLQm+BiSX4jKtnyJiTU2s/qvvKkIu\nxAOk3OtniT9NaUAHEZE9tI71dDN6IgTLQlAcPCzkVh6Sc5eG0MObqOO7WOMCWBkI\nlaAKKBbd6cGDJkwGCJKnx0pxC9f8R4dw3fmXWgWAr8ENiekMuvjSfjZ5AoGBAObd\ns2L5uiUPTtpyh8WZ7rEvrun3djBhzi+d7rgxEGdditeiLQGKyZbDPMSMBuus/5wH\nwfi0xUq50RtYDbzQQdC3T/C20oHmZbjWK5mDaLRVzWS89YG/NT2Q8eZLBstKqxkx\ngoT77zoUDfRy+CWs1xvXzgxagD5Yg8/OrCuXOqWFAoGAPIw3r6ELknoXEvihASxU\nS4pwInZYIYGXpygLG8teyrnIVOMAWSqlT8JAsXtPNaBtjPHDwyazfZrvEmEk51JD\nX0tA8M5ah1NYt+r5JaKNxp3P/8wUT6lyszyoeubWJsnFRfSusuq/NRC+1+KDg/aq\nKnSBu7QGbm9JoT2RrmBv5RECgYBRn8Lj1I1muvHTNDkiuRj2VniOSirkUkA2/6y+\nPMKi+SS0tqcY63v4rNCYYTW1L7Yz8V44U5mJoQb4lvpMbolGhPljjxAAU3hVkItb\nvGVRlSCIZHKczADD4rJUDOS7DYxO3P1bjUN4kkyYx+lKUMDBHFzCa2D6Kgt4dobS\n5qYajQKBgQC7u7MFPkkEMqNqNGu5erytQkBq1v1Ipmf9rCi3iIj4XJLopxMgw0fx\n6jwcwNInl72KzoUBLnGQ9PKGVeBcgEgdI+a+tq+1TJo6Ta+hZSx+4AYiKY18eRKG\neNuER9NOcSVJ7Eqkcw4viCGyYDm2vgNV9HJ0VlAo3RDh8x5spEN+mg==\n-----END RSA PRIVATE KEY-----\n"
分区键 调用 Pulsar 发布订阅时，可以通过在请求 url 中使用 metadata 查询参数来提供可选的分区键。
参数名称是 partitionKey。
示例：
curl -X POST http://localhost:3500/v1.0/publish/myPlusar/myTopic?metadata.partitionKey= key1 \
"Content-Type: application/json" \
'{
"data": {
"message": "Hi"
}
}'
消息头 所有其他元数据键/值对（不是 partitionKey）都将在 Pulsar 消息中设置为头。例如，为消息设置 correlationId：
curl -X POST http://localhost:3500/v1.0/publish/myPlusar/myTopic?metadata.correlationId= myCorrelationID& metadata.partitionKey= key1 \
"Content-Type: application/json" \
'{
"data": {
"message": "Hi"
}
}'
顺序保证 要确保为订阅特定键的每个消费者按顺序到达消息，必须满足三个条件。
subscribeType 应设置为 key_shared。必须设置 partitionKey。 processMode 应设置为 sync。创建 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
相关链接 13 - RabbitMQ RabbitMQ 发布订阅组件的详细文档
组件格式 要设置 RabbitMQ 发布订阅，请创建类型为 pubsub.rabbitmq 的组件。请参阅发布订阅代理组件文件 以了解如何自动生成 ConsumerID。阅读如何：发布和订阅指南 以了解如何创建和应用发布订阅配置。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : rabbitmq-pubsub
spec :
type : pubsub.rabbitmq
version : v1
metadata :
- name : connectionString
value : "amqp://localhost:5672"
- name : protocol
value : amqp
- name : hostname
value : localhost
- name : username
value : username
- name : password
value : password
- name : consumerID
value : channel1
- name : durable
value : false
- name : deletedWhenUnused
value : false
- name : autoAck
value : false
- name : deliveryMode
value : 0
- name : requeueInFailure
value : false
- name : prefetchCount
value : 0
- name : reconnectWait
value : 0
- name : concurrencyMode
value : parallel
- name : publisherConfirm
value : false
- name : enableDeadLetter # Optional enable dead Letter or not
value : true
- name : maxLen # Optional max message count in a queue
value : 3000
- name : maxLenBytes # Optional maximum length in bytes of a queue.
value : 10485760
- name : exchangeKind
value : fanout
- name : saslExternal
value : false
- name : ttlInSeconds
value : 60
- name : clientName
value : {podName}
- name : heartBeat
value : 10s
- name : publishMessagePropertiesToMetadata
value : "true"
警告 上述示例使用纯文本字符串作为密钥。建议按照
此处 所述使用密钥存储来管理密钥。
规格元数据字段 字段 必填 详情 示例 connectionString Y* RabbitMQ 连接字符串。与 protocol、hostname、username、password 字段互斥 amqp://user:pass@localhost:5672protocol N* RabbitMQ 协议。与 connectionString 字段互斥 amqphostname N* RabbitMQ 主机名。与 connectionString 字段互斥 localhostusername N* RabbitMQ 用户名。与 connectionString 字段互斥 usernamepassword N* RabbitMQ 密码。与 connectionString 字段互斥 passwordconsumerID N 消费者 ID（消费者标签）将一个或多个消费者组织成一个组。具有相同消费者 ID 的消费者作为一个虚拟消费者工作；例如，一条消息只会被组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时将其设置为 Dapr 应用程序 ID（appID）的值。 可以设置为字符串值（如上例中的 "channel1"）或字符串格式的值（如 "{podName}" 等）。查看您可以在组件元数据中使用的所有模板标签 durable N 是否使用持久化 队列。默认为 "false" "true", "false"deletedWhenUnused N 是否将队列配置为自动删除 默认为 "true" "true", "false"autoAck N 队列消费者是否应自动确认 消息。默认为 "false" "true", "false"deliveryMode N 发布消息时的持久化模式。默认为 "0"。RabbitMQ 将 "2" 视为持久化，将所有其他数字视为非持久化 "0", "2"requeueInFailure N 在失败时发送否定确认 时是否重新排队。默认为 "false" "true", "false"prefetchCount N 要预取 的消息数量。考虑在生产环境中将其更改为非零值。默认为 "0"，这意味着将预取所有可用的消息。 "2"publisherConfirm N 如果启用，客户端在发布消息后会等待发布者确认 。默认为 "false" "true", "false"reconnectWait N 在连接失败时重新连接之前等待的时间（秒） "0"concurrencyMode N parallel 是默认值，允许并行处理多条消息（如果配置了，则受 app-max-concurrency 注解限制）。设置为 single 可禁用并行处理。在大多数情况下，没有必要更改此设置。parallel, singleenableDeadLetter N 启用将无法处理的消息转发到死信主题。默认为 "false" "true", "false"maxLen N 队列及其死信队列（如果启用了死信）的最大消息数。如果同时设置了 maxLen 和 maxLenBytes，则两者都将应用；先达到哪个限制将强制执行。默认为无限制。 "1000"maxLenBytes N 队列及其死信队列（如果启用了死信）的最大字节长度。如果同时设置了 maxLen 和 maxLenBytes，则两者都将应用；先达到哪个限制将强制执行。默认为无限制。 "1048576"exchangeKind N RabbitMQ 交换机的交换机类型。默认为 "fanout"。 "fanout","topic"saslExternal N 使用 TLS 时，是否应从附加字段（例如 CN）获取用户名。请参阅 RabbitMQ 身份验证机制 。默认为 "false"。 "true", "false"ttlInSeconds N 在组件级别设置消息 TTL，可以根据每个请求的消息级 TTL 覆盖。 "60"caCert 使用 TLS 时必需 用于验证服务器 TLS 证书的 PEM 格式证书颁发机构 (CA) 证书。 "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"clientCert 使用 TLS 时必需 PEM 格式的 TLS 客户端证书。必须与 clientKey 一起使用。 "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"clientKey 使用 TLS 时必需 PEM 格式的 TLS 客户端密钥。必须与 clientCert 一起使用。可以是 secretKeyRef 以使用密钥引用。 "-----BEGIN RSA PRIVATE KEY-----\n<base64-encoded PKCS8>\n-----END RSA PRIVATE KEY-----"clientName N 此 RabbitMQ 客户端提供的连接名称 是一个自定义标识符。如果设置，该标识符将在 RabbitMQ 服务器日志条目和管理 UI 中提及。可以设置为 {uuid}、{podName} 或 {appID}，Dapr 运行时会将其替换为实际值。 "app1", {uuid}, {podName}, {appID}heartBeat N 定义与服务器的心跳间隔，检测与 RabbitMQ 服务器的对等 TCP 连接的存活状态。默认为 10s。 "10s"publishMessagePropertiesToMetadataN 是否将 AMQP 消息属性（头部、消息 ID 等）发布到元数据。 “true”, “false”
使用 TLS 进行通信 要配置使用 TLS 进行通信，请确保 RabbitMQ 节点已启用 TLS，并在组件配置中提供 caCert、clientCert、clientKey 元数据。例如：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : rabbitmq-pubsub
spec :
type : pubsub.rabbitmq
version : v1
metadata :
- name : host
value : "amqps://localhost:5671"
- name : consumerID
value : myapp
- name : durable
value : false
- name : deletedWhenUnused
value : false
- name : autoAck
value : false
- name : deliveryMode
value : 0
- name : requeueInFailure
value : false
- name : prefetchCount
value : 0
- name : reconnectWait
value : 0
- name : concurrencyMode
value : parallel
- name : publisherConfirm
value : false
- name : enableDeadLetter # Optional enable dead Letter or not
value : true
- name : maxLen # Optional max message count in a queue
value : 3000
- name : maxLenBytes # Optional maximum length in bytes of a queue.
value : 10485760
- name : exchangeKind
value : fanout
- name : saslExternal
value : false
- name : caCert
value : ${{ myLoadedCACert }}
- name : clientCert
value : ${{ myLoadedClientCert }}
- name : clientKey
secretKeyRef :
name : myRabbitMQClientKey
key : myRabbitMQClientKey
注意，虽然 caCert 和 clientCert 值可能不是密钥，但为了方便起见，也可以从 Dapr 密钥存储中引用它们。
启用消息传递重试 RabbitMQ 发布订阅组件没有内置对重试策略的支持。这意味着边车只向服务发送一条消息。当服务返回结果时，无论消息是否被正确处理，该消息都将被标记为已消费。请注意，这是所有 Dapr PubSub 组件的常见行为，而不仅仅是 RabbitMQ。
当 autoAck 设置为 false 且 requeueInFailure 设置为 true 时，Dapr 可以尝试第二次重新传递消息。
要使 Dapr 使用更复杂的重试策略，您可以将重试弹性策略 应用于 RabbitMQ 发布订阅组件。
两种重试消息的方式之间存在关键区别：
当使用 autoAck = false 和 requeueInFailure = true 时，RabbitMQ 负责重新传递消息，任何 订阅者都可以获得重新传递的消息。如果您有多个消费者实例，那么另一个消费者可能会获得该消息。这通常是更好的方法，因为如果存在暂时性故障，不同的工作人员更有可能成功处理消息。
使用弹性功能使同一个 Dapr 边车重试重新传递消息。因此，它将是同一个 Dapr 边车和同一个应用程序接收同一条消息。
创建 RabbitMQ 服务器 您可以使用 Docker 在本地运行 RabbitMQ 服务器：
docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3
然后，您可以使用客户端端口与服务器交互：localhost:5672。
在 Kubernetes 上安装 RabbitMQ 的最简单方法是使用 Helm chart ：
helm install rabbitmq stable/rabbitmq
查看 chart 输出并获取用户名和密码。
这会将 RabbitMQ 安装到 default 命名空间中。要与 RabbitMQ 交互，请使用以下命令查找服务：kubectl get svc rabbitmq。
例如，如果使用上面的示例安装，RabbitMQ 服务器客户端地址将是：
rabbitmq.default.svc.cluster.local:5672
使用主题交换机路由消息 将 exchangeKind 设置为 "topic" 使用主题交换机，这通常用于消息的多播路由。要使用主题交换机路由消息，您必须设置以下元数据：
例如，如果应用程序配置了路由键 keyA 和 queueName 为 queue-A：
apiVersion : dapr.io/v2alpha1
kind : Subscription
metadata :
name : orderspubsub
spec :
topic : B
routes :
default : /B
pubsubname : pubsub
metadata :
routingKey : keyA
queueName : queue-A
它将接收路由键为 keyA 的消息，而不会接收具有其他路由键的消息。
// 使用路由键 `keyA` 发布消息，这些消息将被上面的示例接收。
client.PublishEvent(context.Background(), "pubsub", "B", []byte("this is a message"), dapr.PublishEventWithMetadata(map[string]string{"routingKey": "keyA"}))
// 使用路由键 `keyB` 发布消息，这些消息将不会被上面的示例接收。
client.PublishEvent(context.Background(), "pubsub", "B", []byte("this is another message"), dapr.PublishEventWithMetadata(map[string]string{"routingKey": "keyB"}))
绑定多个 routingKey 多个路由键可以用逗号分隔。routingKey：keyA、keyB 和 ""。注意空键的绑定方法。
apiVersion : dapr.io/v2alpha1
kind : Subscription
metadata :
name : orderspubsub
spec :
topic : B
routes :
default : /B
pubsubname : pubsub
metadata :
routingKey : keyA,keyB,
有关更多信息，请参阅 rabbitmq exchanges 。
使用优先级队列 Dapr 支持 RabbitMQ 优先级队列 。要为队列设置优先级，请使用 maxPriority 主题订阅元数据。
声明式优先级队列示例 apiVersion : dapr.io/v2alpha1
kind : Subscription
metadata :
name : pubsub
spec :
topic : checkout
routes :
default : /orders
pubsubname : order-pub-sub
metadata :
maxPriority : 3
编程式优先级队列示例 @app.route ( '/dapr/subscribe' , methods = [ 'GET' ])
def subscribe ():
subscriptions = [
{
'pubsubname' : 'pubsub' ,
'topic' : 'checkout' ,
'routes' : {
'default' : '/orders'
},
'metadata' : { 'maxPriority' : '3' }
}
]
return jsonify ( subscriptions )
const express = require ( 'express' )
const bodyParser = require ( 'body-parser' )
const app = express ()
app . use ( bodyParser . json ({ type : 'application/*+json' }));
const port = 3000
app . get ( '/dapr/subscribe' , ( req , res ) => {
res . json ([
{
pubsubname : "pubsub" ,
topic : "checkout" ,
routes : {
default : '/orders'
},
metadata : {
maxPriority : '3'
}
}
]);
})
package main
"encoding/json"
"net/http"
const appPort = 3000
type subscription struct {
PubsubName string `json:"pubsubname"`
Topic string `json:"topic"`
Metadata map [ string ] string `json:"metadata,omitempty"`
Routes routes `json:"routes"`
}
type routes struct {
Rules [] rule `json:"rules,omitempty"`
Default string `json:"default,omitempty"`
}
// This handles /dapr/subscribe
func configureSubscribeHandler ( w http . ResponseWriter , _ * http . Request ) {
t := [] subscription {
{
PubsubName : "pubsub" ,
Topic : "checkout" ,
Routes : routes {
Default : "/orders" ,
},
Metadata : map [ string ] string {
"maxPriority" : "3"
},
},
}
w . WriteHeader ( http . StatusOK )
json . NewEncoder ( w ). Encode ( t )
}
发布消息时设置优先级 要为消息设置优先级，请将发布元数据键 maxPriority 添加到发布端点或 SDK 方法。
HTTP API (Bash)
Python
JavaScript
Go curl -X POST http://localhost:3601/v1.0/publish/order-pub-sub/orders?metadata.priority= 3 -H "Content-Type: application/json" -d '{"orderId": "100"}'
with DaprClient () as client :
result = client . publish_event (
pubsub_name = PUBSUB_NAME ,
topic_name = TOPIC_NAME ,
data = json . dumps ( orderId ),
data_content_type = 'application/json' ,
metadata = { 'priority' : '3' })
await client . pubsub . publish ( PUBSUB_NAME , TOPIC_NAME , orderId , { 'priority' : '3' });
client . PublishEvent ( ctx , PUBSUB_NAME , TOPIC_NAME , [] byte ( strconv . Itoa ( orderId )), map [ string ] string { "priority" : "3" })
使用仲裁队列 默认情况下，Dapr 创建 classic 队列。要创建 quorum 队列，请将以下元数据添加到您的发布订阅订阅 中
apiVersion : dapr.io/v2alpha1
kind : Subscription
metadata :
name : pubsub
spec :
topic : checkout
routes :
default : /orders
pubsubname : order-pub-sub
metadata :
queueType : quorum
生存时间 您可以在消息级别或组件级别设置生存时间 (TTL) 值。使用组件规格中的 ttlInSeconds 字段设置默认的组件级 TTL。
注意 如果您同时设置了组件级和消息级 TTL，则默认的组件级 TTL 将被忽略，而使用消息级 TTL。单一活跃消费者 RabbitMQ 单一活跃消费者 设置确保一次只有一个消费者处理来自队列的消息，并在活跃消费者被取消或失败时切换到另一个注册的消费者。当消息必须按照到达队列的精确顺序消费并且不支持多实例分布式处理时，可能需要这种方法。
当此选项在队列上由 Dapr 启用时，Dapr 运行时的实例将成为单一活跃消费者。为了允许另一个应用程序实例在失败时接管，Dapr 运行时必须探测应用程序的健康状况 并取消订阅发布订阅组件。
注意 此模式将阻止应用程序扩展，因为只有一个实例可以处理负载。虽然这对于与遗留或敏感应用程序的 Dapr 集成可能很有趣，但如果您需要可扩展性，您应该考虑允许分布式处理的设计。apiVersion : dapr.io/v2alpha1
kind : Subscription
metadata :
name : pubsub
spec :
topic : orders
routes :
default : /orders
pubsubname : order-pub-sub
metadata :
singleActiveConsumer : "true"
将消息属性发布到元数据 要启用发布到元数据中的消息属性 ，请在组件规格中将 publishMessagePropertiesToMetadata 字段设置为 "true"。
这将包括消息 ID、时间戳和标头等属性在已发布消息的元数据中。
相关链接 14 - Redis Streams Redis Streams 发布订阅组件的详细文档
组件格式 要设置 Redis Streams 发布订阅，请创建一个类型为 pubsub.redis 的组件。请参阅发布订阅代理组件文件 以了解 ConsumerID 是如何自动生成的。阅读操作指南：发布和订阅 了解如何创建和应用发布订阅配置。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : redis-pubsub
spec :
type : pubsub.redis
version : v1
metadata :
- name : redisHost
value : localhost:6379
- name : redisPassword
value : "KeFg23!"
- name : consumerID
value : "channel1"
- name : useEntraID
value : "true"
- name : enableTLS
value : "false"
警告 上述示例将密钥作为纯字符串使用。建议按照
此处 所述使用密钥存储来管理密钥。
规范元数据字段 字段 必填 详情 示例 redisHost Y redis 主机的连接字符串。如果 "redisType" 为 "cluster"，则可以是多个以逗号分隔的主机或单个主机。使用 Redis Sentinel（"failover" 为 "true"）时，也可以提供多个 sentinel 地址，以逗号分隔。 localhost:6379, redis-master.default.svc.cluster.local:6379, sentinel1:26379,sentinel2:26379,sentinel3:26379redisPassword N Redis 主机的密码。无默认值。可以是 secretKeyRef 以使用密钥引用 "", "KeFg23!"redisUsername N Redis 主机的用户名。默认为空。请确保你的 redis 服务器版本为 6 或以上，并且已正确创建 acl 规则。 "", "default"consumerID N 消费者组 ID。 可设置为字符串值（如上例中的 "channel1"）或字符串格式值（如 "{podName}" 等）。查看您可以在组件元数据中使用的所有模板标签。 useEntraID N 为 Azure Cache for Redis 实现 EntraID 支持。在启用此功能之前：redisHost 名称必须以 "server:port" 的形式指定必须启用 TLS 创建 Redis 实例 > Azure Cache for Redis 下了解有关此设置的更多信息 "true", "false"enableTLS N 如果 Redis 实例支持使用公共证书的 TLS，可以配置为启用或禁用。默认为 "false" "true", "false"clientCert N 客户端证书的内容，用于需要客户端证书的 Redis 实例。必须与 clientKey 一起使用，并且 enableTLS 必须设置为 true。建议按照此处 所述使用密钥存储 "----BEGIN CERTIFICATE-----\nMIIC..."clientKey N 客户端私钥的内容，与 clientCert 一起用于身份验证。建议按照此处 所述使用密钥存储 "----BEGIN PRIVATE KEY-----\nMIIE..."redeliverInterval N 检查待重新传递消息的间隔时间。可以使用 Go 持续时间字符串（例如 “ms”、“s”、“m”）或毫秒数。默认为 "60s"。"0" 禁用重新传递。 "30s", "5000"processingTimeout N 消息在尝试重新传递之前必须待处理的时间量。可以使用 Go 持续时间字符串（例如 “ms”、“s”、“m”）或毫秒数。默认为 "15s"。"0" 禁用重新传递。 "60s", "600000"queueDepth N 用于处理的消息队列大小。默认为 "100"。 "1000"concurrency N 处理消息的并发工作线程数。默认为 "10"。 "15"redisType N redis 的类型。有两个有效值，一个是 "node" 用于单节点模式，另一个是 "cluster" 用于 redis 集群模式。默认为 "node"。 "cluster"redisDB N 连接到 redis 后选择的数据库。如果 "redisType" 为 "cluster"，则忽略此选项。默认为 "0"。 "0"redisMaxRetries N 放弃前重试命令的最大次数。默认为不重试失败的命令。 "5"redisMinRetryInterval N 每次重试之间 redis 命令的最小退避时间。默认为 "8ms"；"-1" 禁用退避。 "8ms"redisMaxRetryInterval N 每次重试之间 redis 命令的最大退避时间。默认为 "512ms"；"-1" 禁用退避。 "5s"dialTimeout N 建立新连接的拨号超时时间。默认为 "5s"。 "5s"readTimeout N socket 读取的超时时间，如果设置为 "0s"，读取将是阻塞的。如果达到超时，redis 命令将因超时而失败而不是阻塞。默认为 "0s"，"-1" 表示无超时。 "3s"writeTimeout N socket 写入的超时时间。如果达到超时，redis 命令将因超时而失败而不是阻塞。默认为 readTimeout。 "3s"poolSize N socket 连接的最大数量。默认为 runtime.NumCPU 报告的每个 CPU 10 个连接。 "20"poolTimeout N 如果所有连接都忙，客户端等待连接的时间，然后返回错误。默认为 readTimeout + 1 秒。 "5s"maxConnAge N 客户端关闭连接的连接时长。默认为不关闭旧连接。 "30m"minIdleConns N 保持打开的最小空闲连接数，以避免与新关联的性能下降。默认为 "0"。 "2"idleCheckFrequency N 空闲连接回收器进行空闲检查的频率。默认为 "1m"。"-1" 禁用空闲连接回收器。 "-1"idleTimeout N 客户端关闭空闲连接的时间量。应小于服务器的超时时间。默认为 "5m"。"-1" 禁用空闲超时检查。 "10m"failover N 启用故障转移配置的属性。需要设置 sentinelMasterName。启用后，redisHost 应包含 sentinel 地址。默认为 "false" "true", "false"sentinelMasterName N sentinel 主节点名称。请参阅 Redis Sentinel 文档 "", "mymaster"sentinelUsername N Redis Sentinel 的用户名。仅当 “failover” 为 true 且 Redis Sentinel 已启用身份验证时适用 "username"sentinelPassword N Redis Sentinel 的密码。仅当 “failover” 为 true 且 Redis Sentinel 已启用身份验证时适用 "password"maxLenApprox N 流中的最大项目数。当达到指定长度时，旧条目将自动驱逐，以使流保持恒定大小。默认为无限制。 "10000"streamTTL N 流条目的 TTL 持续时间。超过此持续时间的条目将被驱逐。这是一个近似值，因为它是通过使用 ‘~’ 修饰符的 Redis 流 MINID 修剪来实现的。实际保留可能包含略多于 TTL 严格定义的条目，因为 Redis 优化了修剪操作以提高效率，可能会保留一些额外的条目。 "30d"
创建 Redis 实例 Dapr 可以使用任何 Redis 实例——容器化的、在本地开发机器上运行的或托管的云服务，只要 Redis 的版本是 5.x 或 6.x。
自托管
Kubernetes
AWS
Azure
GCP Dapr CLI 将自动为您创建和设置 Redis Streams 实例。
当您运行 dapr init 时，Redis 实例将通过 Docker 安装，组件文件将在默认目录中创建。（$HOME/.dapr/components 目录 或 Windows 上的 %USERPROFILE%\.dapr\components）。
您可以使用 Helm 在我们的 Kubernetes 集群中快速创建 Redis 实例。此方法需要安装 Helm 。
将 Redis 安装到您的集群中。
helm repo add bitnami https://charts.bitnami.com/bitnami
helm install redis bitnami/redis --set image.tag= 6.2
运行 kubectl get pods 以查看现在集群中运行的 Redis 容器。
在 redis.yaml 文件中添加 redis-master:6379 作为 redisHost。例如：
metadata :
- name : redisHost
value : redis-master:6379
接下来，我们将获取 Redis 密码，具体取决于我们使用的操作系统：
Windows ：运行 kubectl get secret --namespace default redis -o jsonpath="{.data.redis-password}" > encoded.b64，这将创建一个包含编码密码的文件。接下来，运行 certutil -decode encoded.b64 password.txt，这将把您的 redis 密码放入名为 password.txt 的文本文件中。复制密码并删除这两个文件。
Linux/MacOS ：运行 kubectl get secret --namespace default redis -o jsonpath="{.data.redis-password}" | base64 --decode 并复制输出的密码。
将此密码作为 redisPassword 值添加到您的 redis.yaml 文件中。例如：
- name : redisPassword
value : "lhDOkwTlp0"
使用官方 Microsoft 文档创建 Azure Cache for Redis 实例。
实例创建后，从 Azure 门户获取主机名（FQDN）和您的访问密钥。
将您的密钥和主机名添加到 Dapr 可以应用到集群的 redis.yaml 文件中。
如果您正在运行示例，请将主机和密钥添加到提供的 redis.yaml 中。 如果您从头开始创建项目，请按照组件格式部分 中的说明创建 redis.yaml 文件。 将 redisHost 键设置为 [上一步的主机名]:6379，将 redisPassword 键设置为您之前保存的密钥。
**注意：**在生产级应用程序中，请遵循密钥管理 说明来安全地管理您的密钥。
启用 EntraID 支持：
在您的 Azure Redis 服务器上启用 Entra ID 身份验证。这可能需要几分钟。 将 useEntraID 设置为 "true" 以实现 Azure Cache for Redis 的 EntraID 支持。 将 enableTLS 设置为 "true" 以支持 TLS。
注意： useEntraID 假定您的 UserPrincipal（通过 AzureCLICredential）或系统分配的托管标识具有 RedisDataOwner 角色权限。如果使用用户分配的标识，您需要指定 azureClientID 属性 。
注意 Dapr CLI 在自托管模式下作为 dapr init 命令的一部分自动部署本地 redis 实例。Redis Sentinel 配置 使用 Redis Sentinel 实现高可用性时，将 redisType 设置为 "node"，使用 failover: "true" 启用故障转移模式，并提供 sentinel 主节点名称。可以在 redisHost 字段中指定多个以逗号分隔的 sentinel 地址以实现冗余。
```yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: redis-pubsub
spec:
type: pubsub.redis
version: v1
metadata:
- name: redisHost
value: "sentinel1:26379,sentinel2:26379,sentinel3:26379"
- name: redisType
value: "node"
- name: failover
value: "true"
- name: sentinelMasterName
value: "mymaster"
```
弹性和重新传递 Redis Streams 发布订阅组件遵循 Dapr 弹性策略 。在您的应用程序处理程序返回（成功、重试或丢弃）后，组件向 Redis 确认消息。Redis 然后停止重新传递该消息。重试和死信行为由您的弹性策略控制（例如 maxRetries 和重试持续时间）。
相关链接 15 - RocketMQ RocketMQ 发布订阅组件的详细文档
组件格式 要设置 RocketMQ 发布订阅，需创建一个类型为 pubsub.rocketmq 的组件。请参阅发布订阅代理组件文件 以了解 ConsumerID 如何自动生成。阅读如何操作：发布和订阅指南 了解如何创建和应用发布订阅配置。
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
警告 上述示例将密钥作为纯字符串使用。建议按照
此处 所述使用密钥存储来管理密钥。
规范元数据字段 字段 必填 详情 默认值 示例 instanceName N 实例名称 time.Now().String()dapr-rocketmq-testconsumerGroup N 消费者组名称。推荐。如果 producerGroup 为 null，则使用 groupName。 dapr-rocketmq-test-g-cproducerGroup (consumerID) N 生产者组名称。推荐。如果 producerGroup 为 null，则使用 consumerID。如果 consumerID 也为 null，则使用 groupName。 dapr-rocketmq-test-g-pconsumerID N 消费者 ID（消费者标签）将一个或多个消费者组织到一个组中。具有相同消费者 ID 的消费者作为一个虚拟消费者工作；例如，一条消息仅由组中的一个消费者处理一次。如果未提供 consumerID，Dapr 运行时会将其设置为 Dapr 应用程序 ID（appID）的值。 可设置为字符串值（如上面示例中的 "channel1"）或字符串格式值（如 "{podName}" 等）。查看您可以在组件元数据中使用的所有模板标签。 groupName N 消费者/生产者组名称。已弃用 。 dapr-rocketmq-test-gnameSpace N RocketMQ 命名空间 dapr-rocketmqnameServerDomain N RocketMQ 名称服务器域名 https://my-app.net:8080/nsaddrnameServer N RocketMQ 名称服务器，用 “,” 或 “;” 分隔 127.0.0.1:9876;127.0.0.2:9877,127.0.0.3:9877accessKey N 访问密钥（用户名） "admin"secretKey N 密钥（密码） "password"securityToken N 安全令牌 retries N 向代理发送消息的重试次数 33producerQueueSelector (queueSelector) N 生产者队列选择器。队列选择器有五种实现：hash、random、manual、roundRobin、dapr。 daprhashconsumerModel N 定义消息如何传递给每个消费者客户端的消息模型。RocketMQ 支持两种消息模型：clustering 和 broadcasting。 clusteringbroadcasting、clusteringfromWhere (consumeFromWhere) N 消费者启动时的消费点。有三个消费点：CONSUME_FROM_LAST_OFFSET、CONSUME_FROM_FIRST_OFFSET、CONSUME_FROM_TIMESTAMP CONSUME_FROM_LAST_OFFSETCONSUME_FROM_LAST_OFFSETconsumeTimestamp N 以秒精度回溯消费时间。时间格式为 yyyymmddhhmmss。例如，20131223171201 表示时间为 17:12:01，日期为 2013 年 12 月 23 日 time.Now().Add(time.Minute * (-30)).Format("20060102150405")20131223171201consumeOrderly N 确定是否使用 FIFO 顺序的有序消息。 falsefalseconsumeMessageBatchMaxSize N 批量消费大小，超出范围 [1, 1024] 51210consumeConcurrentlyMaxSpan N 并发最大跨度偏移量。这对顺序消费没有影响。范围：[1, 65535] 10001000maxReconsumeTimes N 最大重新消费次数。-1 表示 16 次。如果消息在成功之前重新消费的次数超过 {@link maxReconsumeTimes}，它们将被定向到删除队列。 顺序消息为 MaxInt32；并发消息为 16 16autoCommit N 启用自动提交 truefalseconsumeTimeout N 消息可能阻塞消费线程的最长时间。时间单位：分钟 1515consumerPullTimeout N Socket 超时时间（毫秒） pullInterval N 消息拉取间隔 100100pullBatchSize N 一次从代理拉取的消息数量。如果 pullBatchSize 为 null，则使用 ConsumerBatchSize。pullBatchSize 超出范围 [1, 1024] 3210pullThresholdForQueue N 队列级别的流控阈值。每个消息队列默认最多缓存 1000 条消息。考虑 PullBatchSize - 瞬时值可能会超过限制。范围：[1, 65535] 10241000pullThresholdForTopic N 主题级别的流控阈值。如果 pullThresholdForTopic 不受限，pullThresholdForQueue 的值将被覆盖并基于 pullThresholdForTopic 计算。例如，如果 pullThresholdForTopic 的值为 1000，并且为此消费者分配了 10 个消息队列，那么 pullThresholdForQueue 将被设置为 100。范围：[1, 6553500] -1(不限)10pullThresholdSizeForQueue N 限制队列级别的缓存消息大小。考虑 pullBatchSize - 瞬时值可能会超过限制。消息的大小仅通过消息体测量，因此不准确。范围：[1, 1024] 100100pullThresholdSizeForTopic N 限制主题级别的缓存消息大小。如果 pullThresholdSizeForTopic 不受限，pullThresholdSizeForQueue 的值将被覆盖并基于 pullThresholdSizeForTopic 计算。例如，如果 pullThresholdSizeForTopic 的值为 1000 MiB，并且为此消费者分配了 10 个消息队列，那么 pullThresholdSizeForQueue 将被设置为 100 MiB。范围：[1, 102400] -1100content-type N 消息内容类型。 "text/plain""application/cloudevents+json; charset=utf-8"、"application/octet-stream"logLevel N 日志级别 warninfosendTimeOut N 向 RocketMQ 的代理发送消息的超时时间，以纳秒为单位测量。已弃用 。 3 秒 10000000000sendTimeOutSec N 发布消息的超时时长（秒）。如果 sendTimeOutSec 为 null，则使用 sendTimeOut。 3 秒 3mspProperties N 此集合中的 RocketMQ 消息属性在数据分离中传递给 APP。用 “,” 分隔多个属性 key,mkey
出于向后兼容的原因，元数据中支持以下值，但不鼓励使用。
字段（支持但已弃用） 必填 详情 示例 groupName N RocketMQ 发布者的生产者组名称 "my_unique_group_name"sendTimeOut N 发布消息的超时时长（纳秒） 0consumerBatchSize N 一次从代理拉取的消息数量 32
设置 RocketMQ 请参阅 https://rocketmq.apache.org/docs/quick-start/ 以设置本地 RocketMQ 实例。
每次调用元数据字段 分区键 调用 RocketMQ 发布订阅时，可以通过在请求 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 \
"Content-Type: application/json" \
'{
"data": {
"message": "Hi"
}
}'
QueueSelector RocketMQ 组件总共包含五个队列选择器。RocketMQ 客户端提供以下队列选择器：
HashQueueSelectorRandomQueueSelectorRoundRobinQueueSelectorManualQueueSelector要了解有关这些 RocketMQ 客户端队列选择器的更多信息，请阅读 RocketMQ 文档 。
Dapr RocketMQ 组件实现了以下队列选择器：
本文重点介绍 DaprQueueSelector 的设计。
DaprQueueSelector DaprQueueSelector 集成了三个队列选择器：
HashQueueSelectorRoundRobinQueueSelectorManualQueueSelectorDaprQueueSelector 从请求参数中获取队列 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。
相关链接 16 - Solace-AMQP Solace-AMQP 发布订阅组件的详细文档
组件格式 要设置 Solace-AMQP 发布订阅，需创建类型为 pubsub.solace.amqp 的组件。请参阅 发布订阅代理组件文件 以了解 ConsumerID 的自动生成方式。阅读 操作指南：发布与订阅 以了解如何创建并应用发布订阅配置。
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : solace
spec :
type : pubsub.solace.amqp
version : v1
metadata :
- name : url
value : 'amqp://localhost:5672'
- name : username
value : 'default'
- name : password
value : 'default'
- name : consumerID
value : 'channel1'
警告 上述示例将密钥以纯文本形式使用。建议使用密钥存储来管理密钥，如
此处 所述。
规范元数据字段 字段 必填 详情 示例 url Y AMQP 代理的地址。可使用 secretKeyRef 引用密钥。amqp://amqps:// "amqp://host.domain[:port]"username Y 连接到代理的用户名。仅在未指定 anonymous 或其设置为 false 时必需。 defaultpassword Y 连接到代理的密码。仅在未指定 anonymous 或其设置为 false 时必需。 defaultconsumerID N 消费者 ID（consumer tag）将一个或多个消费者组织成一个组。相同消费者 ID 的消费者作为单一虚拟消费者协作；例如，一条消息仅由组内的一个消费者处理一次。若未提供 consumerID，Dapr 运行时会将其设置为 Dapr 应用程序 ID（appID）的值。 可设置为字符串值（如上例中的 "channel1"）或字符串格式值（如 "{podName}" 等）。查看组件元数据中可使用的所有模板标签。 anonymous N 在不验证凭据的情况下连接到代理。仅在代理上启用时有效。若设置为 true，则无需用户名和密码。 truecaCert 使用 TLS 时必需 用于验证服务器 TLS 证书的 PEM 格式证书颁发机构（CA）证书。 "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"clientCert 使用 TLS 时必需 PEM 格式的 TLS 客户端证书。必须与 clientKey 一起使用。 "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"clientKey 使用 TLS 时必需 PEM 格式的 TLS 客户端私钥。必须与 clientCert 一起使用。可使用 secretKeyRef 引用密钥。 "-----BEGIN RSA PRIVATE KEY-----\n<base64-encoded PKCS8>\n-----END RSA PRIVATE KEY-----"
使用 TLS 进行通信 要配置使用 TLS 的通信：
确保 Solace 代理配置为支持证书。 在组件配置中提供 caCert、clientCert 和 clientKey 元数据。 例如：
apiVersion : dapr.io/v1alpha1
kind : Component
metadata :
name : solace
spec :
type : pubsub.solace.amqp
version : v1
metadata :
- name : url
value : "amqps://host.domain[:port]"
- name : username
value : 'default'
- name : password
value : 'default'
- name : caCert
value : ${{ myLoadedCACert }}
- name : clientCert
value : ${{ myLoadedClientCert }}
- name : clientKey
secretKeyRef :
name : mySolaceClientKey
key : mySolaceClientKey
auth :
secretStore : <SECRET_STORE_NAME>
虽然 caCert 和 clientCert 值可能不是密钥，但为方便起见，它们也可以从 Dapr 密钥存储中引用。
向主题和队列发布/订阅消息 默认情况下，消息通过主题进行发布和订阅。如果您希望目标是队列，请在主题前加上 queue: 前缀，Solace AMQP 组件将连接到队列。
创建 Solace 代理 您可以使用 Docker 在本地运行 Solace 代理：
docker run -d -p 8080:8080 -p 55554:55555 -p 8008:8008 -p 1883:1883 -p 8000:8000 -p 5672:5672 -p 9000:9000 -p 2222:2222 --shm-size= 2g --env username_admin_globalaccesslevel = admin --env username_admin_password = admin --name= solace solace/solace-pubsub-standard
然后，您可以使用客户端端口与服务器交互：mqtt://localhost:5672
您还可以在 Solace Cloud 上注册免费的 SaaS 代理。
相关链接