什么是发布与订阅（Publish & Subscribe）？

在分布式环境中，各服务之间通常存在信息传递的需求，除了前面介绍的点对点的直接通信模式，有些应用场景还会存在间接解耦的异步通信模式，而 Pub/Sub 模式（通常称为“发布/订阅”）就是其中一种广泛使用的消息传递模式。架构师通常在分布式应用程序中采用它。但是，实现此模式的管道可能会很复杂。在不同的消息传递产品中，通常会有功能差异。

下面是一个典型的消息发布与订阅模式，在分布式环境中各服务彼此独立且需要相互通信时，此模式特别有用。

创建者（或发布者，Publish） 将消息发送到 主题（Topic），而不知道哪个应用程序将接收消息。这涉及将它们写入输入通道（Input channel）。
使用者（或订阅者，Subscriber） 绑定到消息代理上的订阅主题并接收其消息，而不知道是什么服务生成了这些消息。这涉及从输出通道接收消息（Output channel）。
中间消息代理（Message Broker） 负责将每条消息的副本从输入通道复制到对该消息感兴趣的所有订阅者的输出通道。

发布与订阅模式解决的问题

发布-订阅模式的主要优点是 松散耦合，有时称为临时分离。该模式将发送消息的服务（发布服务器）与使用消息的服务（订阅服务器）分离。发布服务器和订阅服务器都不知道彼此的存在，两者都依赖于分发消息的集中式消息代理（Message Broker）。

大多数消息代理封装了一种在收到消息后可持久保存消息的排队机制。使用该消息代理可通过存储消息来保证持久性。在发布服务器发送消息时，订阅服务器无需立即可用，甚至不需要联机。订阅服务器将在可用后接收并处理消息。

与传统 SDK 模式的区别

在传统使用 SDK 模式实现消息的传递中，通常需要去实现各种的消息传递（MQ）产品的适配应用，必须编写和维护抽象及其基础实现。此方法对应用代码是一种入侵式的，需要自定义代码，这些代码可能很复杂、重复且容易出错，增加了对不同 MQ 产品的学习成本和维护成本。

Dapr 发布和订阅构建基块使用 Sidecar 代理模式提供现成的消息传递抽象和实现。传统 SDK 模式必须编写的自定义代码已预先生成并封装在 Dapr 构建基块中。只需绑定 Dapr Sidecar 模块使用即可。让团队更佳专注于创建可为客户增加价值的业务功能上，而不是编写消息传递管道代码。Sidecar 代理模式对应用是 “零入侵”的，且无需关心各种 MQ 的适配维护，简化了对 MQ 的使用体验。实现应用的松散耦合，使得应用之间无感知且彼此独立互不影响；

Dapr 的发布与订阅（Publish & Subscribe）

Dapr 提供了一个可显著简化【发布/订阅】功能的实现的构建基块。Dapr 为消息传递保证 “至少一次保证（at least once guarantees）” 语义。发布消息后，该消息将至少传递到任何相关订阅服务器一次。

注意：如果服务只能处理一次消息，则需要提供 幂等性检查，确保不会多次处理同一消息。虽然可对此类逻辑进行编码，但某些消息代理（例如 Azure 服务总线）提供内置的重复检测消息传递功能。
分布式系统中的幂等性，指用户对同一操作发起一次或多次请求的处理结果是一致的，不会因为多次操作产生了错误的结果。

Dapr 发布与订阅构建基块工作模式

Dapr 发布和订阅构建基块提供了平台无关的 API 框架来发送和接收消息。你的服务将消息发布到一个命名主题，其他服务并订阅主题以使用消息。

该服务对 Dapr 发布/子构建基块进行网络调用，该构建基块公开为边车（Sidecar）。然后，此构建基块对封装特定消息代理产品的 Dapr 发布/订阅组件进行调用。若要接收主题，Dapr 代理服务订阅 Dapr 发布/订阅组件，并在消息到达时将消息传递到终结点。

