异步/等待或任务。在控制台应用程序/ Windows服务中运行

Question

我一直在研究（包括查看此主题的所有其他SO帖子）实现（最可能的）Windows Service worker的最佳方法，该服务将从数据库中提取工作项并在“解雇后-forget'方式在后台进行（工作项管理全部以异步方法处理）。工作项将是Web服务调用和数据库查询。这些工作项的生产者将受到一些限制，以确保某种可衡量的方法来安排工作。下面的示例非常基础，仅用来突出while循环和for循环的逻辑。哪种方法比较理想或没关系？有没有更合适/更有效的方法来实现这一目标？

异步/等待...

    private static int counter = 1;

    static void Main(string[] args)
    {
        Console.Title = "Async";

        Task.Run(() => AsyncMain());

        Console.ReadLine();            
    }

    private static async void AsyncMain()
    {
        while (true)
        {
            // Imagine calling a database to get some work items to do, in this case 5 dummy items
            for (int i = 0; i < 5; i++)
            {
                var x = DoSomethingAsync(counter.ToString());

                counter++;
                Thread.Sleep(50);
            }

            Thread.Sleep(1000);
        }
    }

    private static async Task<string> DoSomethingAsync(string jobNumber)
    {
        try
        {
            // Simulated mostly IO work - some could be long running
            await Task.Delay(5000);
            Console.WriteLine(jobNumber);
        }
        catch (Exception ex)
        {
            LogException(ex);
        }

        Log("job {0} has completed", jobNumber);

        return "fire and forget so not really interested";
    }

Task.Run ...

    private static int counter = 1;

    static void Main(string[] args)
    {
        Console.Title = "Task";

        while (true)
        {
            // Imagine calling a database to get some work items to do, in this case 5 dummy items
            for (int i = 0; i < 5; i++)
            {
                var x = Task.Run(() => { DoSomethingAsync(counter.ToString()); });

                counter++;
                Thread.Sleep(50);
            }

            Thread.Sleep(1000);
        }
    }

    private static string DoSomethingAsync(string jobNumber)
    {
        try
        {
            // Simulated mostly IO work - some could be long running
            Task.Delay(5000);
            Console.WriteLine(jobNumber);
        }
        catch (Exception ex)
        {
            LogException(ex);
        }

        Log("job {0} has completed", jobNumber);

        return "fire and forget so not really interested";
    }

Answer 1

从数据库中提取工作项目并在后台以“即发即忘”的方式并行异步处理它们

从技术上讲，您需要并发。你想要异步并发还是并行并发还有待观察......

工作项将是 Web 服务调用和数据库查询。

这项工作受 I/O 限制，因此这意味着异步并发是更自然的方法。

将对这些工作项的生产者进行一些限制，以确保采用某种可衡量的方法来安排工作。

这里隐含了生产者/消费者队列的概念。这是一种选择。TPL Dataflow 提供了一些不错的生产者/消费者队列，它们与异步兼容并支持节流。

或者，您可以自己进行节流。对于异步代码，有一个称为SemaphoreSlim.

TPL 数据流方法，带节流：

private static int counter = 1;

static void Main(string[] args)
{
    Console.Title = "Async";
    var x = Task.Run(() => MainAsync());
    Console.ReadLine();          
}

private static async Task MainAsync()
{
  var blockOptions = new ExecutionDataflowBlockOptions
  {
    MaxDegreeOfParallelism = 7
  };
  var block = new ActionBlock<string>(DoSomethingAsync, blockOptions);
  while (true)
  {
    var dbData = await ...; // Imagine calling a database to get some work items to do, in this case 5 dummy items
    for (int i = 0; i < 5; i++)
    {
      block.Post(counter.ToString());
      counter++;
      Thread.Sleep(50);
    }
    Thread.Sleep(1000);
  }
}

private static async Task DoSomethingAsync(string jobNumber)
{
  try
  {
    // Simulated mostly IO work - some could be long running
    await Task.Delay(5000);
    Console.WriteLine(jobNumber);
  }
  catch (Exception ex)
  {
    LogException(ex);
  }
  Log("job {0} has completed", jobNumber);
}

手动节流的异步并发方法：

private static int counter = 1;
private static SemaphoreSlim semaphore = new SemaphoreSlim(7);

static void Main(string[] args)
{
    Console.Title = "Async";
    var x = Task.Run(() => MainAsync());
    Console.ReadLine();          
}

private static async Task MainAsync()
{
  while (true)
  {
    var dbData = await ...; // Imagine calling a database to get some work items to do, in this case 5 dummy items
    for (int i = 0; i < 5; i++)
    {
      var x = DoSomethingAsync(counter.ToString());
      counter++;
      Thread.Sleep(50);
    }
    Thread.Sleep(1000);
  }
}

private static async Task DoSomethingAsync(string jobNumber)
{
  await semaphore.WaitAsync();
  try
  {
    try
    {
      // Simulated mostly IO work - some could be long running
      await Task.Delay(5000);
      Console.WriteLine(jobNumber);
    }
    catch (Exception ex)
    {
      LogException(ex);
    }
    Log("job {0} has completed", jobNumber);
  }
  finally
  {
    semaphore.Release();
  }
}

最后一点，我几乎从来没有推荐过我自己的关于 SO的书，但我认为它真的会让你受益。特别是第 8.10 节（阻塞/异步队列）、11.5（节流）和 4.4（节流数据流块）。