如何每隔 X 秒批量处理来自服务总线队列的消息

Question

我的服务总线队列之前一次处理一条消息。 我想更改它以允许消息排队，这样我就可以每 X 秒批量处理它们。我最初是这样做的：

var messagingFactorySettings = new MessagingFactorySettings
{
    NetMessagingTransportSettings = { 
        BatchFlushInterval = TimeSpan.FromMilliseconds(10000) },
        TokenProvider = credentials
};

现在我想接收消息，但似乎我必须执行这个无限循环来获取它们：

while ((messages = myQueueClient.ReceiveBatch(1000).ToList()) != null)
{
    foreach (var message in messages)
    { ...

据我了解，BatchFlushInterval 将允许请求累积 X 时间，因此我不会一一接收消息，而是批量接收消息。因此，我不确定为什么我不能像以前那样做：

myQueueClient.OnMessage((m) =>
{

但是批量版本：

myQueueClient.OnBulkMessage((listOfMessages) =>
{

我是否遗漏了什么，或者是不断轮询完成此任务的唯一方法？我的 BatchFlushInterval 似乎也被忽略了。我预计它只会每 10 秒检查一次新消息，但它会立即接收第一批消息，并且会立即处理任何进来的新消息。

假设我想每隔 X（即：1）秒从队列中提取 Y 条消息（即：1000）并立即处理它们，我该怎么做？ 为什么BatchFlushInterval 没有任何影响吗？

Answer 1

看来，一个简单的Thread.Sleep(x) 就可以了。

我发现暂停直到总时间过去的问题很有趣，所以我开始实施一个秒表子类，使这个问题可以以更清晰、更易读的方式解决。

while ((var messages = myQueueClient.ReceiveBatch(1000)) != null)
{
    var sw = WaitableStopwatch.StartNew();

    // ReceiveBatch() return IEnumerable<>. No need for .ToList().
    foreach (var message in messages)
    {
        ...
    }

    // If processing took less than 10 seconds, sleep
    // for the remainder of that time span before getting
    // the next batch.
    sw.Wait(Timespan.FromSeconds(10));
}



/// <summary>
/// Extends Stopwatch with the ability to wait until a specified
/// elapsed time has been reached.
/// </summary>
public class WaitableStopwatch : Stopwatch
{
    /// <summary>
    /// Initializes a new WaitableStopwatch instance, sets the elapsed
    /// time property to zero, and starts measuring elapsed time.
    /// </summary>
    /// <returns>A WaitableStopwatch that has just begun measuring elapsed time.</returns>
    public static new WaitableStopwatch StartNew()
    {
        WaitableStopwatch sw = new WaitableStopwatch();

        sw.Start();

        return sw;
    }

    /// <summary>
    /// Waits until the ElapsedMilliseconds property reaches <paramref name="elapsedMilliseconds"/>.
    /// </summary>
    /// <param name="elapsedMilliseconds"></param>
    public void Wait(int elapsedMilliseconds)
    {
        Wait(TimeSpan.FromMilliseconds(elapsedMilliseconds));
    }

    /// <summary>
    /// Waits until when the Elapsed property reaches <paramref name="elapsed"/>.
    /// </summary>
    /// <param name="elapsed"></param>
    public void Wait(TimeSpan elapsed)
    {
        TimeSpan diff;

        while ((diff = elapsed - this.Elapsed) > TimeSpan.Zero)
        {
            Thread.Sleep(diff);
        }
    }

    /// <summary>
    /// Waits until when the ElapsedMilliseconds property reaches <paramref name="elapsedMilliseconds"/>.
    /// </summary>
    /// <param name="elapsedMilliseconds"></param>
    public Task WaitAsync(int elapsedMilliseconds)
    {
        return WaitAsync(TimeSpan.FromMilliseconds(elapsedMilliseconds));
    }

    /// <summary>
    /// Waits until when the Elapsed property reaches <paramref name="elapsed"/>.
    /// </summary>
    /// <param name="elapsed"></param>
    public async Task WaitAsync(TimeSpan elapsed)
    {
        TimeSpan diff;

        while ((diff = elapsed - this.Elapsed) > TimeSpan.Zero)
        {
            await Task.Delay(diff);
        }
    }
}