下图显示了 “运输（shipping）” 服务 和 "电子邮件（email）"服务 的示例，它们都订阅了由 "购物车（cart）"服务 发布的主题。每个服务都会加载指向同一发布/订阅消息总线组件的发布/订阅组件配置文件，例如 Redis 流、NATS 流式处理、Azure 服务总线或 GCP 发布/订阅。

pub/sub 功能分为发布和订阅两个部分

用户可以通过调用 dapr sidecar API 发送消息到某个 topic；
用户可以通过配置将 handler 绑定到某个 topic, 由 dapr sidecar 订阅 message broker 收到消息后交给用户 handler 处理；

Dapr 允许同一个 Topic 被多个不同的服务订阅，相当于RabbitMQ 中的 Fanout（广播模式）类型。但是同一个服务多个实例副本（competing consumers）则是会被放入一个消费者组（comsume group）中，也就是一个消息只会有某个实例拿到。（相互竞争的消费模式）

RabbitMQ 中的 Fanout（广播模式）简单的讲，就是把交换机（Exchange）里的消息副本发送给所有绑定该交换机的队列（Queue），忽略 routingKey。

Dapr 的 Pub/Sub 组件支持情况，请查看：

Reference / Component specs / Pub & sub brokers

Pub & Sub 在 Dapr 中既是一个 component 又是一个 building block。

Dapr 发布和订阅构建基块调用方式

在最低级别，任何编程平台均可使用 Dapr 本机 API 通过 HTTP 或 gRPC 调用构建基块。

发布（Pub）消息的 Sidecar API URL

http://localhost:<dapr-port>/v1.0/publish/<pub-sub-name>/<topic>

Dapr 发布消息特定 url 说明：

< dapr-port > ：提供 Dapr Sidecar API 正在侦听的端口号；
< pub-sub-name > ：提供选定的 Dapr 发布/订阅组件的名称；
< topic > ：提供消息要发布到消息代理（Message Broker）的主题名称；

通过订阅主题（topic）来接收消息。在启动时，Dapr 运行时将调用已知终结点上的应用程序来识别和创建所需的订阅：

订阅（Sub）消息的 Sidecar API URL

http://localhost:<appPort>/dapr/subscribe

Dapr 订阅消息特定 url 说明：

< appPort > ：通知 Dapr Sidecar API 应用程序正在侦听的端口。

【订阅消息】原则上，当订阅者在处理消息后应答非错误响应时，Dapr 认为成功发送了的消息。为了进行更精细的控制，Dapr 的发布/订阅 API 还提供显式状态（在响应负载中定义），订阅者可以使用这些状态向 Dapr 指示特定的处理指令（例如： RETRY 或 DROP）。

Dapr 发布/订阅的特性

CloudEvents 消息格式

为了启用消息路由并为每个消息提供额外的上下文，Dapr 使用 CloudEvents 1.0 规范作为其消息格式。默认情况下，应用程序使用 Dapr 发送到 Topic 的任何消息都会自动 “包装” 在 CloudEvents 信封中，并使用 datacontenttype 属性的 Content-Type 头值，并且会自动集成分布式追踪。

Dapr 实现以下 Cloud Events 字段：

id
source
specversion
type
datacontenttype (可选)

下面的示例显示了 CloudEvent v1.0 中序列化为 JSON 的 XML 内容：

{"specversion" : "1.0","type" : "xml.message","source" : "https://example.com/message","subject" : "Test XML Message","id" : "id-1234-5678-9101","time" : "2020-09-23T06:23:21Z","datacontenttype" : "text/xml","data" : "<note><to>User1</to><from>user2</from><message>hi</message></note>"
}

pub/sub 主题（Topic）作用域限定

除了此常规组件范围外，对于 pub/sub 组件，还可以限制以下操作：

哪些主题可以使用(发布或订阅)；
哪些应用程序被允许发布到特定主题；
哪些应用程序被允许订阅特定主题；

有关消息路由的更多信息，请阅读 Dapr Pub/Sub API 文档：

限定 Pub/Sub 主题访问权限 - 使用范围（scopes）限制 Pub/Sub 主题到特定的应用程序

