如何使用 BlockingCollection 通知生产者-消费者模式中的失败？

Question

我正在尝试创建一个生命周期过程，用于批量插入数据库批量插入的传入消息。 新消息一次传入 1 条，间隔不规则。 我的解决方案类似于使用 BlockingCollection 的生产者-消费者模式。 消息通过各种事件自由添加到 BlockingCollection 中，并按固定时间间隔（5 秒）从 BlockingCollection 中批量取出以进行数据库插入。

然而，当前的解决方案是“即发即忘”。如果批量插入因任何原因失败，我需要一种方法让处理器通知失败的原始源，因为源包含恢复和重试的逻辑。

我是否应该使用特定的模式来实现我想要实现的目标？ 非常感谢任何建议或帮助！

        private BlockingCollection<Message> _messageCollection;

        public async Task<bool> InsertMessage(Message message)
        {
            if (!_messageCollection.TryAdd(message)) return false;

            // TODO: check message has been successfully processed, if not return false
            // return false;

            return true;
        }

        private void BulkInsertProcess()
        {
            Task consumerThread = Task.Factory.StartNew(async () =>
            {
                while (!_messageCollection.IsCompleted)
                {
                    List<Message> messages = new List<Message>();

                    for (int i = 0; i < 50; i++)
                    {
                        if (_messageCollection.Any())
                        {
                            messages.Add(_messageCollection.Take());
                        }
                        else
                        {
                            break;
                        }
                    }

                    bool insertResult = await _database.BulkInsertMessages(messages);

                    // TODO: check result and inform the consumer if insert failed

                    await Task.Delay(5000);
                }
            });
        }

Answer 1

您必须以某种方式将每个 Message 与专用的 TaskCompletionSource。您可能希望使第二个成为第一个的属性：

public class Message
{
    public TaskCompletionSource<bool> TCS { get; } = new();
}

...或者使第一个成为第二个的属性：

private class Entry : TaskCompletionSource<bool>
{
    public Message Message { get; init; }
}

...或者创建一个包含两者的自定义类，或者使用 ValueTuple> 就像我一样已经在下面的示例中选择：

private BlockingCollection<(Message, TaskCompletionSource<bool>)> _queue;

public Task<bool> InsertMessage(Message message)
{
    var tcs = new TaskCompletionSource<bool>(
        TaskCreationOptions.RunContinuationsAsynchronously);
    if (!_queue.TryAdd((message, tcs)))
        return Task.FromResult(false);
    return tcs.Task;
}

private void BulkInsertProcess()
{
    Task consumerTask = Task.Run(async () =>
    {
        while (!_queue.IsCompleted)
        {
            var delayTask = Task.Delay(5000);

            var batch = new List<(Message, TaskCompletionSource<bool>)>();
            while (batch.Count < 50 && _queue.TryTake(out var entry))
                batch.Add(entry);

            if (batch.Count > 0)
            {
                var messages = batch.Select(e => e.Item1).ToList();
                bool insertResult = await _database.BulkInsertMessages(messages);

                foreach (var (message, tcs) in batch)
                    tcs.SetResult(insertResult);
            }
            await delayTask;
        }
    });
}

我对您的代码进行了一些改进，以使其工作更顺畅：

1. 用 Task.Run 而不是 Task.Factory.StartNew。前者理解异步委托。后面的则不会。
2. TryTake 而不是 任何。 Any 是 IEnumerable 接口上的扩展方法，这些不保证线程安全。它很可能是线程安全的，但使用 BlockingCollection 类的公共成员更安全、更高效。
3. 在执行批量插入操作之前创建Task.Delay，然后await它。通过这种方式，您可以在后续批量插入操作之间获得稳定的间隔，这不依赖于操作本身的持续时间。

如果您在一批中收到 50 条消息，您可能会考虑完全跳过 await delayTask，因为这表明您的服务面临压力，并且消息正在队列中堆积。