我们决定从这周开始在实际开发中使用 dapr，先在 pub/sub 场景使用。这篇博文记录一下在 kubernetes 集群中基于 ASP.NET Core 使用 dapr 发送/订阅消息的试验过程。
Dapr 环境准备
在 k8s 集群上部署好 dapr
# dapr status -k
NAME NAMESPACE HEALTHY STATUS REPLICAS VERSION AGE CREATED
dapr-placement-server dapr-system True Running 3 1.5.0 7d 2021-11-13 11:22.53
dapr-sentry dapr-system True Running 3 1.5.0 7d 2021-11-13 10:51.45
dapr-dashboard dapr-system True Running 1 0.9.0 7d 2021-11-13 10:50.39
dapr-operator dapr-system True Running 3 1.5.0 7d 2021-11-13 10:51.10
dapr-sidecar-injector dapr-system True Running 3 1.5.0 7d 2021-11-13 10:50.40
部署 pub/sub component ，这里用 redis
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
name: pubsub
namespace: production
spec:
type: pubsub.redis
version: v1
metadata:
- name: redisHost
value: redis-master.production.svc.cluster.local:6379
- name: redisPassword
secretKeyRef:
name: redis
key: redis-password
给发布与订阅消息的应用添加 dapr 注解（k8s deployment），添加之后 dapr 会自动向 pod 中注入 sidecar。
spec:
template:
metadata:
annotations:
dapr.io/app-id: ing-web
dapr.io/enabled: "true"
dapr.io/app-port: "80"
注：dapr.io/app-port: "80" 一定不能少，默认端口不是80，开始没有加这个，造成订阅的消息总是收不到。
应用A发送消息
安装 dapr .net sdk 的 nuget 包 Dapr.AspNetCore
dotnet add package
在 Startup 的 ConfigureServices 中添加 AddDaprClient
services.AddDaprClient();
注：这个项目只发消息，不订阅消息，所以不需要 AddDapr 与 Configure 的 MapSubscribeHandler。
在构造函数中注入 DaprClient
public IngService(DaprClient daprClient)
{
}
用 PublishEventAsync 方法发消息到消息队列
await _daprClient.PublishEventAsync("pubsub", "newIng", ing);
在 redis 中检查消息是否发送成功
kubectl exec -it StatefulSet/redis-master -- redis-cli
127.0.0.1:6379> KEYS *
1) "newIng"
127.0.0.1:6379> TYPE newIng
stream
确认发送成功，消息是以 stream 类型保存在 redis 中的，key 名称就是 topic 名称。
应用B订阅消息
nuget 安装 Dapr.AspNetCore 并在 Startup 的 ConfigureServices 中添加 AddDapr
public void ConfigureServices(IServiceCollection services)
{
services.AddMvc().AddDapr();
}
在 Configure 中添加 endpoints.MapSubscribeHandler()
app.UseEndpoints(endpoints =>
{
endpoints.MapSubscribeHandler();
// ...
});
实现订阅消息的 Controller，并在 Action 处添加 Topic 属性
[ApiController]
public class SubscriptionController : ControllerBase
{
private readonly ILogger _logger;
public SubscriptionController(ILogger<SubscriptionController> logger)
{
_logger = logger;
}
[Topic("pubsub", "newIng")]
[HttpPost("/sub/newing")]
public IActionResult NewIng(Ing ing)
{
_logger.LogInformation(
"Received message: {Content} {DateAdded}",
ing.Content,
ing.DateAdded.ToString("yyyy-MM-dd HH:mm:ss"));
return Ok();
}
}
endpoints.MapSubscribeHandler 的作用是让 dapr 发现消息订阅者，dapr 收到消息后会向应用的 /dapr/subscribe 路径发请求，如果发现有对应消息的订阅者，会向订阅者的请求路径 POST 消息。
对于上面的示例代码，我们可以进入应用的容器用 curl 命令验证一下
# curl localhost/dapr/subscribe
[{"topic":"newIng","pubsubName":"pubsub","route":"sub/newing"}]
点火试验
应用A对应的是园子的闪存，我发了一条闪存“[dapr]此闪会通过 dapr 向消息队列发一条消息3”，随即对应的消息被发出，看看应用B的日志中是否记录了这条订阅消息。
Received message: [dapr]此闪会通过 dapr 向消息队列发一条消息3 2021-11-21 15:19:41
收到了，试验初步成功。
待解决问题
- 每次
/dapr/subscribe被请求时，日志中都会出现下面的错误，对应实现代码在 DaprEndpointRouteBuilderExtensions
A default subscription to topic newIng on pubsub pubsub already exists.
- 调用订阅者 Action 的鉴权问题，比如调用上面加了 Topic 属性 NewIng 方法，如果应用是暴露在公网上的，没有鉴权，就谁都可以调用。