This is the multi-page printable view of this section. Click here to print.

Return to the regular view of this page.

发布与订阅消息

服务间安全、可扩展的消息传递

1: 发布订阅概述
2: 操作指南：发布消息并订阅主题
3: 使用 CloudEvents 发布和订阅消息
4: 不使用 CloudEvents 发布和订阅消息
5: 操作指南：将消息路由到不同的事件处理程序
6: 声明式、流式和编程式订阅类型
7: 死信主题
8: 如何操作：设置发布订阅命名空间消费者组
9: 如何：通过 StatefulSet 水平扩展订阅者
10: 限制发布订阅主题访问
11: 消息生存时间 (TTL)
12: 批量发布和订阅消息

1 - 发布订阅概述

发布订阅 API 构建块概述

发布订阅（pub/sub）使微服务能够使用消息进行通信，实现事件驱动架构。

生产者（发布者）将消息写入输入通道并发送到主题，但不知道哪个应用程序会接收它们。
消费者（订阅者）订阅主题并从输出通道接收消息，但不知道这些消息是由哪个服务产生的。

中间消息代理将每条消息从发布者的输入通道复制到所有对该消息感兴趣的订阅者的输出通道。当你需要将微服务相互解耦时，这种模式特别有用。

发布订阅 API

Dapr 中的发布订阅 API：

提供与平台无关的 API 来发送和接收消息。
提供至少一次消息传递保证。
集成各种消息代理和排队系统。

服务使用的特定消息代理是可插拔的，在运行时配置为 Dapr 发布订阅组件。这消除了服务对消息代理的依赖，使服务更具可移植性和灵活性。

使用 Dapr 中的发布订阅时：

你的服务向 Dapr 发布订阅构建块 API 发起网络调用。
发布订阅构建块调用封装了特定消息代理的 Dapr 发布订阅组件。
为了接收主题上的消息，Dapr 代表你的服务订阅发布订阅组件上的主题，并在消息到达时将消息传递到你的服务端点。

以下概述视频和演示演示了 Dapr 发布订阅的工作原理。

在下图中，“shipping"服务和 “email” 服务都订阅了 “cart” 服务发布的主题。每个服务加载指向同一发布订阅消息代理组件的发布订阅组件配置文件；例如：Redis Streams、NATS Streaming、Azure Service Bus 或 GCP pub/sub。

在下图中，Dapr API 将来自发布 “cart” 服务的 “order” 主题发布到订阅服务 “shipping” 和 “email” 的 “order” 端点。

查看 Dapr 支持的完整发布订阅组件列表。

功能

发布订阅 API 构建块为你的应用程序带来多项功能。

使用 CloudEvents 发送消息

为了实现消息路由并在服务之间提供每个消息的额外上下文，Dapr 使用 CloudEvents 1.0 规范作为其消息格式。任何使用 Dapr 发送到主题的应用程序消息都会自动包装在 Cloud Events 信封中，使用 Content-Type 头值作为 datacontenttype 属性。

与未使用 Dapr 和 CloudEvents 的应用程序通信

如果你的一个应用程序使用 Dapr 而另一个不使用，你可以为发布者或订阅者禁用 CloudEvent 包装。这允许在无法同时采用 Dapr 的应用程序中部分采用 Dapr 发布订阅。

更多信息，请阅读如何在不使用 CloudEvents 的情况下使用发布订阅。

设置消息内容类型

发布消息时，指定正在发送的数据的内容类型非常重要。除非指定，否则 Dapr 将假定为 text/plain。

HTTP 客户端：可以在 Content-Type 头中设置内容类型
gRPC 客户端和 SDK：有专用的内容类型参数

消息传递

原则上，Dapr 认为订阅者处理消息并返回非错误响应后，消息即成功传递。为了更精细的控制，Dapr 的发布订阅 API 还提供明确的状态（在响应负载中定义），订阅者使用这些状态向 Dapr 指示特定的处理指令（例如，RETRY 或 DROP）。

使用主题订阅接收消息

Dapr 应用程序可以通过支持相同功能的三种订阅类型订阅已发布的主题：声明式、流式和编程式。