消息生存时间（TTL，time-to-live）

Dapr 允许对每个消息设置生存时间(TTL)。这意味着应用程序可以设置每条消息的生存时间，并且这些消息过期后订阅者不会收到。

所有 Dapr Pub/Sub 组件（components ）都与消息 TTL 兼容，因为 Darp 在运行时（ runtime）处理 TTL 逻辑。只需在发布消息时设置 ttlInSecond 元数据（metadata）即可。

原生的消息 TTL 支持

当消息生存时间在发布/订阅组件中具有本机支持时，Dapr 只需转发生存时间配置，而无需添加任何额外的逻辑，从而保持可预测的行为。当组件以不同方式处理过期消息时，这会很有帮助。例如，在 Azure Service Bus 中，过期消息存储在死信队列中，而不是简单地删除。

Azure Service Bus 支持实体级别的 TTL。这意味着消息有默认的 TTL，但也可以在发布时间设置更短的时间。 Dapr 会为消息传播 TTL 元数据，并允许 Azure Service Bus 直接处理过期时间。

非 Dapr 订阅者

如果消息被不使用 Dapr 的订阅者使用，则过期的消息不会自动丢弃，因为当 Dapr Sidecar 收到消息时，过期由 Dapr 运行时（Runtime）处理。但是，订阅者可以通过添加逻辑来处理云事件（ cloud event）中的属性（遵循 RFC3339 格式），以编程方式丢弃过期的消息。

当非 Dapr 订阅者使用以原生方式处理消息 TTL 的组件（如 Azure Service Bus，服务总线）时，他们不会收到过期的消息。在这里，不需要额外的逻辑。

Message Time-to-Live (TTL) - Use time-to-live in Pub/Sub messages

消息传递（At-least-once guarantee）

Dapr 保证消息传递的 “至少一次 / at-least-once” 语义。这意味着当应用程序使用发布/订阅 API 将消息发布到主题时，Dapr 可确保此消息至少传递给每个订阅者一次（at least once）。

相互竞争的消费者模式

多个消费组（Consumer groups）、多个应用程序实例使用一个消费组，这些都将由 Dapr 自动处理。当同一个应用程序的多个实例(相同的 ID) 订阅主题（Topic）时，Dapr 只将每个消息传递给该应用程序的一个实例。

同样，如果两个不同的应用程序 (不同的 ID) 订阅同一主题，那么 Dapr 将每个消息仅传递到每个应用程序的一个实例。

使用 Dapr for dotnet SDK 项目演示

新建 asp.net core web api（此处我们继续使用前面创建的 FrontEnd 项目）
添加 Nuget 包依赖，Dapr.AspNetCore（FrontEnd 项目已添加）

在 Dapr 中存在两种订阅方式，分别如下：

配置声明式
应用编码式

声明和编程方式都支持相同的功能。声明式方法会从您的代码中移除 Dapr 依赖，并允许现有的应用程序订阅 topics，而无需更改代码。编程方法在用户代码中实现订阅。

接下来我们分别演示两种方式的 Pub/Sub ，下面的代码为公用部分：

1》新建【PubSubController】，构造函数 DI 注册 DaprClient；

using Dapr;
using Dapr.Client;
using FrontEnd.Model;
using Microsoft.AspNetCore.Mvc;
using System.Text;namespace FrontEnd.Controllers;[Route("api/[controller]")]
[ApiController]
public class PubSubController : ControllerBase
{private readonly ILogger<PubSubController> _logger;private readonly DaprClient _daprClient;public PubSubController(ILogger<PubSubController> logger, DaprClient daprClient){_logger = logger;_daprClient = daprClient;}}

2》为了方便演示，【PubSubController】中添加如下模拟生产数据方法

