基于任务的编程，效率与应用程序的多个实例相同

Question

我已经创建了一个使用来自RabbitMq的消息的windows服务。消费是以异步方式完成的，其中消息被推送到服务。

它在水平上有很好的尺度，大约有一个。即使只是在同一台服务器上创建更多的服务实例，也可以提高线性性能。由于我们有一个关于吞吐量的定义目标，所以我发现通过简单地在windows服务中创建n个任务而不是部署n个服务实例来控制扩展更简单、更明智。

重写windows服务以简单地启动n个任务，每个任务运行"MQ消费“代码，因此本质上做的是windows服务的多个实例所做的事情，并不能产生相同的性能，实际上根本没有性能增益，实际上代码执行的开销似乎更大！

我知道有日程安排的问题，但底线问题是。是否有可能使用任务并获得“大致相同”的性能，就像启动应用程序的多个实例一样？，如果是的话，实现会是什么样的，因为我当前的设置扩展得很差。

一些代码片段：

这是基本的任务创建，每个任务都有一个任务包装器的实例，基本上是消费和处理消息的整个业务逻辑的抽象：

public void Start()
    {
        for (var i = 0; i < _levelOfConcurrency; i++)
        {
            var task = Task.Factory.StartNew(() => new TaskWrapper().TaskRun(_cancelationTokenSource.Token,_sleeptime),
                TaskCreationOptions.LongRunning);
            _tasks[i] = task;
            Console.WriteLine("Task created {0}", i);
        }
    }

 public void Stop()
    {
        _cancelationTokenSource.Cancel();
        Task.WaitAll(_tasks);
    }

这是每个任务在taskwrapper类中运行循环中的基本内容：

public void TaskRun(CancellationToken cancellationToken,TimeSpan sleeptime)
    {
        _semaphoreDataController.Start(); // Process messages async, fully selfcontained/threaded

        //Keep alive and occasionally check if cancelation is requested
        while (!cancellationToken.IsCancellationRequested)
        {
            Thread.Sleep(sleeptime);
        }
        Dispose();
    }

我希望有人能启发我：

Answer 1

你问题的核心是：

它几乎不取决于您的编程风格和程序本身。

可以通过一个多线程进程获得与多个实例相同的性能。但正如我说的，这取决于你的服务。我建议你阅读这本书的在线版本：用.net进行并行编程。它打开了我关于PLinq，任务和线程的眼睛。

但最终，它的所有运行在您的CPU和进程间的通信比线程之间的通信更加复杂。

Answer 2

因此，进一步的研究确定了规范中的主要限制因素。

我使用与RabbitMQ通信的方式是通过Burrow.NET，它将为应用程序创建一个与队列通信的专用通道。

这意味着，即使有许多并行任务正在运行，它们也只能从一个通道使用队列，从而限制了服务的速度。

具有多个服务实例，向RabbitMQ提供n个信道，从而获得线性性能增益。

Answer 3

DedicatedPublishingChannel作为其名称仅用于发布消息，因此我认为它不会造成使用消息的速度问题。此外，尽管只有一个通道可以从队列中订阅，但与处理这些消息的时间相比，发送到应用程序的消息的速度总是非常快。我开发了一个应用程序，将数百万条msgs (Tweet )从队列迁移到MongoDB，这一直是将数据写入数据库的瓶颈。

Burrow.NET处理任务管理，为您并行处理消息：

ITunnel tunnel;
// Create tunnel
tunnel.Subscribe(new SubscriptionOption<Message>
{
    BatchSize = 10, // Number of threads you want to process those messages
    MessageHandler = message => 
    {
        // Process your message here
    },
    QueuePrefetchSize = 10, // Should be double the message processing speed
    SubscriptionName = "Anything"
});
// Message will be acked automatically after finish

在我看来，我只会尽可能地使用有限的IO资源。我遇到了一个问题，我的应用程序创建了这么多的通道，并且吃了大量的服务器内存，如果应用程序创建了许多不必要的连接，那就更糟糕了，这就是为什么Burrow.NET将帮助您尽可能少建立连接/通道。

如果您认为您的消息可以处理得足够快，则可以使用高数字设置预取大小，并使用BatchSize (也称为线程计数)对其进行调优。例如，如果我的应用程序可以以5msgs/秒的速度处理来自队列的消息，我将保留QueuePrefetchSize = 10。

如果您将上述方法与auto一起使用，您将不必担心不会破坏消息，因为它会阻止即将出现的新消息。例如，您有QueuePrefetchSize = 10，让10个线程并行处理10条消息。您决定以异步方式对消息使用tunnel.SubscribeAsync，但由于某些原因(未处理的异常:D可能是)4的消息没有被添加，这相当于QueuePrefetchSize为6。因此，只能同时处理6条消息，并且适当地降低了速度。所以，请试试上面的方法，看看结果如何。