任务可靠性（或者我应该做其他事情吗？）

Question

我有一个使用 Nancy 和 Nancy.Hosting.Self 的 C# 控制台应用程序。

这个想法是它将通过 Nancy 提供 API，并且主应用程序将定期轮询到各种应用程序的多个连接 + 在通过 API（通过 Nancy）请求时从这些连接中获取数据。

所以我将有 2 个正在运行的进程，持续轮询和 HTTP 服务器。

我的 Program.cs 包含以下 sn-ps。

Task pollTask = null;
try {
  pollTask = Task.Run(async () => {
    while (processTask) {
      connectionPool.PollEvents();

      await Task.Delay(configLoader.config.connectionPollDelay, wtoken.Token);
    }
    keepRunning = false;
  }, wtoken.Token);
}
catch (AggregateException ex) {
  Console.WriteLine(ex);
}
catch (System.Threading.Tasks.TaskCanceledException ex) {
  Console.WriteLine("Task Cancelled");
  Console.WriteLine(ex);
}

后来……

using (var host = new Nancy.Hosting.Self.NancyHost(hostConfigs, new Uri(serveUrl))) {
  host.Start();
  // ...
  // routinely checking console for a keypress to quit which then sets
  // processTask to false, which would stop the polling task, which
  // in turn sets keepRunning to false which stops the application entirely.
}

轮询任务似乎只是死亡/停止，没有任何输出到控制台以指示它停止的原因。在检查控制台输入按键时，我还查询了 pollTask​​.Status，它最终详细说明了“故障”。但我不知道为什么。我也在质疑长期/永久运行的任务的可靠性。

为了防止这种含糊不清，我有一个主要问题。 Task 是否适合以上述方式永久运行的任务。如果不是，我应该使用什么来实现 2 个并行进程，其中一个是 Nancy。

更新（2018 年 7 月 17 日）：
在采纳了迄今为止的建议和答案之后，我已经能够确定最终发生的异常并终止进程：