private static readonly string[] Summaries = new[]
{"Freezing", "Bracing", "Chilly", "Cool", "Mild", "Warm", "Balmy", "Hot", "Sweltering", "Scorching"
};private static WeatherForecast[] GetWeatherForecasts() => Enumerable.Range(1, 3).Select(index => new WeatherForecast
{Date = DateTime.Now.AddDays(index),TemperatureC = Random.Shared.Next(-20, 55),Summary = Summaries[Random.Shared.Next(Summaries.Length)]
}).ToArray();

3》Dapr 发布（Publish）的 demo，此处代码公用（配置声明式和应用编码式的订阅均可使用）

const string PubSubName = "pubsub";
const string TopicName = "test_topic";[HttpGet("pub")]
public async Task<ActionResult> PubAsync()
{_logger.LogInformation("进入 PubSub.PubAsync");var data = GetWeatherForecasts();await _daprClient.PublishEventAsync(PubSubName, TopicName, data);return Ok();
}

配置声明式

1》定义 Dapr 的订阅（Subscribe）demo

[HttpPost("sub")]
public async Task<ActionResult> SubAsync()
{_logger.LogInformation("进入 PubSub.SubAsync");Stream stream = Request.Body;byte[] buffer = new byte[Request.ContentLength.Value];stream.Position = 0L;await stream.ReadAsync(buffer, 0, buffer.Length);string content = Encoding.UTF8.GetString(buffer);_logger.LogInformation($"sub => {content}");return Ok(content);
}

2》添加【subscription.yaml】配置文件

apiVersion: dapr.io/v1alpha1
kind: Subscription
metadata:name: myevent-subscription
spec:topic: test_topicroute: /api/PubSub/subpubsubname: pubsub
scopes:
- frontend

yaml 配置文件说明：

kind 指定资源类型 Subscription；
topic 指定主题（topic）名称，务必保证和应用代码中的相同；
route 指定 sub 订阅的路由 /api/PubSub/sub；
scopes 限制可访问的应用服务名称 frontend，可设置多个；

windows 环境 dapr 的 components 文件路径 C:\Users<user>.dapr\components

3》在 FrontEnd\Program.cs 中添加如下代码

var app = builder.Build();// 开启重复读取 body
app.Use((context, next) =>
{context.Request.EnableBuffering();return next();
});

应用编码式

1》开启订阅映射
和上面一样，发布代码依然使用上面的方式，在 FrontEnd\Program.cs 中添加开启重复读取 body 代码，然后再【Program.cs】中新增如下代码：

// 开启订阅映射 (在 app.MapControllers(); 之前)
app.MapSubscribeHandler();

2》定义 Dapr 的订阅模式 demo

[Topic(PubSubName, TopicName)]
[HttpPost("sub_code")]
public async Task<ActionResult> SubCodeAsync()
{_logger.LogInformation("进入 PubSub.SubCodeAsync");Stream stream = Request.Body;byte[] buffer = new byte[Request.ContentLength.Value];stream.Position = 0L;await stream.ReadAsync(buffer, 0, buffer.Length);string content = Encoding.UTF8.GetString(buffer);_logger.LogInformation($"sub_code => {content}");return Ok(content);
}

测试发布&订阅

dapr run 命令启动 FrontEnd 服务，测试发布&订阅

dapr run --dapr-http-port 3501 --app-port 5001 --app-id frontend dotnet run

注意：此处演示 demo 两种订阅模式的发布代码公用（ToppicName 和 PublishName 相同），为了方便测试不同模式的 pub/sub 功能，在测试 配置声明式 的时候记得先关闭 应用编码式 的【subscription.yaml】配置文件。
配置声明式和应用编码式共存时，优先应用编码式订阅生效。

dapr 日志记录信息

响应数据结果，遵循 CloudEvents 1.0 规范

{"data": [{"summary": "Balmy","date": "2022-03-30T15:20:46.3886649+08:00","temperatureC": 2,"temperatureF": 35}, {"date": "2022-03-31T15:20:46.3890319+08:00","temperatureC": 40,"temperatureF": 103,"summary": "Hot"}, {"date": "2022-04-01T15:20:46.3890345+08:00","temperatureC": 15,"temperatureF": 58,"summary": "Mild"}],"id": "a04b9e06-a42e-4099-aacb-77bed2140e41","specversion": "1.0","datacontenttype": "application/json","source": "frontend","topic": "test_topic","pubsubname": "pubsub","tracestate": "","type": "com.dapr.event.sent","traceid": "00-a8b868502b0042c446ceac38eeb82458-6786e8a3e154a7a1-01"
}

