TPL DataFlow 一一处理

我有一个持续处理消息的系统.我想确保仅在处理上一条消息时才从外部队列请求消息.让我们想象一下 GetMessages 方法从外部队列请求消息.

I am having system that continuously processing messages. I want to make sure that I request messages from an external queue only when previous message was processed. Lets imagine that GetMessages method requests messages from external queue.

• 有事件 1. 将推送它
• 推送 1
• 有活动 2.将推动它 - 我的音乐会到了.由于我们在处理之前获得项目
• 处理 1
• 已处理 1
• 已删除 1 个

• Got event 1. Will push it
• Pushed 1
• Got event 2. Will push it - my concert is here. As we get item before processing previous
• Processing 1
• Processed 1
• Deleted 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 DataFlow，但我认为这有点矫枉过正.

I thought that i can create TDL DataFlow for the each message, but i think it will be an overkill.