行为类似于线程的 C# 可重用或持久任务

Question

使用线程，您可以创建持久的、可重用的局部变量，这些变量对于客户端连接等非常有用。但是，对于 System.Threading.Tasks.Dataflow 中的 ActionBlock 之类的任务，似乎没有任何类型的操作块的持久性或可重用性。因此，对于涉及与客户端交互的 ActionBlock，我的理解是，您要么需要从头开始初始化客户端连接，要么在更高范围内重用一个客户端连接（使用锁定？）。

用例：我正在使用一个反转控制的 .NET 库。大部分逻辑（除了启动和关闭）必须位于名为 ProcessEventsAsync 的单个 Task 方法中，由库调用，该方法接收 IEnumerable 数据。 ProcessEventsAsync 必须对所有数据进行一些处理，然后将其发送给一些下游消费者。为了提高性能，我尝试使用 Tasks 并行化 ProcessEventsAsync 中的逻辑。我还想从这个任务中收集一些性能指标。

让我举一个详细的例子来说明我在做什么：

internal class MyClass
{

  private String firstDownStreamConnectionString;
  private String secondDownStreamConnectionString;
  private SomeClient firstClient;
  private SomeClient secondClient;
  private ReportingClient reportingClient;
  private int totalUnhandledDataCount;

  public MyClass(String firstDownStreamConnectionString, String secondDownStreamConnectionString, String reportingClientKey)
  {
      this.firstDownStreamConnectionString = firstDownStreamConnectionString;
      this.secondDownStreamConnectionString = secondDownStreamConnectionString;
      this.DegreeOfParallelism = Math.Max(Environment.ProcessorCount - 1, 1);
      this.reportingClient = new ReportingClient (reportingClientKey, DegreeOfParallelism);
      this.totalUnhandledDataCount = 0;
  }
  // called once when the framework signals that processing is about to be ready
  public override async Task OpenAsync(CancellationToken cancellationToken, PartitionContext context)
  {
    this.firstClient = SomeClient.CreateFromConnectionString(this.firstDownStreamConnectionString);
    this.secondClient = SomeClient.CreateFromConnectionString(this.secondDownStreamConnectionString );
    await Task.Yield();
  }

  // this is called repeatedly by the framework
  // outside of startup and shutdown, it is the only entrypoint to my logic
  public override async Task ProcessEventsAsync(CancellationToken cancellationToken, PartitionContext context, IEnumerable<Data> inputData)
  {
    ActionBlock<List<Data>> processorActionBlock = new ActionBlock<List<Data>>(
      inputData =>
      {
        SomeData firstDataset = new SomeData();
        SomeData secondDataset = new SomeData();
        int unhandledDataCount = 0;
        foreach (Data data in inputData)
        {
          // if data fits one set of criteria, put it in firstDataSet
          // if data fits other set of criteria, put it in secondDataSet
          // otherwise increment unhandledDataCount
        }
        Interlocked.Add(ref this.totalUnhandledDataCount, unhandledDataCount);
        lock (this.firstClient)
        {
          try
          {
            firstDataset.SendData(this.firstClient);
          } catch (Exception e)
          {
            lock(this.reportingClient)
            {
              this.reportingClient.LogTrace(e);
            }
          }
        }
        lock (this.secondClient)
        {
          try
          {
            secondDataset.SendData(this.secondClient);
          } catch (Exception e)
          {
            lock(this.reportingClient)
            {
              this.reportingClient.LogTrace(e);
            }
          }
        }
      },
      new ExecutionDataflowBlockOptions
      {
        MaxDegreeOfParallelism = this.DegreeOfParallelism
      });
    // construct as many List<Data> from inputData as there is DegreeOfParallelism
    // put that in a variable called batches
    for(int i = 0; i < DegreeOfParallelism; i++)
    {
      processorActionBlock.Post(batches[i]);
    }
    processorActionBlock.Complete();
    processorActionBlock.Completion.Wait();
    await context.CheckpointAsync();
  }
}

我试图只保留相关代码，我省略了处理逻辑、大多数指标收集、数据如何发送、关闭逻辑等。

我想利用一些允许重用性的Task。我不想为这种类型的所有正在运行的任务重用单个客户端连接，我也不希望每个任务在每次调用时都创建一个新的客户端连接。我确实希望每个类似线程的任务都有一组持久的客户端连接。理想情况下，我也不想在 System.Threading.Tasks.Dataflow 中创建一个包装 Task 或扩展抽象类/接口的新类。

Answer 1

您所描述的内容听起来像是异步委托或 Func。

例如：

Func<Task> TestFunc = async () =>
{
    Console.WriteLine("Begin");
    await Task.Delay(100);
    Console.WriteLine("Delay");
    await Task.Delay(100);
    Console.WriteLine("End");
};

如果函数在范围内，您只需：

await TestFunc();

您可以根据需要多次重复使用它。您还可以更改函数以接受参数。

---

编辑

您也可以尝试 AsyncLocal。根据文档：

因为基于任务的异步编程模型倾向于抽象线程的使用，所以可以使用 AsyncLocal 实例来跨线程持久化数据。

当与当前线程关联的值发生更改时，AsyncLocal 类还提供可选通知，因为它是通过设置 Value 属性显式更改的，或者当线程遇到等待或其他上下文转换时隐式更改。

Answer 2

听起来你只需要一个存储依赖项的类？

void Main()
{
    var doer1 = new ThingDoer();
    var doer2 = new ThingDoer();

    // A & B use one pair of clients, and C & D use another pair
    var taskA = doer1.DoTheThing();
    var taskB = doer1.DoTheThing();

    var taskC = doer2.DoTheThing();
    var taskD = doer2.DoTheThing();
}

public class ThingDoer
{
    private SomeClient _someClient;
    private SomeErrorReportingClient _someErrorReportingClient;

    public ThingDoer(SomeClient someClient, SomeErrorReportingClient someErrorReportingClient)
    {
        _someClient = someClient;
        _someErrorReportingClient = someErrorReportingClient;
    }

    public ThingDoer()
        : this(new SomeClient, new SomeErrorReportingClient)
    {
    }

    public async Task DoTheThing()
    {
        // Implementation here
    }
}

“可重用性”的概念与任务并不真正兼容。