TPL DataFlow 一对一处理

Question

我有一个持续处理消息的系统。我想确保仅在处理先前的消息时才从外部队列请求消息。 让我们假设 GetMessages 方法从外部队列请求消息。

• 有事件1.会推送它
• 推1
• 有活动 2。将推动它 - 我的音乐会在这里。因为我们在处理之前的项目之前得到了项目
• 处理 1
• 已处理 1
• 已删除 1

代码：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace ConsoleApp1
{
    class Program
    {
        static void Main(string[] args)
        {
            EventProcessor a = new EventProcessor();
            Task task = Task.Run(async ()=> await a.Process());

            task.Wait();
        }
    }

    public class EventProcessor
    {
        private readonly TransformBlock<int, string> _startBlock;
        private readonly ActionBlock<string> _deleteBlock;
        private readonly ActionBlock<int> _recieveBlock;

        public EventProcessor()
        {
            var executionDataflowBlockOptions = new ExecutionDataflowBlockOptions {
                MaxDegreeOfParallelism = 1,
                BoundedCapacity = 1,
            };

            this._startBlock = new TransformBlock<int, string>(
                async @event => await this.ProcessNotificationEvent(@event),
                executionDataflowBlockOptions
            );

            this._deleteBlock = new ActionBlock<string>(async @event => {
                await this.DeleteMessage(@event);
            }, executionDataflowBlockOptions);
            var trashBin = DataflowBlock.NullTarget<string>();


            var dataflowLinkOptions = new DataflowLinkOptions {
                PropagateCompletion = true,
            };

            this._startBlock.LinkTo(
                this._deleteBlock,
                dataflowLinkOptions,
                (result => result != "o")
            );

            this._startBlock.LinkTo(
                trashBin,
                dataflowLinkOptions,
                (result => result == "o")
            );
        }

        private async Task<string> ProcessNotificationEvent(int @event)
        {
            Console.WriteLine($"Processing {@event}");
            await Task.Delay(5000);
            Console.WriteLine($"Processed {@event}");
            return @event.ToString();
        }

        public async Task Process()
        {

            //while (this._cancellationTokenSource.IsCancellationRequested == false) {
            foreach (var notificationEvent in GetMessages()) {
                Console.WriteLine($"Got event {notificationEvent}. Will push it");
                if (await this._startBlock.SendAsync(notificationEvent) == false) {
                    Console.WriteLine($"Failed to push {notificationEvent}");
                    return;
                }
                Console.WriteLine($"Pushed {notificationEvent}");
            }
            //}
            this._startBlock.Complete();
            this._deleteBlock.Completion.Wait();
        }

        private static IEnumerable<int> GetMessages() {
            return Enumerable.Range(1, 5);
        }

        private async Task DeleteMessage(string @event)
        {
            Console.WriteLine($"Deleted {@event}");
        }
    }
}

输出将是

Got event 1. Will push it
Pushed 1
Got event 2. Will push it
Processing 1
Processed 1
Deleted 1
Processing 2
Pushed 2
Got event 3. Will push it
Processed 2
Processing 3
Deleted 2
Pushed 3
Got event 4. Will push it
Processed 3
Deleted 3
Processing 4
Pushed 4
Processed 4
Deleted 4
Press any key to continue . . .

我认为我可以为每条消息创建 TDL 数据流，但我认为这有点过头了。

Answer 1

问题是您有一个缓冲区，因此您的生产者循环将始终在处理第一个项目时处理下一个项目。这是使用 TPL 数据流的自然结果。

如果您想一次处理一个，最简单的方法可能是删除 TPL 数据流：

public class EventProcessor
{
  private async Task<string> ProcessNotificationEvent(int @event)
  {
    Console.WriteLine($"Processing {@event}");
    await Task.Delay(5000);
    Console.WriteLine($"Processed {@event}");
    return @event.ToString();
  }

  public async Task Process()
  {
    foreach (var notificationEvent in GetMessages()) {
      Console.WriteLine($"Got event {notificationEvent}. Will push it");
      var result = await this.ProcessNotificationEvent(notificationEvent);
      if (result != "o")
        await DeleteMessage(result);
    }
  }

  private static IEnumerable<int> GetMessages() => Enumerable.Range(1, 5);

  private async Task DeleteMessage(string @event) => Console.WriteLine($"Deleted {@event}");
}