RabbitMQ Несколько уникальных потребителей в службе Windows - PullRequest
Новая
       13

RabbitMQ Несколько уникальных потребителей в службе Windows

0 голосов
/ 06 мая 2019

Я искал StackOverflow и не думаю, что другие проблемы, связанные с несколькими потребителями, совпадают с тем, что я пытаюсь.

У меня есть служба Windows, которая запускается и регистрируется для отслеживания двух разных RabbitMQ очереди.Если я регистрирую его только для прослушивания 1 очереди, он работает нормально.Когда я пытаюсь заставить его слушать обе очереди, работает только второй слушатель. 

var responseConsumer = new EventingBasicConsumer(_responseChannel);

responseConsumer.Received += (model, ea) =>
{
    var message = Encoding.UTF8.GetString(ea.Body);
    // PROCESS MESSAGE
    IssuerProcessor processor = new IssuerProcessor("Processed Message(" + pause + "ms) : ");

    Thread.Sleep(pause);

    var ret = processor.ProcessResponseMessage();
    Console.WriteLine("Processed message: " + ret);
    ulong deliveryTag = ea.DeliveryTag;

    _responseChannel.BasicAck(deliveryTag, false);
    Console.WriteLine("ACK RESPONSE Delivery Tag (" + deliveryTag.ToString() + "): " + ret);
}; // end requestConsumer.Received

_responseChannel.BasicConsume(queue: RESPONSE_QUEUE, autoAck: autoAck, consumer: responseConsumer);

Другой потребитель выглядит так же, как и этот, но он слушает другую очередь, используя то же соединение, сдругой канал.

Вот инициализация для обоих: 

_requestChannel = _connection.CreateModel();
_responseChannel = _connection.CreateModel();

_requestChannel.ExchangeDeclare(EXCHANGE, ExchangeType.Direct, true, false, null);
_requestChannel.QueueDeclare(queue: REQUEST_QUEUE, durable: true, exclusive: false, autoDelete: false, arguments: null);
_requestChannel.BasicQos(0, 1, false);
_requestChannel.QueueBind(REQUEST_QUEUE, EXCHANGE, REQUEST_QUEUE, null);

_responseChannel.ExchangeDeclare(EXCHANGE, ExchangeType.Direct, true, false, null);
_responseChannel.QueueDeclare(RESPONSE_QUEUE, durable: true, exclusive: false, autoDelete: false, arguments: null);
_responseChannel.BasicQos(0, 1, false);
_responseChannel.QueueBind(RESPONSE_QUEUE, EXCHANGE, RESPONSE_QUEUE, null);

1 Ответ

0 голосов
/ 07 мая 2019

Проблема закончилась тем, что служба работала в одном потоке и не могла контролировать обе очереди одновременно.

Я только что создал два разных потока и запустил каждого потребителя в отдельном потоке. 

public void StartWorkerThreads()
{
    Thread receivingConsumerThread = new Thread(StartListenReceivingQueue);
    receivingConsumerThread.Start();

    Thread respondingConsumerThread = new Thread(StartListenRespondingQueue);
    respondingConsumerThread.Start();
}
...