订阅类型	描述
声明式	订阅在外部文件中定义。声明式方法从代码中移除 Dapr 依赖，允许现有应用程序订阅主题，而无需更改代码。
流式	订阅在用户代码中定义。流式订阅是动态的，意味着它们允许在运行时添加或移除订阅。它们不需要应用程序中的订阅端点（这是编程式和声明式订阅都需要的），使得在代码中配置变得简单。流式订阅也不需要应用程序配置边车来接收消息。使用流式订阅，由于消息被发送到消息处理程序代码，因此没有路由或批量订阅的概念。
编程式	订阅在用户代码中定义。编程式方法实现静态订阅，需要代码中的端点。

更多信息，请阅读关于订阅类型的订阅。

重新加载主题订阅

要重新加载以编程方式或声明方式定义的主题订阅，需要重启 Dapr 边车。通过启用 HotReload 特性门控可以使 Dapr 边车动态重新加载已更改的声明式主题订阅，而无需重启。主题订阅的热重载目前是预览功能。重新加载订阅时，正在传输的消息不受影响。

消息路由

Dapr 提供基于内容的路由模式。发布订阅路由是此模式的实现，允许开发者使用表达式根据 CloudEvents 的内容将消息路由到应用程序中的不同 URI/路径和事件处理程序。如果没有路由匹配，则使用可选的默认路由。随着你的应用程序扩展以支持多个事件版本或特殊情况，这非常有用。

此功能适用于声明式和编程式订阅方法。

有关消息路由的更多信息，请阅读 Dapr 发布订阅 API 参考。

使用死信主题处理失败消息

有时，消息由于多种可能的问题无法处理，例如生产者或消费者应用程序内部的错误条件或导致应用程序代码出现问题的意外状态更改。Dapr 允许开发者设置死信主题来处理无法传递到应用程序的消息。此功能在所有发布订阅组件上可用，可防止消费者应用程序无休止地重试失败的消息。更多信息，请阅读关于死信主题.

启用发件箱模式

Dapr 使开发者能够使用发件箱模式在事务性状态存储和任何消息代理之间实现单一事务。更多信息，请阅读如何启用事务性发件箱消息传递。

命名空间消费者组

Dapr 通过消费者组命名空间在大规模场景下解决多租户问题。只需在组件元数据中包含 "{namespace}" 值，即可让具有相同 app-id 的多个命名空间的应用程序发布和订阅到同一消息代理。

至少一次保证

Dapr 保证消息传递的至少一次语义。当应用程序使用发布订阅 API 向主题发布消息时，Dapr 确保消息至少一次传递到每个订阅者。

即使消息传递失败或应用程序崩溃，Dapr 也会重试重新传递消息直到成功传递。

所有 Dapr 发布订阅组件都支持至少一次保证。

订阅启动可靠性

Dapr 自动重试失败的订阅启动，以提高部署场景下的可靠性。这确保了你的发布订阅应用程序即使在面临临时连接或权限问题时也能保持弹性。

当 Dapr 遇到启动订阅的错误时，它会在日志中显示错误消息并继续尝试启动订阅。

消费者组和竞争消费者模式

Dapr 处理消费者组和竞争消费者模式的负担。在竞争消费者模式中，使用单个消费者组的多个应用程序实例竞争消息。当副本使用相同的 app-id 而没有明确的消费者组覆盖时，Dapr 强制执行竞争消费者模式。

当同一应用程序的多个实例（具有相同的 app-id）订阅主题时，Dapr 将每条消息传递到该应用程序的只有一个实例。此概念如下图所示。

同样，如果两个不同的应用程序（具有不同的 app-id）订阅同一主题，Dapr 将每条消息传递到每个应用程序的只有一个实例。

并非所有 Dapr 发布订阅组件都支持竞争消费者模式。目前，以下（非详尽）发布订阅组件支持此功能：

限制主题以增强安全性

默认情况下，与发布订阅组件实例关联的所有主题消息都可用于使用该组件配置的每个应用程序。你可以使用 Dapr 主题限制来限制哪些应用程序可以发布或订阅主题。更多信息，请阅读：发布订阅主题限制。

