Azure Worker角色以异步方式处理作业

2 回答

• 0

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

  对不起，我没有VS方便，但我希望你明白这一点 .

  回复于 2024-05-05T08:50:44+08:00
• 0

  谢谢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);
                      }
  
                  });
  

  UPDATE: 事实证明这仍然不能正常工作，我在早先测试时误解了结果 . 基于MyFixIt示例，在Work Role 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.WaitHandle.WaitOne实际上无法等到所有引用令牌的任务都完成，因此即使任务仍处于完成处理中，角色也会继续并停止 . 有什么方法可以在取消实际完成时正确使用令牌发出信号吗？

  谢谢！

  UPDATE 2

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

  /// <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();
      }
  

  回复于 2024-05-05T08:50:44+08:00