PollEvents process appears to be throwing an exception...
System.AggregateException: One or more errors occurred. ---> System.InvalidOperationException: There were not enough free threads in the ThreadPool to complete the operation.
   at System.Net.HttpWebRequest.BeginGetRequestStream(AsyncCallback callback, Object state)
   at System.Net.Http.HttpClientHandler.StartGettingRequestStream(RequestState state)
   at System.Net.Http.HttpClientHandler.PrepareAndStartContentUpload(RequestState state)
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at ApiProxy.ServiceA.Connection.<>c__DisplayClass22_0.<<Heartbeat>b__0>d.MoveNext() in \api-proxy\src\ServiceA\Connection.cs:line 281
   --- End of inner exception stack trace ---
   at System.Threading.Tasks.Task.ThrowIfExceptional(Boolean includeTaskCanceledExceptions)
   at System.Threading.Tasks.Task.Wait(Int32 millisecondsTimeout, CancellationToken cancellationToken)
   at ApiProxy.ServiceA.Connection.Heartbeat() in \api-proxy\src\ServiceA\Connection.cs:line 274
   at ApiProxy.ServiceA.Connection.PollEvents(Nullable`1 sinceEventId) in \api-proxy\src\ServiceA\Connection.cs:line 313
   at ApiProxy.ConnectionPool.PollEvents() in \api-proxy\src\ConnectionPool.cs:line 50
   at ApiProxy.Program.<>c.<<Main>b__5_0>d.MoveNext() in \api-proxy\Program.cs:line 172
---> (Inner Exception #0) System.InvalidOperationException: There were not enough free threads in the ThreadPool to complete the operation.
   at System.Net.HttpWebRequest.BeginGetRequestStream(AsyncCallback callback, Object state)
   at System.Net.Http.HttpClientHandler.StartGettingRequestStream(RequestState state)
   at System.Net.Http.HttpClientHandler.PrepareAndStartContentUpload(RequestState state)
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.CompilerServices.TaskAwaiter.ThrowForNonSuccess(Task task)
   at System.Runtime.CompilerServices.TaskAwaiter.HandleNonSuccessAndDebuggerNotification(Task task)
   at ApiProxy.ServiceA.Connection.<>c__DisplayClass22_0.<<Heartbeat>b__0>d.MoveNext() in \api-proxy\src\ServiceA\Connection.cs:line 281<---

因为try...catch 现在位于任务中，这意味着虽然此错误反复发生并快速发生，但最终它似乎会自行纠正。然而，在要求更多线程池之前能够查询线程池的可用性将是理想的。看起来 ThreadPool 问题与我的代码无关，而是与 Nancy 无关。

查找错误的来源后，我发现它发生在以下情况：

public bool Heartbeat() {
  if (connectionConfig.events.heartbeatUrl == "") {
    return false;
  }

  var val = false;
  var task = Task.Run(async () => {
    var heartbeatRequest = new HeartbeatRequest();
    heartbeatRequest.host = connectionConfig.host;
    heartbeatRequest.name = connectionConfig.name;
    heartbeatRequest.eventId = lastEventId;

    var prettyJson = JToken.Parse(JsonConvert.SerializeObject(heartbeatRequest)).ToString(Formatting.Indented);
    var response = await client.PostAsync(connectionConfig.events.heartbeatUrl, new StringContent(prettyJson, Encoding.UTF8, "application/json"));

    // todo: create a heartbeatResponse extending a base response type
    PingResponse heartbeatResponse = JsonConvert.DeserializeObject<PingResponse>(await response.Content.ReadAsStringAsync());
    if (heartbeatResponse != null) {
      Console.WriteLine("Heartbeat: " + heartbeatResponse.message);
      val = heartbeatResponse.success;
    }
    else {
      // todo: sentry?
    }
  });
  task.Wait();

  return val;
}

我将调用包装在 Task 中，因为否则我最终会得到到处都是 async 定义的海洋。 这是线程池饥饿的可能来源吗？

更新 2
通过删除包装PostAsync 的Task.Run 更正了上述代码。然后调用代码将调用Heartbeat().Wait()，因此该方法现在看起来像：

public async Task<bool> Heartbeat() {
  if (connectionConfig.events.heartbeatUrl == "") {
    return false;
  }

  var val = false;
  var heartbeatRequest = new HeartbeatRequest();
  heartbeatRequest.host = connectionConfig.host;
  heartbeatRequest.name = connectionConfig.name;
  heartbeatRequest.eventId = lastEventId;

  var prettyJson = JToken.Parse(JsonConvert.SerializeObject(heartbeatRequest)).ToString(Formatting.Indented);
  var response = await client.PostAsync(connectionConfig.events.heartbeatUrl, new StringContent(prettyJson, Encoding.UTF8, "application/json"));

  PingResponse heartbeatResponse = JsonConvert.DeserializeObject<PingResponse>(await response.Content.ReadAsStringAsync());
  if (heartbeatResponse != null) {
    Console.WriteLine("Heartbeat: " + heartbeatResponse.message);
    val = heartbeatResponse.success;
  }
  else {
    // todo: sentry?
  }

  return val;
}

希望我的一些经验可以帮助其他人。我还不确定上述更改（有很多这样的更改）是否会防止线程饥饿。

Answer 1

当任务失败时……也就是内部发生异常时

pollTask = Task.Run(async () => {
        while (processTask) {
            connectionPool.PollEvents();

            await Task.Delay(configLoader.config.connectionPollDelay, wtoken.Token);
        }
        keepRunning = false;
    }, wtoken.Token);

异常将存储在返回的任务中。即Task.Exception。并且该任务将被标记为故障。

除非您调用Task.Wait、await 任务或类似名称，否则不会在您的上下文中引发此异常。请参阅下文了解我建议您使用的内容。

---

connectionPool.PollEvents(); 很可能最终会抛出一些异常，因为我上面提到的，你不会捕捉到。

如果您无法阻止异常，您可能希望在任务中处理它...除非我们谈论的条件意味着要拆除 connectionPool 或更激烈的事情。

我不知道是什么让PollEvents 跳脱。一种可能性是您在其他地方使用了相同的对象，它不是线程安全的。

说到线程安全。我希望processTask 和keepRunning 是volatile。虽然，正如您将在下面看到的，您不需要它们。

---

关于长时间运行的任务，请使用这种方式：

Task.Factory.StartNew(mehtod, TaskCreationOptions.LongRunning);

或者任何Factory.StartNew 的重载都需要TaskCreationOptions。

在内部，当您使用 TaskCreationOptions.LongRunning 创建任务时，它会将 Thread 专用于您的任务，而不是从线程池 (preventing starvation) 中窃取一个，并确保在该设置下一切正常。

---

附录：StartNew is dangerous，WBuck 表示容易出错。除了启动一个任务，Task.Run 将进行错误处理（我告诉你这样做）并设置TaskCreationOptions.DenyChildAttach。

还有一个问题是它无法识别async 方法（没有重载需要Func<Task<TResult>>，并且当您通过Func<Task> 时使用Func<TResult> 是一个问题）。

因此，直接在StartNew 中使用异步方法是个坏主意。当然，天真的方法是将 async 方法包装在常规方法中：

Task.Factory.StartNew
(
    ()=>
    {
        awyncMethod().Wait();
    },
    TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach
);

然而，这违背了目的，因为现在异步方法在线程池中，我们创建了一个线程只是为了等待它。

当我们将async 方法传递给Factory.StartNew 时会发生什么，我们得到了一个任务，它代表了您真正想要的任务的创建。让我们称之为“faketask”。这个 faketask 将立即完成，而您想要的任务就是结果......要正确使用它，您需要向 faketask 添加一个延续（使用适当的创建选项），以检索实际任务。此外，理想情况下，您希望此延续作为实际任务的代理（为您提供它返回的内容或抛出的异常）。谢天谢地，TaskExtensions.Unwrap 做到了。

因此，我们得出这样的结论：

Task.Factory.StartNew
(
    async ()=>
    {
        /*...*/
    },
    TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach
).Unwrap();

另见Parallel Programming with .NET - Task.Run vs Task.Factory.StartNew。

---

如果您打算保留其中的多个，都取自同一个connectionPool...首先要确保connectionPool 是线程安全的。除此之外，您还需要一种方法来监视任务的状态并重新启动它们。使用另一个长时间运行的任务，进入所有这些的Task.WaitAny 并处理异常 - 并进行日志记录 - 只要应用程序应该保持活动状态，就会重新启动它们。 这就是我建议你用来等待任务的方法。

另外，你可以使用CancellationToken退出循环，检查CancellationToken.IsCancellationRequested。要知道任务已停止，您可以查看Task.IsCompleted。 这就是你不需要这些变量的原因。

附录：其实你可以用一个CancellationToken把它全部拆掉。

Answer 2

正如 cmets 中指出的那样，您的 try-catch 是无用的。您需要等待任务完成才能查看是否引发了异常。

作为一个建议，与其在无限循环中运行，为什么不使用计时器来定期轮询？

var pollTask = Task.Run(async () => 
 { 
    while (processTask) 
    {
      wToken.ThrowIfCancellationRequested();
      connectionPool.PollEvents();
      await Task.Delay(configLoader.config.connectionPollDelay, wtoken.Token);
    }
  keepRunning = false;
}, wtoken.Token);

try 
{
     pollTask.Wait(wToken);
}
catch( AggregateException ex )
{
    // Handle exception
}

或者，如果您的方法被标记为异步，您可以等待任务。 Await 将为您解开聚合异常。

try 
{
 await Task.Run(async () => 
 { 
    while (processTask) 
    {
      wToken.ThrowIfCancellationRequested();
      connectionPool.PollEvents();
      await Task.Delay(configLoader.config.connectionPollDelay, wtoken.Token);
    }
  keepRunning = false;
}, wtoken.Token);
}
catch( Exception ex )
{
    // Handle exception
}

为了完整起见，您还可以继续检查任务的异常属性。

Task.Run( async ( ) =>
        {
            wToken.ThrowIfCancellationRequested( );
            connectionPool.PollEvents( );
            await Task.Delay( configLoader.config.connectionPollDelay, wToken );
        }, wToken ).ContinueWith( task =>
        {
            if( task.IsFaulted )
            {
                // Inspect the exception property of the task
                // to view the exception / exceptions.
                Console.WriteLine( task.Exception?.InnerException );
            }
        }, wToken );