使用 redis 客户端工具查看数据，主题（test_topic）

以上 Pub / Sub 的演示，默认情况下 Dapr 是使用的 Redis 作为消息传递代理组件（Message Broker），相应的 pubsub.yaml 配置文件如下：

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:name: pubsub # 务必与代码中 PubSubName 相同namespace: production
spec:type: pubsub.redisversion: v1metadata:- name: redisHostvalue: localhost:6379- name: redisPasswordvalue: ""

设置密码方式，可以使用 secret 方式

    - name: redisPasswordsecretKeyRef:name: rediskey: redis-password

修改 Message Broker 为 RabbitMQ

前面我们提到 Dapr 是 Sidecar 模式运行的，同样 Pub / Sub 消息传递代理组件（Message Broker）也是可以替换的，接下来我们使用 RabbitMQ 作为消息代理组件，yaml 配置如下：

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:name: pubsub # 务必与代码中 PubSubName 相同namespace: production
spec:type: pubsub.rabbitmqversion: v1metadata:- name: hostnamevalue: my-rabbit- name: hostvalue: "amqp://admin:admin@localhost:5672" #用户名:密码访问 user:password- name: consumerIDvalue: "chait" # Required. Any unique ID. Example: "myConsumerID"- name: durablevalue: "true" # Optional. Default: "false"- name: deletedWhenUnusedvalue: "false" # Optional. Default: "false"- name: autoAckvalue: "false" # Optional. Default: "false"- name: deliveryModevalue: "2" # Optional. Default: "0". Values between 0 - 2.- name: requeueInFailurevalue: "true" # Optional. Default: "false".- name: ttlInSecondsvalue: 60- name: prefetchCountvalue: 0- name: reconnectWaitvalue: "3"- name: concurrencyvalue: parallel
scopes:- frontend # 设置 frontend 服务可访问

Docker 拉取 RabbitMQ 镜像

docker pull rabbitmq:3.9.13-management

启动 RabbitMQ 容器

docker run -d --hostname my_rabbit --name dapr_rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 5671:5671 -p 5672:5672 -p 25672:25672 -p 15672:15672 rabbitmq:3.9.13-management

dapr run 命令启动 FrontEnd 服务（同上）
浏览器查看 RabbitMQ 管理界面

http://localhost:15672

RabbitMQ 管理界面显示如下信息，说明 Pub/Sub 已经生效。

dapr 输出日志信息

== APP == info: FrontEnd.Controllers.PubSubController[0]
== APP ==       进入 PubSub.PubAsync
== APP == info: FrontEnd.Controllers.PubSubController[0]
== APP ==       进入 PubSub.SubCodeAsync
== APP == info: FrontEnd.Controllers.PubSubController[0]
== APP ==       sub_code => {"specversion":"1.0","datacontenttype":"application/json","source":"frontend","type":"com.dapr.event.sent","tracestate":"","data":[{"date":"2022-03-30T21:12:26.8892942+08:00","temperatureC":47,"temperatureF":116,"summary":"Freezing"},{"date":"2022-03-31T21:12:26.8897324+08:00","temperatureC":17,"temperatureF":62,"summary":"Mild"},{"date":"2022-04-01T21:12:26.8897368+08:00","temperatureC":38,"temperatureF":100,"summary":"Cool"}],"id":"e6b90225-08a4-48ed-b3c5-10ac3929988d","topic":"test_topic","pubsubname":"pubsub","traceid":"00-579e0cc949d6dd4465cf25f2c185b299-c63d6a55386ef74f-01"}

以上就是 Dapr 使用 Dotnet SDK 方式，分别在 Redis 和 RabbitMQ 消息代理中间件中的 Pub / Sub 演示 Demo。

