使用两个背景服务生产并消耗渠道以缓存消息

Question

我正在使用RabbitMQ队列中的背景服务消费消息进行项目。

我有一个背景服务使用背景任务队列，例如 this 和在这里处理任务并行处理。

我想将消耗的缓冲消息存储并通过ID进行分组，并在指定的间隔或缓冲区大小后将它们发送到另一个RabbitMQ队列中。

我已经发布问题，解决方案将是实施通道＆lt; t＆gt; 。

但是工人不能以同一线程生产和消费。

因此，我考虑使用2名工人，一名工人从RabbitMQ中食用消息并将其存储在频道（写）和另一个工人中 该读取频道和小组项目将它们发送到另一个队列。

这是实现这一目标的最佳方法吗？

Answer 1

事情要简单得多。而且没有泄漏，除非您严重滥用渠道。

背景服务或iHostedService不是工人或任务，它是主机启动时创建的类。当应用程序启动时，主机调用 startAsync 在其上，最后 stopAsync 在最后。 backgroundService 基类简化了aa位的内容，因此您可以将所有代码放在被覆盖的 executeasync 中，而无需处理启动和停止。您可以在 startAsync 或 executeasync 中开始任意多个任务。您无需创建多个背景/托管服务。

在托管服务中， startAsync（concellationToken）可以创建您的客户端，连接到RabbitMQ，然后注册事件处理程序以将任何用于处理消息的内容发布。 .NET提供多个处理事件流的方法，这两者都有：

EventingBasicConsumer? _consumer;

void OnReceived(object sender,BasicDeliverEventArgs ea)
{
   var body = ea.Body.ToArray();
   //Process that message
}

protected override async Task StartAsync(CancellationToken stoppingToken)
{

   _consumer=StartRabbitAndCreateConsumer();
   _consumer.Received += OnReceived;
}

protected override async Task StopAsync(CancellationToken stoppingToken)
{
    _consumer.Received -= OnReceived;
    _consumer.Dispose();
}

public void Dispose()
{
    CleanupRabbitMqEtc();
}

构造处理管道来处理这些事件的方法有很多方法

• channels 是“只是”螺纹安全，异步（即非阻滞），订单保存，多录，多阅读器，有限或无界队列。使用可配置的溢出行为（例如，丢弃最古老或最新消息）。它们与GO的渠道非常相似，并以类似的方式用于构建处理任务的管道。您必须编写从频道读取的代码，处理消息并写入下一个代码，通常使用任务。
• dataflow 库是一个提供块的较高级别库，它同时包含输入/输出缓冲区和处理。块使用工人任务来处理其输入缓冲区的消息，并将其发布到其输出缓冲区。可以将块链接到构造处理管道或网格
• 反应性扩展是专门为使用IOBServable＆lt;＆gt;＆gt;＆gt;＆gt;＆gt;＆gt;＆gt;专门构建的。来源，LINQ在IEnumerable上的工作方式相同。它们也可用于创建管道，但既包含基于消息的操作员，又包含基于时间的操作员。

使用rx.net，按键进行分组，可以通过类似于此的查询进行缓冲和超时进行缓冲，借用从这个问题中，

eventSource
    .GroupBy(e => e.GroupingKey)
    .SelectMany(group => group.Buffer(TimeSpan.FromSeconds(60), 100))
    .Subscribe(list => SendToStorage(list));

因为您关心查询可以的分组密钥（ID）可以 使用rx.net调整为

eventSource
    .GroupBy(e => e.ID)
    .SelectMany(group => new {
                         ID=group.Key,
                         Items=group.Buffer(TimeSpan.FromSeconds(60)), 100)
               })
    .Subscribe(batch => SendToQueue(batch.ID,batch.Items));

gotcha ，它是针对事件流处理优化的，与其他选项不同，因此它是单线的默认！您可以指定可观察到的，订阅甚至各个步骤都会以不同的线程运行。 .subscribeon（Scheduler.threadPool）将在TreadPool线程上运行订阅，

eventSource
    .GroupBy(e => e.ID)
    .SelectMany(group => new {
                         ID=group.Key,
                         Items=group.Buffer(TimeSpan.FromSeconds(60)), 100)
               })
    .SubcribeOn(Scheduler.ThreadPool)
    .Subscribe(batch => SendToQueue(batch.ID,batch.Items));

从事件中创建可观察的可观察到的，但有人要求明确的rx.net支持在兔子中，甚至提供了一个来源。不幸的是，该请求被拒绝，因为一些评论者想要RX.NET操作员，其他人想要数据流操作员，因此决定不包括特定的实现。

在该问题中使用实现：

public IObservable<DisposableValue<T>> Receive<T>(string exchangeName, string routingKey)
{
    if (exchangeName == null)
    {
        throw new ArgumentNullException(nameof(exchangeName));
    }

    if (routingKey == null)
    {
        throw new ArgumentNullException(nameof(routingKey));
    }

    var queueName = this.model.QueueDeclare().QueueName;
    this.model.QueueBind(queueName, exchangeName, routingKey);

    using (var subscription = new Subscription(model, queueName, false))
    {
        var consumer = new EventingBasicConsumer(this.model);

        var observable = Observable
            .FromEventPattern<BasicDeliverEventArgs>(
                x => consumer.Received += x,
                x => consumer.Received -= x)
            .Select(x => new DisposableValue<T>(
                Deserialize<T>(x.EventArgs.Body),
                () => this.model.BasicAck(x.EventArgs.DeliveryTag, false)));

        this.model.BasicConsume(queueName, false, consumer);

        return observable;
    }
}

public class DisposableValue<T> : IDisposable
{
    private readonly Action disposeAction;

