Dapr专题之05发布订阅

• Dapr发布订阅
• Dapr发布订阅的优势
• 调用终结点
• • 使用默认的Redis
  • 运行与测试
  • 切换RabbitMQ
  • 测试
• 总结

提示：以下是本篇文章正文内容，下面案例可供参考

• 可插拔的存储组件
  • 支持的存储组件：

    https://docs.dapr.io/zh-hans/reference/components-reference/supported-pubsub/
    

• 代码不需要耦合相关的SDK

http://localhost:/v1.0/publish//

• :Dapr正在侦听的HTTP端口
  

• : 提供选定的 Dapr pub/sub 组件的名称。
  

• : 提供消息要发布到的主题的名称。
  

1. FrontEnd项目下，新建PubSubController

2. 注入DaprClient

    private readonly ILogger _logger;
    private readonly DaprClient _daprClient;
    public PubSubController(ILogger logger, DaprClient daprClient)
    {
        _logger = logger;
        _daprClient = daprClient;
    }
   

3. 声明要使用的队列组件，这里需要注意PUB_SUN 需要和默认pubsub.yaml中的name保持一直

     const string PUB_SUN = "pubsub";
     const string TOPIC_NAME = "test_topic_code";
   

4. 新增接口

    [HttpGet("pub_code")]
    public async Task PubCodeAsync()
    {
        var data = new WeatherForecast();
        await _daprClient.PublishEventAsync(PUB_SUN, TOPIC_NAME, data);
        return Ok();
    }
   
    [Topic(PUB_SUN, TOPIC_NAME)]
    [HttpPost("sub_code")]
    public async Task SubCode()
    {
        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("testsub" + content);
        return Ok(content);
    }
    
   
   

5. 配置开启重复读取Request.Body

   app.Use((context, next) =>
   {
       context.Request.EnableBuffering();
       return next();
   });
   

6. 配置开启订阅映射

    app.MapSubscribeHandler();
   

• 启动FrontEnd

  dapr run --dapr-http-port 3501 --app-port 5001  --app-id frontend dotnet  .FrontEndbinDebugnet6.0FrontEnd.dll
  

调用API，显示200已成功：

可以在控制台看到刚才写入的日志

• 新增yaml文件pubsub-rabbitmq.yaml，写入以下代码

  apiVersion: dapr.io/v1alpha1
  kind: Component
  metadata:
   name: rabbitmq-pubsub
   namespace: default
  spec:
   type: pubsub.rabbitmq
   version: v1
   metadata:
   - name: host
     value: amqp://guest:guest@192.168.11.12:5672
  

• 修改控制器中的PUB_SUN，注意这里的PUB_SUN需要和yaml文件中metadata的name保持一致

  const string PUB_SUN= "rabbitmq-pubsub";
  

• 重启项目

调用API，显示200已成功：

查看RabbitMQ：