总结

解耦应用服务：发布/订阅模式可帮助你分离分布式应用程序中的服务。 Dapr 发布和订阅构建基块简化了在应用程序中实现此行为的操作。
基础通信管道下沉：通过 Dapr 发布/订阅，你可以将消息发布到特定的主题。同样，构建基块将查询你的服务以确定要订阅的主题。
Any 使用发布/订阅：可通过 HTTP 或使用特定语言的 SDK 之一（例如适用于 Dapr 的 .NET SDK）来本机使用 Dapr 发布/订阅。 .NET SDK 与 ASP.NET Core 平台紧密集成。
可插播的消息代理（Message Broker）：使用 Dapr，可将受支持的消息代理产品插入应用程序。然后，可交换消息代理，而不需要对应用程序进行代码更改。

推荐参考文章

Dapr 发布和订阅构建基块
Publish and subscribe overview
Pub/sub brokers

Dapr for dotnet | 发布与订阅- Publish Subscribe相关推荐

Redis发布与订阅——PUBLISH SUBSCRIBE
2019独角兽企业重金招聘Python工程师标准>>> Redis发布与订阅--PUBLISH & SUBSCRIBE 一般来说,发布与订阅(又称pub/sub)的特点是 ...
RabbitMQ教程 3.发布/订阅(Publish/Subscribe)
搜索:Java课代表,关注公众号,及时获取更多Java干货. 3 发布/订阅(Publish/Subscribe) 在上一节中,我们创建了一个工作队列.其目的是将每个任务只分发给一个worker.本节 ...
RabbitMq 发布订阅 Publish/Subscribe fanout/direct
目录概述交换机临时队列代码概述在上篇中了解到rabbitmq 生产者生产消息到队列,多个消费者可以接受.这篇文章主要记录广播类型为fanout.生产者不在将产生的消息发送到队列,而是将消息 ...
Meteor：发布与订阅
我们可以使用安全的方法让用户端不直接操作数据库,但是还是可以直接读取数据库内容,如果我们还需要保护私有的数据存储,在客户端直接使用Collection.find(),这样的操作方式在实际的项目中并不会 ...
译: 3. RabbitMQ Spring AMQP 之 Publish/Subscribe 发布和订阅
在第一篇教程中,我们展示了如何使用start.spring.io来利用Spring Initializr创建一个具有RabbitMQ starter dependency的项目来创建spring-am ...
php订阅系统,php redis pub/sub(Publish/Subscribe,发布/订阅的信息系统)之基本使用
一.场景介绍最近的一个项目需要用到发布/订阅的信息系统,以做到最新实时消息的通知.经查找后发现了redis pub/sub(发布/订阅的信息系统)可以满足我的开发需求,而且学习成本和使用成本也比较低 ...
AKKA 集群中的发布与订阅Distributed Publish Subscribe in Cluster
Distributed Publish Subscribe in Cluster 基本定义在单机环境下订阅与发布是很常用的,然而在集群环境是比较麻烦和不好实现的: AKKA已经提供了相应的实现,集群 ...
RabbitMQ发布/订阅模式（Publish/Subscribe）
工作队列模式是直接在生产者与消费者里声明好一个队列,这种情况下消息只会对应同类型的消费者. 举个用户注册的列子:用户在注册完后一般都会发送消息通知用户注册成功(失败).如果在一个系统中,用户注册信息有 ...
RabbitMQ入门：发布/订阅（Publish/Subscribe）
在前面的两篇博客中 RabbitMQ入门:Hello RabbitMQ 代码实例 RabbitMQ入门:工作队列(Work Queue) 遇到的实例都是一个消息只发送给一个消费者(工作者),他们的消息 ...
RabbitMQ工作模式Publish/Subscribe发布订阅,test测试代码
RabbitMQ有以下几种工作模式 : 1.Work queues 工作队列 2.Publish/Subscribe 发布订阅 3.Routing 路由 4.Topics 通 ...

Dapr for dotnet | 发布与订阅- Publish Subscribe