    public DisposableValue(T value, Action disposeAction)
    {
        this.disposeAction = this.disposeAction ??
            throw new ArgumentNullException(nameof(DisposableValue<T>.disposeAction));
        this.Value = value;
    }

    public T Value { get; }

    public void Dispose() => this.disposeAction();
}

您可以在 startAsync 中创建RX.NET订阅，然后将其放在 stopAsync 中：

protected override async Task StartAsync(CancellationToken stoppingToken)
{
    _sub=Receive<Foo>("Exchange", "RoutingKey")
            .GroupBy(e => e.ID)
            .SelectMany(group => new {
                         ID=group.Key,
                         Items=group.Buffer(TimeSpan.FromSeconds(60)), 100)
               })
            .SubcribeOn(Scheduler.ThreadPool)
            .Subscribe(batch => SendToQueue(batch.ID,batch.Items));
}

protected override async Task StopAsync(CancellationToken stoppingToken)
{
    _sub.Dispose();
}

Answer 2

看来您正在搜索这样的linq操作员：

/// <summary>
/// Groups the elements of an asynchronous sequence according to a specified
/// key selector function and comparer. The groupTimeSpan and groupCount parameters
/// are used to control the lifetime of groups. A group is emitted when the
/// specified time-span has elapsed after receiving the first element of the group,
/// or when a group contains the specified number of elements. Multiple groups
/// with the same key can be emitted by the resulting sequence.
/// </summary>
public static IAsyncEnumerable<IGrouping<TKey, TSource>>
    GroupByUntil<TSource, TKey>(
        this IAsyncEnumerable<TSource> source,
        Func<TSource, TKey> keySelector,
        TimeSpan groupTimeSpan,
        int groupCount);

然后，您可以使用它来消耗 channel＆gt; message＆gt; 在具有相同 id> id 的消息组中：

var groupedMessages = channel.Reader.ReadAllAsync()
    .GroupByUntil(msg => msg.Id, TimeSpan.FromSeconds(15), groupCount: 50);
await foreach (IGrouping<int, Message> messages in groupedMessages)
{
    // Process group of messages
}

问题在于， groupByuntil 异步枚举序列的操作员不存在。 system.linq.async 软件包包含功能，这些功能远不及此级别。精致。您期望在 groupByuntil 确实存在可观察到的序列（System.Reactive）的操作员，并具有此签名：

// Groups the elements of an observable sequence according to a specified key
// selector function and comparer. A duration selector function is used to control
// the lifetime of groups. When a group expires, it receives an OnCompleted
// notification. When a new element with the same key value as a reclaimed group
// occurs, the group will be reborn with a new lifetime request.
public static IObservable<IGroupedObservable<TKey, TSource>>
    GroupByUntil<TSource, TKey, TDuration>(
    this IObservable<TSource> source,
    Func<TSource, TKey> keySelector,
    Func<IGroupedObservable<TKey, TSource>, IObservable<TDuration>> durationSelector);

与 to Observable /结合使用此操作员很容易使用此操作员。 toasyncenumerable 转换器（system.linq.async），以实现理想的操作员，但存在问题。让我们首先查看实现，然后再讨论问题：

// Caution: Hidden queue
public static IAsyncEnumerable<IGrouping<TKey, TSource>>
    GroupByUntil<TSource, TKey>(
        this IAsyncEnumerable<TSource> source,
        Func<TSource, TKey> keySelector,
        TimeSpan groupTimeSpan,
        int groupCount,
        IEqualityComparer<TKey> keyComparer = null)
{
    ArgumentNullException.ThrowIfNull(source);
    ArgumentNullException.ThrowIfNull(keySelector);
    if (groupTimeSpan < TimeSpan.Zero)
        throw new ArgumentOutOfRangeException(nameof(groupTimeSpan));
    if (groupCount < 1) throw new ArgumentOutOfRangeException(nameof(groupCount));
    keyComparer ??= EqualityComparer<TKey>.Default;
    return source
        .ToObservable()
        .GroupByUntil(keySelector, g => g.Skip(groupCount - 1).Select(_ => 0L)
            .Amb(Observable.Timer(groupTimeSpan)), keyComparer)
        .SelectMany(g => g.ToList().Select(x => x.GroupBy(_ => g.Key, keyComparer)))
        .ToAsyncEnumerable();
}

问题是 source> source 序列的枚举不是由结果序列的枚举驱动的。取而代之的是，源序列是由后台消费者以最大速度列举的，并且消耗的消息在隐藏的队列中被缓冲。您无法控制该队列的大小，也无法知道其当前大小（至少没有直接）。当您将 to Observable 转换器附加到它时，这就是枚举序列发生的情况。

这有多大的问题？这取决于 source 序列中包含的消息总数，这些消息的发射频率以及所得顺序的消费者在消耗每个组后必须执行的工作量。在极端情况下，无限序列的速度比消费者的速度更快，隐藏的队列将越来越大，从 source 和处理过程中消费消息之间的延迟将会更长和更长的时间，最终，迟早或更晚的应用程序将使用 OutofMemoryException 崩溃。

不幸的是，正确实施 groupByuntil 操作员并不是微不足道的。您可以看到，执行最大间隔连续批次之间的策略？”>这里 iasyncencenumerable＆lt; t＆gt; 序列（ buffer 操作员在这种情况下）。我不会尝试在此处实现 groupByuntil ，因为这是一项重大任务，您可能不需要它。也许您的情况不会受到 to Observable / toasyncenumerable 实现的缺点的影响。如果确实如此，您可能会尝试自己实施它，如果您被卡住了，则可以发布一个有关麻烦的新问题，我们也许可以提供帮助。