消息生存时间（TTL）

Dapr 可以按消息设置超时消息，这意味着如果消息未从发布订阅组件中读取，则消息将被丢弃。此超时消息可防止未读消息堆积。如果消息在队列中停留的时间超过配置的 TTL，它将被标记为死信。更多信息，请阅读发布订阅消息 TTL。

发布和订阅批量消息

Dapr 支持在单个请求中发送和接收多条消息。编写需要发送或接收大量消息的应用程序时，使用批量操作可以减少请求总数来实现高吞吐量。更多信息，请阅读发布订阅批量消息。

使用 StatefulSets 扩展订阅者

在 Kubernetes 上运行时，订阅者结合 StatefulSets 和 {podName} 标记可以拥有每个实例的粘性 consumerID。请参阅如何使用 StatefulSets 水平扩展订阅者。

尝试发布订阅

快速入门和教程

想测试 Dapr 发布订阅 API？请完成以下快速入门和教程来了解发布订阅的实际应用：

快速入门/教程	描述
发布订阅快速入门	使用发布订阅 API 发送和接收消息。
发布订阅教程	演示如何使用 Dapr 启用发布订阅应用程序。使用 Redis 作为发布订阅组件。

直接在应用程序中使用发布订阅

想跳过快速入门？没问题。你可以直接在应用程序中试用发布订阅构建块来发布消息和订阅主题。安装 Dapr 后，你可以从发布订阅操作指南开始使用发布订阅 API。

下一步

了解使用 CloudEvents 进行消息传递以及何时可能需要发送不带 CloudEvents 的消息。
遵循操作指南：使用多个命名空间配置发布订阅组件。
查看发布订阅组件列表。
阅读 API 参考。

2 - 操作指南：发布消息并订阅主题

了解如何通过一个服务向主题发送消息，并在另一个服务中订阅该主题

既然您已经了解了 Dapr 发布订阅构建块提供的能力，接下来了解如何在您的服务中使用它。下面的代码示例简单描述了一个包含两个服务的订单处理应用程序，每个服务都配置了 Dapr 边车：

一个结账服务，使用 Dapr 订阅消息队列中的主题。
一个订单处理服务，使用 Dapr 向 RabbitMQ 发布消息。

Diagram showing state management of example service

Dapr 会自动使用 Content-Type 头部的值作为 datacontenttype 属性，将用户负载封装在一个符合 CloudEvents v1.0 标准的信封中。了解更多关于 CloudEvents 消息的信息。

下面的示例演示了您的应用程序如何发布和订阅名为 orders 的主题。

注意

如果您还没有尝试过，请先体验发布订阅快速入门，快速了解如何使用发布订阅。

设置发布/订阅组件

第一步是设置发布/订阅组件：

当您运行 dapr init 时，Dapr 会创建一个默认的 Redis pubsub.yaml 并在您的本地机器上运行一个 Redis 容器，位于：

在 Windows 上，位于 %UserProfile%\.dapr\components\pubsub.yaml
在 Linux/MacOS 上，位于 ~/.dapr/components/pubsub.yaml

通过 pubsub.yaml 组件，您可以轻松更换底层组件而无需更改应用程序代码。在本示例中，使用的是 RabbitMQ。

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: order-pub-sub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: host
    value: "amqp://localhost:5672"
  - name: durable
    value: "false"
  - name: deletedWhenUnused
    value: "false"
  - name: autoAck
    value: "false"
  - name: reconnectWait
    value: "0"
  - name: concurrency
    value: parallel
scopes:
  - orderprocessing
  - checkout

您可以通过创建一个包含该文件的组件目录（在本例中为 myComponents）并在 dapr run CLI 命令中使用 --resources-path 标志，用另一个发布订阅组件覆盖此文件。

