StreamFlow.RabbitMq.Prometheus 9.1.0

StreamFlow - 另一个 RabbitMQ 处理程序

理由

我知道在野外有多个非常酷的 RabbitMQ 包装器（MassTransit、EasyNetQ 等），它们比这个库做得更好。然而，我从未能够像现在这样使用它们，而不需要重写多个方法或类。所以，我决定根据自己的需求制作一个自己的包装器。

通过 NuGet 安装

如果您想将 HTTP 溢出包含到您的项目中，您可以直接从 NuGet 安装它。

要安装包，请在包管理器控制台中运行以下命令

PM> Install-Package StreamFlow.RabbitMq

概念

在 RabbitMQ 服务器中，所有名称都可以通过实现 IRabbitMqConventions 并在依赖注入中注册来定义。但是默认行为如下

• 交换名称使用请求类名称创建
• 队列名称使用 <request class name>:<request handler name>[:<service id>][:<consumer group name>] 创建
• 错误队列名称使用 <request class name>:<request handler name>[:<service id>][:<consumer group name>]:Error 创建

A consumer group is a group of consumers that share the same group id. 
When a topic is consumed by consumers in the same group, every record 
will be delivered to only one consumer.
If all the consumer instances have the same consumer group, then the 
records will effectively be load-balanced over the consumer instances.

Service id identifies service implementation. As for example if we have 
multiple services like customers and orders - they will have different 
service id's. Most simple case here would be to use a short memorable
identifier like: "customers", "orders". 

The purpose of such id is to distinguish message handlers which can have
exactly the same name and can handle exactly the same message. Since these
are in different services - load balancing scenario is not acceptable in this case.

It can be also treated as a consumer group but in this case it applies to whole service.

组件生命周期

连接

连接对象作为单例注册。然而，消费者和发布者连接之间存在分离。这意味着单个服务实例可以与 RabbitMQ 服务器建立最多两个连接。

IPublisher

使用单例生命周期。内部使用大小为 100_000 项的 Channel 集合。如果代码中的发布操作比此集合的处理速度快得多，则您可能会收到 RabbitMqPublisherException，原因：InternalQueueIsFull。

底层数据发布者使用异步发布确认。这意味着每次PublishAsync实际都是异步等待服务器确认。有关更详细的解释，请参阅RabbitMQ文档：https://rabbitmq.cn/confirms.html#publisher-confirms 在这里，您还可以看到更多关于该行为的详细示例：https://rabbitmq.cn/tutorials/tutorial-seven-dotnet.html 概括来说：如果您对每个PublishAsync方法进行await，实际上就像以下示例一样等待每个发表的确认

public async Task PublishMessagesAsync<T>(T [] messages)
{
    foreach (var message in messages)
    {
      try
      {
          await _publisher.PublishAsync(message);
      }
      catch (Exception e)
      {
          // handle publication errors
      }
    }
}

上面的示例可能相对较慢，但它非常简单易实现。为了提高发布性能，您可以在示例中采取类似的方法

public async Task PublishMessagesAsync<T>(T [] messages)
{
    var tasks = new Task[messages.Length];
    for (var i = 0; i < messages.Length; ++i)
    {
        var message = messages[i];
        tasks[i] = _publisher.PublishAsync(message);
    }
    
    foreach (var task in tasks)
    {
        try
        {
            await task;
        }
        catch (Exception e)
        {
            // handle publication errors
        }
    }
}

注意：如果您没有指定取消令牌（或者指定默认），或者发布选项中没有指定超时，库会自动添加60秒的超时。如果应用程序在指定时间内未收到响应，则会标记消息为已取消。

中间件

每个中间件都是作为瞬态添加的。每个中间件将在每个接收或发布的消息上创建。

消费者

每个消费者都是单例的。但是对于每个接收到的消息，会创建一个自己的作用域，在处理结束后进行释放。

处理器

处理器作为范围实例进行注册。

非常简单易用

定义消息

public class PingRequest
{
    public DateTime Timestamp { get; set; }
}

发布消息

public class PingController
{
    private readonly IPublisher _publisher;
    
    public PingController(IPublisher publisher)
    {
        _publisher = publisher;
    }
    
    public async Task SendPing()
    {
        await _publisher.PublishAsync(new PingRequest { Timestamp = DateTime.UtcNow; });
    }
}

定义消息消费者

public class PingRequestConsumer : IConsumer<PingRequest>
{
    public Task Handle(IMessage<PingRequest> message, CancellationToken cancellation)
    {
        Console.WriteLine(message.Body.Timestamp);
        return Task.CompletedTask;
    }
}

使用ASP.NET Core进行配置

services.AddStreamFlow(transport =>
{
    transport
        .UsingRabbitMq(mq => mq
            .Connection("localhost", "guest", "guest")
            .StartConsumerHostedService()
        )
        .Consumers(builder => builder
            .Add<PingRequest, PingRequestConsumer>(options => options
                .ConsumerCount(5)
                .ConsumerGroup("gr1"))
            .Add<PingRequest, PingRequestConsumer>(options => options
                .ConsumerCount(5)
                .ConsumerGroup("gr2"))
        )
        .ConfigureConsumerPipe(builder => builder
            .Use<LogAppIdMiddleware>()
        )
        .ConfigurePublisherPipe(builder => builder
            .Use(_ => new SetAppIdMiddleware("Published from StreamFlow.Tests.AspNetCore"))
        );
});

上面的代码注册了流流类，配置了RabbitMQ连接，并指示启动ASP.NET Core托管服务，该服务启动了配置的消费者。它配置了2个消费者分组，每组启动5个实例。

