Dapr微服务应用开发系列5：发布订阅构建块

题记：这篇介绍发布订阅构建块，这是对事件驱动架构设计的一种实现落地。

注：对于“Building Blocks”这个词组的翻译，我之前使用了“构件块”，现在和官方文档（Dapr中文社区的贡献）保持一致，采用“构建块”。

原理

发布订阅的概念来自于事件驱动架构（EDA）的设计思想，这是一种让程序（应用、服务）之间解耦的主要方式，通过发布订阅的思想也可以实现服务之间的异步调用。而大部分分布式应用都会依赖这样的发布订阅解耦模式。

整个发布订阅的思想其实是比较简单的：

如上图所示，把需要解耦的程序分别设定为事件发布者或者事件订阅者（理论上，对于某个事件，一个程序仅能作为一种角色；对于不同事件，一个程序可以既作为发布者又可以作为订阅者）。同时利用消息代理（Message Broker）中间件把两者对接起来，消息代理即作为事件消息的传输通道。

在Dapr中对这种发布订阅模式进行了高度抽象的实现，并提供了自由替换消息代理中间件的特性，如下图所示：

Dapr的发布订阅构建块也可以被看作一种事件总线（Event Bus）的实现，只是你不需要使用特殊的协议，在发布端和订阅端仅使用HTTP/gRPC即可。

在事件总线中，把发布订阅两者关联在一起的是事件类型，那么在Dapr中也引入了一个类似的概念——主题（Topic）。如果对消息队列中间件熟悉的人对于这个概念不会陌生。由此发布端和订阅端的处理过程和针对Dapr的接口也就是围绕主题来展开的。

能力

消息发送

既然Dapr的PubSub是一种事件总线，那么要发送消息，必然需要对代表主题（事件类型）的消息进行封装。Dapr并没有去创造一种独有的格式，而是采用了目前业界比较流行的开放协议——云事件（CloudEvents）规范。这种格式把事件消息封装为如下JSON数据：

{"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>"
}

当然对消息的封装不需要应用程序本身去关心，你只需要给Dapr传递data的字符串即可，而这个字符串本身是以什么格式（不管xml还是json）去承载内容都是由应用程序确定。具体如何发送消息，下面规范部分会介绍。

消息传递

Dapr会自动根据主题把消息发送给所有订阅者，传递过程保证“至少一次”送达。送达的判断标准是基于订阅者的响应是否成功（即HTTP状态码为20X）。

当然，订阅者也可以在响应体中设置 status 属性来给出更为精细的处理指令，比如 RETRY 告知Dapr之前处理失败了，现在是重试成功了；或者 DROP 告知Dapr应用程序对这个消息处理出现问题，已经记录了告警日志，但是不打算继续处理它了。

消息传递还有一个重要的特性需要理解，就是消息的生存期（Time-to-Live，TTL）。TTL规定了消息在Dapr（实际上是在消息代理中间件）里面的存活时间，如果TTL过期，那么消息就不会再被传递（即变成死信）。所有目前支持的发布订阅组件都支持TTL的特性，Dapr会帮助你处理这方面的逻辑。

消息消费

为了消费消息，需要对主题进行注册，可以通过声明式和编程式来进行注册。声明式通过外部的yaml文件定义一个K8S的CRD，来描述服务需要订阅什么主题，接收事件的HTTP API路由地址。编程式通过暴露特定的HTTP API路由地址或者特定的gRPC方法来让Dapr运行时进行访问，从而注册需要订阅什么主题和接收事件的地址。

发布订阅构建块采用的是所谓竞争者消费模式，即同一个应用（AppId相同）的多个实例，只会有一个实例获得消息，这些同个应用的多个实例称之为一个消费组。如果你希望消息被多个应用得到，那么就需要使用多个消费组，也即多个AppId。

主题范围限制

从上面所知，在发送消息和消费消息的时候，都需要针对某个主题。为了对消息的传递进行更加精细的控制，在发布订阅构建块中可以对主题范围进行限制，即某些主题只能由某些应用来发布，某些主题只能由某些应用来订阅。

要进行范围限制，需要对发布订阅组件的配置yaml进行配置，设置 spec.metadata 下面的 publishingScopes， subscriptionScopes 和 allowedTopics 配置。（详细说明见未来的关于组件的文章）。

规范

Dapr给PubSub这一构建块提供了两方面的规范：消息生产端和消息消费端。

消息生产端

通过POST如下地址，来发送消息到特定主题：

POST http://localhost:<daprPort>/v1.0/publish/<pubsubname>/<topic>[?<metadata>]

其中pubsubname代表了PubSub组件的名称；topic代表了目标主题名称。

在 Content-Type 请求头中设置请求体的格式，比如 application/json

请求体根据请求头里面的设置格式，传入JSON或者XML，Dapr会自动把请求体封装为CloudEvent格式。

如果是直接调用gRPC的接口的话，是调用 PublishEvent 接口并传递 PublishEventRequest 实体。

消息消费端

如果你的消费端是通过声明式来向Dapr注册需要订阅什么主题的消息，那么在如下格式的yaml文件中进行定义：

