基于分组的并行消息处理线程安全实现及内存优化问题咨询

核心思路 #

用线程安全字典维护GroupId到ActionBlock<Message>的映射，每个Block配置MaxDegreeOfParallelism = 1保证组内顺序执行。同时引入闲置超时机制，自动回收长时间无消息的Block，避免内存泄漏。

代码实现 #

using System.Collections.Concurrent;
using System.Threading.Tasks.Dataflow;

public class GroupedMessageProcessor
{
    private readonly ConcurrentDictionary<string, BlockEntry> _blockDictionary = new();
    private readonly TimeSpan _idleTimeout = TimeSpan.FromMinutes(5); // 可根据业务调整闲置时长

    private class BlockEntry
    {
        public ActionBlock<Message> Block { get; }
        public DateTime LastActiveTime { get; set; }

        public BlockEntry(ActionBlock<Message> block)
        {
            Block = block;
            LastActiveTime = DateTime.UtcNow;
        }
    }

    public async Task ProcessMessageAsync(Message message)
    {
        if (message == null) throw new ArgumentNullException(nameof(message));

        // 原子性获取或创建对应GroupId的ActionBlock
        var entry = _blockDictionary.GetOrAdd(message.GroupId, key =>
        {
            var block = new ActionBlock<Message>(async msg =>
            {
                await msg.Process();
                // 处理完消息后更新活跃时间
                _blockDictionary[key].LastActiveTime = DateTime.UtcNow;
            }, new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 1,
                // Block完成后自动从字典移除
                Completion.ContinueWith(_ => _blockDictionary.TryRemove(key, out _))
            });

            // 启动该Block的闲置清理任务
            _ = CleanupIdleBlockAsync(key);

            return new BlockEntry(block);
        });

        // 更新活跃时间并发送消息到Block
        entry.LastActiveTime = DateTime.UtcNow;
        await entry.Block.SendAsync(message);
    }

    private async Task CleanupIdleBlockAsync(string groupId)
    {
        while (_blockDictionary.TryGetValue(groupId, out var entry))
        {
            await Task.Delay(_idleTimeout);

            var currentTime = DateTime.UtcNow;
            // 检查是否超过闲置时长
            if (currentTime - entry.LastActiveTime >= _idleTimeout)
            {
                // 停止接受新消息，处理剩余队列
                entry.Block.Complete();
                await entry.Block.Completion;
                break;
            }
        }
    }

    // 应用关闭时手动清理所有Block
    public async Task ShutdownAsync()
    {
        foreach (var entry in _blockDictionary.Values)
        {
            entry.Block.Complete();
        }
        await Task.WhenAll(_blockDictionary.Values.Select(e => e.Block.Completion));
        _blockDictionary.Clear();
    }
}

关键细节说明 #

• 线程安全：ConcurrentDictionary的GetOrAdd方法保证多线程下同一GroupId不会重复创建Block，避免竞态问题。
• 组内顺序执行：每个ActionBlock强制MaxDegreeOfParallelism = 1，确保同一组消息按发送顺序依次处理。
• 闲置回收机制：每个Block创建时启动独立的清理任务，定期检查最后活跃时间，超时后自动终止Block并从字典移除，释放内存。
• 内存泄漏防护：Block完成时通过Completion.ContinueWith自动移除字典条目，结合闲置清理，彻底避免无消息组长期占用资源。

使用示例 #

var processor = new GroupedMessageProcessor();

// 模拟从数据源批量接收消息并处理
foreach (var message in incomingMessages)
{
    await processor.ProcessMessageAsync(message);
}

// 应用关闭前执行清理
await processor.ShutdownAsync();

内容的提问来源于stack exchange，提问作者Jeremy Morren