ConfigureConsumerPipe和ConfigurePublisherPipe注册了在收到消息或发布时执行的类似于中间件的操作。以下是这些中间件的代码

// used at consumer side to log received AppId
public class LogAppIdMiddleware : IStreamFlowMiddleware
{
    private readonly ILogger<LogAppIdMiddleware> _logger;

    public LogAppIdMiddleware(ILogger<LogAppIdMiddleware> logger)
    {
        _logger = logger;
    }

    public Task Invoke(IMessageContext context, Func<IMessageContext, Task> next)
    {
        if (!string.IsNullOrWhiteSpace(context.AppId))
        {
            _logger.LogInformation("AppId: {AppId}", context.AppId);
        }

        return next(context);
    }
}

// used at publisher side to set custom AppId
public class SetAppIdMiddleware : IStreamFlowMiddleware
{
    private readonly string _appId;

    public SetAppIdMiddleware(string appId)
    {
        _appId = appId;
    }

    public Task Invoke(IMessageContext context, Func<IMessageContext, Task> next)
    {
        context.WithAppId(_appId);
        return next(context);
    }
}

尽管这些示例中间件在现实世界场景中非常简单，但可以构建非常强大的实现

• 重试
• 异常处理
• 日志记录
• 度量
• ...

MediatR

有一个为MediatR库实现的示例，这可能会极大地简化消费者实现。

简单安装包

Install-Package StreamFlow.RabbitMq.MediatR

库希望MediatR本身已在依赖注入容器中配置并可用。

MediatR通知

请求类必须实现INotification接口，例如

public class PingNotification : INotification
{
}

然后它可以在StreamFlow中使用

    ....
    .Consumers(builder => builder
        .AddNotification<PingNotification>()
    )
    ....

MediatR无响应请求

请求类必须实现IRequest接口，例如

public class PingRequest : IRequest
{
}

然后它可以在StreamFlow中使用

    ....
    .Consumers(builder => builder
        .AddRequest<PingRequest>()
    )
    ....

MediatR带响应请求

请求类必须实现IRequest接口，例如

public class PongResponse
{
}

public class PingPongRequest : IRequest<PongResponse>
{
}

然后它可以在StreamFlow中使用

    ....
    .Consumers(builder => builder
        .AddRequest<PingPongRequest, PongResponse>()
    )
    ....

然而在这种情况下，您会得到额外的行为：响应是通过RabbitMQ发布者总线发送的。为了使这成为可能，您还需要启用发布主机（请参阅EnablePublisherHost调用）

services.AddStreamFlow(transport =>
{
    transport
        .UseRabbitMq(mq => mq
            .Connection("localhost", "guest", "guest")
            .EnableConsumerHost(consumer => consumer.Prefetch(5))
            .WithPrometheusMetrics()
            .WithPublisherOptions(publisher => publisher
                .EnablePublisherHost()
            )
        )
}    

度量

库支持各种度量，这些度量表示有关发布者和消费者洞察的各种信息。开箱即用，就有Prometheus度量实现的示例，但当然它不仅限于此。

为了启用度量，请添加Prometheus包

PM> Install-Package StreamFlow.RabbitMq.Prometheus

并在配置RabbitMQ时启用度量（请参阅：WithPrometheusMetrics）

services.AddStreamFlow(streamFlowOptions, transport =>
{
    transport
        .UseRabbitMq(mq => mq
            .Connection(options.Host, options.Username, options.Password, options.VirtualHost)
            .WithPrometheusMetrics()
            ....
        );
});

Prometheus度量

• streamflow_messages_published

  表示已发布消息的数量和持续时间的直方图（秒）

  标签：交易所、状态[完成或失败]

• streamflow_bus_messages_published

  表示使用发布者主机时发布消息的数量和持续时间的直方图（秒）

  标签：交易所、状态[完成或失败]

• streamflow_messages_publishing_events

  表示在发布过程中发生的各种事件的直方图，如通道创建、准备、序列化、事务提交等，有助于了解发布过程中是否存在任何瓶颈

  标签：交易所、事件

• streamflow_messages_publishing_errors

  表示在消息发布过程中发生的异常的计数器

  标签：交易所

• streamflow_bus_messages_publishing_errors

  表示使用发布者主机时在消息发布过程中发生的异常的计数器

  标签：交易所

• streamflow_messages_consumed

  表示已消费消息的数量和消费者持续时间的直方图（秒）

  标签：交易所、队列、状态

• streamflow_messages_consumed_errors

  计数器，表示在消费者处理过程中获得的异常

  标签：交换、队列

• streamflow_messages_bus_publishing

  计数器，显示使用发布者主机发布了多少条消息

• streamflow_messages_bus_publishing_errors

  计数器，表示使用发布者主机发布时的错误，因为发布者主机通道是有限制的 - 当发布者主机接收的消息超出其实际能够发送到RabbitMQ的能力时，该计数器将开始增加

• streamflow_publisher_pool_size

  计数器，使用发布者池时，显示池大小（已创建但当前未使用的发布者）

• streamflow_publisher_pool_in_use

  计数器，使用发布者池时，显示当前有多少个发布者实例正在使用

错误处理

每个应用程序都会处理错误。我确信 - RabbitMQ处理器可能会遇到多种异常情况。在未处理异常的情况下，将创建IRabbitMqErrorHandler实例，该实例默认会将消息发送到错误队列。尽管你可以通过重写IRabbitMqConventions接口定义自己的错误队列名称，但默认行为仅将“：错误”后缀添加到消费者组队列的末尾，并将带有异常详细信息的消息发送到此队列。应用程序开发人员或支持人员以后可以决定如何处理这些消息。