用于异步处理作业的Azure Worker角色

发生这种情况的原因是因为您没有执行从ProcessMessage返回的任务。因此，ProcessMessageAsync可以在ProcessMessage优雅地完成或取消之前完成。请记住，您不希望等待ProcessMessage，因为它将使消息处理顺序，因此我建议您保留一个正在运行的任务列表。换句话说，在ProcessMessageAsync中创建一个列表，并将从ProcessMessage返回的任务添加到该列表中。然后在while循环结束时，如果token被取消，则应该循环遍历此列表以取消所有挂起的任务。

很抱歉，我手边没有VS，但我希望你能明白。

谢谢你Sanjay，根据你的建议，我提出了以下建议。

      /// <summary>
    /// Continuous loop that monitors the queue and launches jobs when they are retrieved.
    /// </summary>
    /// <param name="token"></param>
    /// <returns></returns>
    public virtual async Task ProcessMessagesAsync(CancellationToken token)
    {
        CloudQueue queue = _queueClient.GetQueueReference(_queueName);
        await queue.CreateIfNotExistsAsync(token);
        var runningTasks = new ConcurrentDictionary<int, Task>();
        while (!token.IsCancellationRequested)
        {
            Debug.WriteLine("inLoop");
            // The default timeout is 90 seconds, so we won’t continuously poll the queue if there are no messages.
            // Pass in a cancellation token, because the operation can be long-running.
            CloudQueueMessage message = await queue.GetMessageAsync(token);
            if (message != null)
            {
                var t = ProcessMessage(message, queue, token);
                var c = t.ContinueWith(z => RemoveRunningTask(t.Id, runningTasks));
                while (true)
                {
                    if (runningTasks.TryAdd(t.Id, t))
                        break;
                    Task.Delay(25);
                }                                    
            }                    
            else
            {
                try
                {
                    await Task.Delay(500, token);
                }
                catch (Exception ex)
                {
                    Debug.WriteLine(ex.Message);
                }
            }
        }
        while (!runningTasks.IsEmpty)
        {
            Debug.WriteLine("Waiting for running tasks");
            Task.Delay(500);
        }
    }
    private static void RemoveRunningTask(int id, ConcurrentDictionary<int, Task> runningTasks)
    {
        while (true)
        {
            Task outTask;
            if (runningTasks.TryRemove(id, out outTask))
                break;
            Task.Delay(25);
        }
    }

这似乎工作，虽然我觉得它是有点笨拙。我开始像这样编码'ContinueWith'，但惊讶的是，传入的任务有一个不同的Id值(我期望它是相同的任务):

                    var task = ProcessMessage(message, queue, token).ContinueWith(x =>
                {
                    while (true)
                    {
                        Task outTask;
                        if (runningTasks.TryRemove(x.Id, out outTask))
                            break;
                        Task.Delay(25);
                    }
                });

更新:事实证明，这仍然不太工作，我不知何故误读了测试时的结果。根据MyFixIt的例子，在工作角色OnStop中，我有以下代码:

        public override void OnStop()
    {
        Debug.WriteLine("OnStop_Begin");
        tokenSource.Cancel();
        tokenSource.Token.WaitHandle.WaitOne();
        base.OnStop();
        Debug.WriteLine("Onstop_End");
        tokenSource.Dispose();
    }

看起来tokensource . token . waithhandle . waitone并不能真正等到所有对令牌有引用的任务都完成，所以角色继续并停止，即使任务仍处于结束的处理中。是否有一些方法可以正确地使用令牌来发出取消实际完成的信号?

谢谢!

更新2

好吧，我想我有一个解决方案，现在是有效的。看起来CancellationToken。调用。cancel时发出WaitHandle信号，所以我不确定在调用。cancel后立即使用它的目的是什么，似乎它总是会立即通过该代码继续?这就是FixIt示例中的情况，但我真的不理解它。出于我的目的，我已经将ProcessMessagesAsync更改为现在在ManualResetEventSlim中传递，然后在所有任务完成后设置它。然后在OnStop中，我在完成Stop之前等待。

       /// <summary>
    /// Continuous loop that monitors the queue and launches jobs when they are retrieved.
    /// </summary>
    /// <param name="token"></param>
    /// <returns></returns>
    public virtual async Task ProcessMessagesAsync(CancellationToken token, ManualResetEventSlim reset)
    {
        CloudQueue queue = _queueClient.GetQueueReference(_queueName);
        await queue.CreateIfNotExistsAsync(token);
        var runningTasks = new ConcurrentDictionary<int, Task>();
        while (!token.IsCancellationRequested)
        {
            Debug.WriteLine("inLoop");
            // The default timeout is 90 seconds, so we won’t continuously poll the queue if there are no messages.
            // Pass in a cancellation token, because the operation can be long-running.
            CloudQueueMessage message = await queue.GetMessageAsync(token);
            if (message != null)
            {
                var t = ProcessMessage(message, queue, token);
                var c = t.ContinueWith(z => RemoveRunningTask(t.Id, runningTasks));

                while (true)
                {
                    if (runningTasks.TryAdd(t.Id, t))
                        break;
                    await Task.Delay(25);
                }                                    
            }                    
            else
            {
                try
                {
                    await Task.Delay(500, token);
                }
                catch (Exception ex)
                {
                    Debug.WriteLine(ex.Message);
                }
            }
        }
        while (!runningTasks.IsEmpty)
        {
            Debug.WriteLine("Waiting for running tasks");
            await Task.Delay(500);
        }
        Debug.WriteLine("All tasks have finished, exiting ProcessMessagesAsync.");
        reset.Set();
    }
        public override void OnStop()
    {
        Debug.WriteLine("OnStop_Begin");
        tokenSource.Cancel();
        tokenSource.Token.WaitHandle.WaitOne();
        _reset.Wait();
        base.OnStop();
        Debug.WriteLine("Onstop_End");
        tokenSource.Dispose();
    }