apiVersion: dapr.io/v1alpha1
kind: Subscription
metadata:name: myevent-subscription
spec:topic: deathStarStatusroute: /dsstatuspubsubname: pubsub
scopes:
- app1
- app2

其中 spec.topic 代表了要订阅的主题名称，spec.route 代表了接收订阅消息的HTTP路由地址，spec.route 代表了针对的PubSub组件是那个。尤为关键是 scopes 里面设置了这样的订阅声明是针对那个（或那几个）应用起作用（填入appid）。

如果你的消费端是通过编程式来向Dapr注册需要订阅什么主题的消息，那么暴露一个如下特殊HTTP路由地址的接口：

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

并返回如下格式的响应体，让Dapr知道你的应用需要订阅什么主题，接收消息的接口路由地址是什么：

[{"pubsubname": "pubsub","topic": "newOrder","route": "/orders"}
]

当然你的应用需要暴露注册的接收路由接口，并支持POST谓词，接口收到请求后返回2xx状态码告诉Dapr消息处理成功了，或者404告诉Dapr出现错误且消息已经丢弃，或者其他状态码让Dapr进行重试。

两种订阅注册方式各有优缺，声明式方便一个主题注册多个应用，编程式方便一个应用注册多个主题。

注意：如果是使用gRPC来注册和暴露消费接口，那么规范有所不同，做法见下面。

DOTNET SDK

Dapr的.NET SDK同样针对消息生产端和消费端提供相关的函数库。

在DaprClient这个客户端类中提供了 PublishEventAsync 这个方法来用于发送事件消息到特定PubSub的特定主题上（底层是请求了gRPC的接口）。比如：

using var client = new DaprClientBuilder().Build();var eventData = new { Id = "17", Amount = 10m, };
await client.PublishEventAsync(pubsubName, "deposit", eventData, cancellationToken);

在消费端，目前针对ASP.NET Core提供了一个特殊的属性标记 TopicAttribute 来简化编程式订阅注册的过程。比如：

[Topic("pubsub", "deposit")]
[HttpPost("deposit")]
public async Task<ActionResult<Account>> Deposit(Transaction transaction, [FromServices] DaprClient daprClient)

如果你是使用gRPC来实现消费端，那么目前并没有一个简化方式来注册（未来我会补上这个坑），只能遵循如下规范：

首先用ListTopicSubscriptions注册：

public override Task<ListTopicSubscriptionsResponse> ListTopicSubscriptions(Empty request, ServerCallContext context)
{var result = new ListTopicSubscriptionsResponse();result.Subscriptions.Add(new TopicSubscription{PubsubName = "pubsub",Topic = "deposit"});result.Subscriptions.Add(new TopicSubscription{PubsubName = "pubsub",Topic = "withdraw"});return Task.FromResult(result);
}

接着用OnTopicEvent接收：

public override async Task<TopicEventResponse> OnTopicEvent(TopicEventRequest request, ServerCallContext context)
{if (request.PubsubName == "pubsub"){var input = JsonSerializer.Deserialize<Models.Transaction>(request.Data.ToStringUtf8(), this.jsonOptions);var transaction = new GrpcServiceSample.Generated.Transaction() { Id = input.Id, Amount = (int)input.Amount, };if (request.Topic == "deposit"){await Deposit(transaction, context);}else{await Withdraw(transaction, context);}}return await Task.FromResult(default(TopicEventResponse));
}

具体见SDK的examples：https://github.com/heavenwing/dapr-dotnet-sdk/blob/master/examples/AspNetCore/GrpcServiceSample/Services/BankingService.cs

用法与例子

使用SDK来进行事件消息的发布和订阅，可以直接参考SDK的examples中的消费端例子 ControllerSample 和生产端例子 PublishSubscribe。

如果是非SDK的用法，可以参考我的quickstarts，消费端 PubSubConsumer和生产端 PubSubProducer。

我的quickstarts的消费端同时使用声明式和编程式两种注册方式。大致代码如下：

[Route("dapr/subscribe")]
[ApiController]
public class DaprSubscribeController : ControllerBase
{public ActionResult<DaprSubscribeOutput[]> Get(){return Ok(new DaprSubscribeOutput[]{new DaprSubscribeOutput{PubSubName="pubsub",Topic="quickstarts/wakeup",Route="/api/wakeup"}});}
}public async Task<IActionResult> PostAsync(TinyCloudEvent<MessageInput> model)
{_logger.LogInformation(model.Data.Name);return Ok();
}using var httpClient = new HttpClient();
httpClient.BaseAddress = new Uri(pubsubUrl);Console.WriteLine($"To {topicName1} ...");
var request1 = new HttpRequestMessage(HttpMethod.Post, topicName1);
request1.Content = new StringContent(JsonSerializer.Serialize(new { name = "Jack" }), Encoding.UTF8, "application/json");
await httpClient.SendAsync(request1);

apiVersion: dapr.io/v1alpha1
kind: Subscription
metadata:name: quickstarts-subscription
spec:topic: quickstarts/sleeproute: /api/sleeppubsubname: pubsub
scopes:
- quickstarts-pbc