代码之家 › 专栏 › 技术社区 › GamerDev

未调用TPL缓冲块消耗方法

bufferblock tpl-dataflow task-parallel-library multithreading c#

GamerDev · 技术社区 · 8 年前

here 和代码 here .

在下面的代码中,当使用此调用将新消息添加到缓冲块时,ConsumeSync方法永远不会被调用: _messageBufferBlock.SendAsync(message)

_messageBufferBlock。SendAsync(消息)

    public class PriorityMessageQueue 
    {
        private volatile ConcurrentDictionary<int,MyMessage> _queue = new ConcurrentDictionary<int,MyMessage>();
        private volatile BufferBlock<MyMessage> _messageBufferBlock;
        private readonly Task<bool> _initializingTask; // not used but allows for calling async method from constructor
        private int _dictionaryKey;

        public PriorityMessageQueue()
        {
            _initializingTask = Init();
        }

        public async Task<bool> EnqueueAsync(MyMessage message)
        {
            return await _messageBufferBlock.SendAsync(message);
        }

        private async Task<bool> ConsumeAsync()
        {
            try
            {
                // This code does not fire when a new message is added to the buffereblock
                while (await _messageBufferBlock.OutputAvailableAsync())
                {
                    // A message object is never received from the bufferblock
                    var message = await _messageBufferBlock.ReceiveAsync();


                }

                return true;
            }
            catch (Exception ex)
            {
                return false;
            }
        }

        private async Task<bool> Init()
        {
            var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = Environment.ProcessorCount,
                BoundedCapacity = 50
            };

            var prioritizeMessageBlock = new ActionBlock<MyMessage>(msg =>
            {
                SetMessagePriority(msg);
            }, executionDataflowBlockOptions);

            _messageBufferBlock = new BufferBlock<MyMessage>();
            _messageBufferBlock.LinkTo(prioritizeMessageBlock, new DataflowLinkOptions { PropagateCompletion = true, MaxMessages = 50});

            return await ConsumeAsync();
        }
    }

编辑我已经删除了所有额外的代码并添加了注释。

1 回复 | 直到 8 年前

JSteward 8 年前

我仍然不能完全确定你想要完成什么,但我会努力为你指明正确的方向。示例中的大多数代码都不是严格必需的。

如果这是您唯一的需求,那么我假设您只需要在传入新消息时运行一些任意代码。在数据流中实现这一点的最简单方法是使用 TransformBlock

public class PriorityMessageQueue {        
    private TransformBlock<MyMessage, MyMessage> _messageReciever;

    public PriorityMessageQueue() {
        var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions {
            MaxDegreeOfParallelism = Environment.ProcessorCount,
            BoundedCapacity = 50
        };

        var prioritizeMessageBlock = new ActionBlock<MyMessage>(msg => {
            SetMessagePriority(msg);
        }, executionDataflowBlockOptions);

        _messageReciever = new TransformBlock<MyMessage, MyMessage>(msg => NewMessageRecieved(msg), executionDataflowBlockOptions);
        _messageReciever.LinkTo(prioritizeMessageBlock, new DataflowLinkOptions { PropagateCompletion = true });
    }

    public async Task<bool> EnqueueAsync(MyMessage message) {
        return await _messageReciever.SendAsync(message);
    }

    private MyMessage NewMessageRecieved(MyMessage message) {
        //do something when a new message arrives

        //pass the message along in the pipeline
        return message;
    }

    private void SetMessagePriority(MyMessage message) {
        //Handle a message
    }
}

当然,你还有另一个选择,那就是立即做你需要做的事情 EnqueAsync 从返回任务之前 SendAsync 但是 TransformBlock变换块 给你额外的灵活性。