ActiveMQ - несколько подключений за сеанс?

Question

В ActiveMQ мне сказали, что наиболее оптимальным решением для увеличения пропускной способности является наличие нескольких соединений, каждое со своим собственным сеансом и потребителем.

Я пытался добиться этого с помощью NMS (Соединениечерез C #), но на экране «Active Consumwers» веб-консоли MQ я вижу все свои подключения и потребителей в списке так, как я ожидал бы их видеть, но в следующем столбце все они имеют sessionId = «1».Я ожидал, что для каждого будет отдельный идентификатор сессии.

Это правильно?И если для каждого соединения / потребителя должны быть разные идентификаторы сеансов, как бы я обеспечил создание этих дополнительных сеансов?

Вот пример кода, который я использую для создания нового соединения (это основано на Замечания Код введения в ActiveMQ для транзакционных сообщений ):

public QueueConnection(IConnectionFactory connectionFactory, string queueName, AcknowledgementMode acknowledgementMode)
{
    this.connection = connectionFactory.CreateConnection();
    this.connection.Start();

    this.session = this.connection.CreateSession(acknowledgementMode);

    this.queue = new ActiveMQQueue(queueName);
}

... и это делается каждый раз для каждого соединения, которое я открываю.

Answer 1

Выглядит этот фрагмент кода, который является корневым источником моей проблемы:

public SimpleQueueListener CreateSimpleQueueListener(IMessageProcessor processor)
{
    IMessageConsumer consumer = this.session.CreateConsumer(this.queue, "2 > 1");
    return new SimpleQueueListener(consumer, processor, this.session);
}

Поскольку он использует общий сеанс (this.session) для всех потребителей. Создавая новый сеанс каждый раз (и удерживая его в коллекции или другими способами), я достиг цели работы каждого слушателя в своем собственном сеансе в рамках одного и того же соединения. Например:

public SimpleQueueListener CreateSimpleQueueListener(IMessageProcessor processor)
{
    var listenerSession = this.connection.CreateSession();
    IMessageConsumer consumer = listenerSession.CreateConsumer(this.queue, "2 > 1");
    return new SimpleQueueListener(consumer, processor, listenerSession);
}