要将其部署到 Kubernetes 集群中，请填写下面 YAML 中发布/订阅组件的 metadata 连接详情，将其保存为 pubsub.yaml，然后运行 kubectl apply -f pubsub.yaml。

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: order-pub-sub
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: durable
    value: "false"
  - name: deletedWhenUnused
    value: "false"
  - name: autoAck
    value: "false"
  - name: reconnectWait
    value: "0"
  - name: concurrency
    value: parallel
scopes:
  - orderprocessing
  - checkout

dapr run --app-id myapp --resources-path ./myComponents -- dotnet run

dapr run --app-id myapp --resources-path ./myComponents -- mvn spring-boot:run

dapr run --app-id myapp --resources-path ./myComponents -- python3 app.py

dapr run --app-id myapp --resources-path ./myComponents -- go run app.go

dapr run --app-id myapp --resources-path ./myComponents -- npm start

订阅主题

Dapr 提供了三种订阅主题的方法：

声明式，订阅在外部文件中定义。
流式，订阅在用户代码中定义。
编程式，订阅在用户代码中定义。

在声明式、流式和编程式订阅文档中了解更多信息。本示例演示声明式订阅。

创建一个名为 subscription.yaml 的文件并粘贴以下内容：

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: order-pub-sub
spec:
  topic: orders
  routes: 
    default: /checkout
  pubsubname: order-pub-sub
scopes:
- orderprocessing
- checkout

上面的示例展示了对 orders 主题的事件订阅，使用的发布订阅组件是 order-pub-sub。

route 字段告诉 Dapr 将所有主题消息发送到应用程序中的 /checkout 端点。
scopes 字段为 ID 为 orderprocessing 和 checkout 的应用程序启用此订阅。

将 subscription.yaml 放在与 pubsub.yaml 组件相同的目录中。当 Dapr 启动时，它会与组件一起加载订阅。

注意

此功能目前处于预览状态。可以使 Dapr “热重载” 声明式订阅，从而自动应用更新而无需重启。这是通过 HotReload 功能门控启用的。为了防止重新处理或丢失未处理的消息，在热重载事件期间，Dapr 与您的应用程序之间正在传输的消息不受影响。

以下是利用 Dapr SDK 订阅您在 subscription.yaml 中定义的主题的代码示例。

using System.Collections.Generic;
using System.Threading.Tasks;
using System;
using Microsoft.AspNetCore.Mvc;
using Dapr;
using Dapr.Client;

namespace CheckoutService.Controllers;

[ApiController]
public sealed class CheckoutServiceController : ControllerBase
{
    //从 "order-pub-sub" 组件订阅名为 "orders" 的主题 
    [Topic("order-pub-sub", "orders")]
    [HttpPost("checkout")]
    public void GetCheckout([FromBody] int orderId)
    {
        Console.WriteLine("Subscriber received : " + orderId);
    }
}

导航到包含上述代码的目录，然后运行以下命令来启动 Dapr 边车和订阅者应用程序：

dapr run --app-id checkout --app-port 6002 --dapr-http-port 3602 --dapr-grpc-port 60002 --app-protocol https dotnet run

//dependencies
import io.dapr.Topic;
import io.dapr.client.domain.CloudEvent;
import org.springframework.web.bind.annotation.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;

//code
@RestController
public class CheckoutServiceController {

