并行消费来自一个队列的消息

Question

我有一个带有N 消息的队列。现在我想创建一个控制台应用程序（消费者），它创建给定数量的任务/线程，每个线程将从队列中获取 1 条消息，处理它并获取一条新消息......直到队列为空。

到目前为止，这就是我所拥有的：

private static void Main(string[] args)
{
    var runner = new Runner();
    for (var j = 0; j < 5; j++)
    {
        var task = Task.Factory.StartNew(() => { runner.ReceiveMessageFromServer(); });
    }
}

和

public class QRegistrator : IDisposable
{
    private const string activeMqUri = "activemq:tcp://localhost:61616";

    private readonly IConnection _connection;

    private readonly ISession _session;

    public QRegistrator(/*string activeMqUri*/)
    {
        var connectionFactory = new ConnectionFactory(activeMqUri);
        _connection = connectionFactory.CreateConnection();
        _connection.Start();
        _session = _connection.CreateSession();
    }

    public void Dispose()
    {
        _connection?.Close();
    }

    public IMessageConsumer CreateConsumer(string queueName, string topicName)
    {
        //there have to be new session for each concurrent consumer
        var session = _connection.CreateSession();
        var queue = new ActiveMQQueue(queueName);
        var consumer = string.IsNullOrWhiteSpace(topicName) ? session.CreateConsumer(queue) : session.CreateConsumer(queue, $"topic = '{topicName}'");
        return consumer;
    }
}

在课堂上Runner

public void ReceiveMessageFromServer()
{
    var consumer = _registrator.CreateConsumer("myQueue", null);

    while (true)
    {
        ITextMessage message = consumer.Receive() as ITextMessage;
        if (message == null)
        {
            break;
        }
        Console.WriteLine("Received message with ID:   " + message.NMSMessageId);
        DoSomething();
    }
}

但是当我运行这段代码时，有时它没有创建连接，有时它没有创建会话等。我真的不明白。当我尝试不使用 for 循环（但仍然是 task()）时，它的行为相同。当我尝试没有任务时，只调用runner.ReceiveMessageFromServer() 它工作正常。谁能告诉我我做错了什么？

Answer 1

您的问题是 prefetchPolicy。

persistent queues (default value: 1000)
non-persistent queues (default value: 1000)
persistent topics (default value: 100)
non-persistent topics (default value: Short.MAX_VALUE - 1)

所有消息都被分派给第一个连接的消费者，当另一个消费者连接时，他不会收到消息，因此如果您有队列的并发消费者，您需要将 prefetchPolicy 设置为低于默认值的值。例如，将此jms.prefetchPolicy.queuePrefetch=1 添加到 activemq.xml 中的 uri 配置中，或者像这样在客户端 url 上设置它

我不知道 C# 是否可行，但试试这个

private const string activeMqUri = "activemq:tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1";

如果它不起作用，您可以通过将其添加到 activemq.xml 文件来在代理端设置它：

<destinationPolicy>
    <policyMap>
      <policyEntries>
        <!--  http://activemq.apache.org/per-destination-policies.html -->
        <policyEntry queue=">" queuePrefetch="1" > 
        </policyEntry>
      </policyEntries>
    </policyMap>
</destinationPolicy>

建议使用较大的预取值以获得高性能和高 消息量。但是，对于较低的消息量，每个 消息需要很长时间来处理，预取应该设置为 1。 这确保了消费者一次只处理一条消息。 但是，将预取限制指定为零会导致消费者 一次轮询消息，而不是消息 推送给消费者。

看看http://activemq.apache.org/what-is-the-prefetch-limit-for.html

还有

http://activemq.apache.org/destination-options.html

Answer 2

旁注：您不必在Main 方法中等待您的任务，您可以使用Task.WaitAll 方法来完成，这将阻塞主线程，直到所有消息都被传递。还有avoid the usage the StartNew method，一般情况下还是用Task.Run比较好。

现在，回到消息。根据current docs：

消费者的并发注意事项

在当前实现中，每个IConnection 实例都由一个单个后台线程支持，该线程从套接字读取并将生成的事件分派给应用程序。

从3.5.0 版本开始，应用程序回调处理程序可以调用阻塞 操作（例如IModel.QueueDeclare 或IModel.BasicCancel）。 IBasicConsumer 回调被并发调用。

因此，根据this question and answers there，您需要为您的消费者使用其他类：EventingBasicConsumer，如下所示：

var channel = _connection.CreateModel();
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (ch, ea) =>
    {
        var body = ea.Body;
        // ... process the message
        channel.BasicAck(ea.DeliveryTag, false);
    };
String consumerTag = channel.BasicConsume(queueName, false, consumer);

在这种情况下，您将以基于事件的方式收到消息，而不会阻塞通道。

可以在here 或here 找到其他示例。可以在Andras Nemes' blog 中找到有关 .Net 消息传递的重要参考。