Dapr微服务应用开发系列5：发布订阅构建块 / 四六文摘

dotNET跨平台 1周前

以下文章来源于dotNET开发经验谈，作者朱永光

dotNET开发经验谈朱永光关于.NET开发各方面的技术分享，有时候也会精选整理、翻译和转载社区里对开发人员、架构师、团队管理者有用的一些精华文章。

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

注：对于“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/v1alpha1kind: Subscriptionmetadata:  name: myevent-subscriptionspec:  topic: deathStarStatus  route: /dsstatus  pubsubname: pubsubscopes:- 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/v1alpha1kind: Subscriptionmetadata:  name: quickstarts-subscriptionspec:  topic: quickstarts/sleep  route: /api/sleep  pubsubname: pubsubscopes:- quickstarts-pbc

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