17

交叉发布到http://social.msdn.microsoft.com/Forums/en-US/tpldataflow/thread/89b3f71d-3777-4fad-9c11-50d8dc81a4a9

我知道......我并没有真正发挥 TplDataflow 的最大潜力。ATM 我只是BufferBlock用作消息传递的安全队列，其中生产者和消费者以不同的速率运行。我看到一些奇怪的行为让我不知道如何进行。

private BufferBlock<object> messageQueue = new BufferBlock<object>();

public void Send(object message)
{
    var accepted=messageQueue.Post(message);
    logger.Info("Send message was called qlen = {0} accepted={1}",
    messageQueue.Count,accepted);
}

public async Task<object> GetMessageAsync()
{
    try
    {
        var m = await messageQueue.ReceiveAsync(TimeSpan.FromSeconds(30));
        //despite messageQueue.Count>0 next line 
        //occasionally does not execute
        logger.Info("message received");
        //.......
    }
    catch(TimeoutException)
    {
        //do something
    }
}

在上面的代码中（它是 2000 行分布式解决方案的一部分），Send每 100 毫秒左右定期调用一次。这意味着一个项目以每秒 10 次左右的速度被Post编辑。messageQueue这是经过验证的。但是，偶尔会出现ReceiveAsync在超时内没有完成（即Post没有导致ReceiveAsync完成）并且TimeoutException在 30 秒后被提升。此时，messageQueue.Count已是数百。这是出乎意料的。在较慢的发布速度（1 个帖子/秒）中也观察到此问题，并且通常发生在 1000 个项目通过BufferBlock.

所以，为了解决这个问题，我使用了以下代码，它可以工作，但在接收时偶尔会导致 1s 延迟（由于上述错误发生）

    public async Task<object> GetMessageAsync()
    {
        try
        {
            object m;
            var attempts = 0;
            for (; ; )
            {
                try
                {
                    m = await messageQueue.ReceiveAsync(TimeSpan.FromSeconds(1));
                }
                catch (TimeoutException)
                {
                    attempts++;
                    if (attempts >= 30) throw;
                    continue;
                }
                break;

            }

            logger.Info("message received");
            //.......
        }
        catch(TimeoutException)
        {
            //do something
        }
   }

对我来说，这看起来像是 TDF 中的竞争条件，但我无法深入了解为什么在我BufferBlock以类似方式使用的其他地方不会发生这种情况。实验性地从ReceiveAsyncto改变Receive并没有帮助。我没有检查过，但我想孤立地看，上面的代码可以完美运行。这是我在“TPL 数据流简介” tpldataflow.docx中看到的一种模式。

我能做些什么来解决这个问题？是否有任何指标可以帮助推断正在发生的事情？如果我无法创建可靠的测试用例，我还能提供哪些信息？

帮助！

4

1 回答 1

1

Stephen seems to think the following is the solution

var m = await messageQueue.ReceiveAsync();

instead of:

var m = await messageQueue.ReceiveAsync(TimeSpan.FromSeconds(30));

Can you confirm or deny this?

于 2012-05-16T15:32:56.600 回答