    private static final Logger log = LoggerFactory.getLogger(CheckoutServiceController.class);
     //订阅一个主题
    @Topic(name = "orders", pubsubName = "order-pub-sub")
    @PostMapping(path = "/checkout")
    public Mono<Void> getCheckout(@RequestBody(required = false) CloudEvent<String> cloudEvent) {
        return Mono.fromRunnable(() -> {
            try {
                log.info("Subscriber received: " + cloudEvent.getData());
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }
}

导航到包含上述代码的目录，然后运行以下命令来启动 Dapr 边车和订阅者应用程序：

dapr run --app-id checkout --app-port 6002 --dapr-http-port 3602 --dapr-grpc-port 60002 mvn spring-boot:run

#dependencies
from cloudevents.sdk.event import v1
from dapr.ext.grpc import App
import logging
import json

#code
app = App()
logging.basicConfig(level = logging.INFO)
#订阅一个主题 
@app.subscribe(pubsub_name='order-pub-sub', topic='orders')
def mytopic(event: v1.Event) -> None:
    data = json.loads(event.Data())
    logging.info('Subscriber received: ' + str(data))

app.run(6002)

导航到包含上述代码的目录，然后运行以下命令来启动 Dapr 边车和订阅者应用程序：

dapr run --app-id checkout --app-port 6002 --dapr-http-port 3602 --app-protocol grpc -- python3 CheckoutService.py

//dependencies
import (
	"log"
	"net/http"
	"context"

	"github.com/dapr/go-sdk/service/common"
	daprd "github.com/dapr/go-sdk/service/http"
)

//code
var sub = &common.Subscription{
	PubsubName: "order-pub-sub",
	Topic:      "orders",
	Route:      "/checkout",
}

func main() {
	s := daprd.NewService(":6002")
   //订阅一个主题
	if err := s.AddTopicEventHandler(sub, eventHandler); err != nil {
		log.Fatalf("error adding topic subscription: %v", err)
	}
	if err := s.Start(); err != nil && err != http.ErrServerClosed {
		log.Fatalf("error listenning: %v", err)
	}
}

func eventHandler(ctx context.Context, e *common.TopicEvent) (retry bool, err error) {
	log.Printf("Subscriber received: %s", e.Data)
	return false, nil
}

导航到包含上述代码的目录，然后运行以下命令来启动 Dapr 边车和订阅者应用程序：

dapr run --app-id checkout --app-port 6002 --dapr-http-port 3602 --dapr-grpc-port 60002 go run CheckoutService.go

//dependencies
import { DaprServer, CommunicationProtocolEnum } from '@dapr/dapr'; 

//code
const daprHost = "127.0.0.1"; 
const serverHost = "127.0.0.1";
const serverPort = "6002"; 

start().catch((e) => {
    console.error(e);
    process.exit(1);
});

async function start(orderId) {
    const server = new DaprServer({
        serverHost,
        serverPort,
        communicationProtocol: CommunicationProtocolEnum.HTTP,
        clientOptions: {
          daprHost,
          daprPort: process.env.DAPR_HTTP_PORT,
        },
    });
    //订阅一个主题
    await server.pubsub.subscribe("order-pub-sub", "orders", async (orderId) => {
        console.log(`Subscriber received: ${JSON.stringify(orderId)}`)
    });
    await server.start();
}

导航到包含上述代码的目录，然后运行以下命令来启动 Dapr 边车和订阅者应用程序：

dapr run --app-id checkout --app-port 6002 --dapr-http-port 3602 --dapr-grpc-port 60002 npm start

发布消息

启动一个 app-id 为 orderprocessing 的 Dapr 实例：

dapr run --app-id orderprocessing --dapr-http-port 3601

然后向 orders 主题发布一条消息：

dapr publish --publish-app-id orderprocessing --pubsub order-pub-sub --topic orders --data '{"orderId": "100"}'

curl -X POST http://localhost:3601/v1.0/publish/order-pub-sub/orders -H "Content-Type: application/json" -d '{"orderId": "100"}'

Invoke-RestMethod -Method Post -ContentType 'application/json' -Body '{"orderId": "100"}' -Uri 'http://localhost:3601/v1.0/publish/order-pub-sub/orders'

以下是利用 Dapr SDK 发布主题的代码示例。

using System;
using System.Collections.Generic;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Threading.Tasks;
using Dapr.Client;
using System.Threading;

const string PUBSUB_NAME = "order-pub-sub";
const string TOPIC_NAME = "orders";

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddDaprClient();

var app = builder.Build();
var random = new Random();

var client = app.Services.GetRequiredService<DaprClient>();

while(true) {
    await Task.Delay(TimeSpan.FromSeconds(5));
    var orderId = random.Next(1,1000);
    var source = new CancellationTokenSource();
    var cancellationToken = source.Token;
    
    //使用 Dapr SDK 发布主题
    await client.PublishEventAsync(PUBSUB_NAME, TOPIC_NAME, orderId, cancellationToken);
    Console.WriteLine("Published data: " + orderId);
}