我有一个RabbitMq消费者(RabbitMQ.Client.Events.EventingBasicConsumer
),它处理传入的消息。
但我注意到,如果关闭连接和模型,则它们不会等待完成库处理线程。例如:
- 如果我将线程睡眠几秒钟添加到
EventingBasicConsumer :: Received
回调中,那么我注意到Close
函数(IModel
和IConnection
的Close()
)在退出此消费者回调之前完成 - 完成
Close
函数后,我继续接收一些消息到EventingBasicConsumer :: Received
回调中。
那么如何正确关闭消费者并等待消费者线程中的所有处理完成呢? 我想确保在关闭所有连接/消费者后,我不会从库中收到任何传入的消息。
简化代码:
RunTest()
{
MyConsumer consumer = new MyConsumer();
consumer.Connect();
// Wait before close for process some count of incoming messages
Thread.Sleep(10 * 1000);
consumer.Disconnect();
}
class MyConsumer
{
private RabbitMQ.Client.IConnection m_Connection = null;
private RabbitMQ.Client.IModel m_Channel = null;
public void Connect()
{
//
// ...
//
m_Channel = m_Connection.CreateModel();
m_Consumer = new RabbitMQ.Client.Events.EventingBasicConsumer(m_Channel);
m_Consumer.Received += OnRequestReceived;
m_ConsumerTag = m_Channel.BasicConsume(m_Config.RequestQueue, false, m_Consumer);
}
public void Disconnect()
{
Console.WriteLine("---> IModel::Close()");
m_Channel.Close();
Console.WriteLine("<--- IModel::Close()");
Console.WriteLine("---> RabbitMQ.Client.IConnection::Close()");
m_Connection.Close();
Console.WriteLine("<--- RabbitMQ.Client.IConnection::Close()");
//
// Maybe there is need to do some RabbitMQ API call of channel/model
// for wait to finish of all consumer callbacks?
//
m_Channel = null;
m_Connection = null;
}
private void OnRequestReceived(object sender, RabbitMQ.Client.Events.BasicDeliverEventArgs mqMessage)
{
Console.WriteLine("---> MyConsumer::OnReceived");
Console.WriteLine("MyConsumer: ThreadSleep started");
Thread.Sleep(10000);
Console.WriteLine("MyConsumer: ThreadSleep finished");
if (m_Channel != null)
{
m_Channel.BasicAck(mqMessage.DeliveryTag, false);
}
else
{
Console.WriteLine("MyConsumer: already closed");
}
Console.WriteLine("<--- MyConsumer::OnReceived");
}
}
结果:
---> MyConsumer::OnReceived
MyConsumer: ThreadSleep started
---> IModel::Close()
<--- IModel::Close()
---> RabbitMQ.Client.IConnection::Close()
<--- RabbitMQ.Client.IConnection::Close()
MyConsumer: ThreadSleep finished
MyConsumer: already closed
<--- MyConsumer::OnReceived
---> MyConsumer::OnReceived
MyConsumer: ThreadSleep started
MyConsumer: ThreadSleep finished
MyConsumer: already closed
<--- MyConsumer::OnReceived
我们看到,在消费者和连接的Close()
函数退出后,MyConsumer::OnReceived
已经完成。此外,我们还看到在上一个OnReceived调用完成并关闭连接之后,还有一条消息进来了(这意味着RqbbitMq会继续处理消费者消息,直到内部库队列为空,而忽略消费者和连接已经关闭的事实)。