项目场景:
最近在项目中使用了RabbitMq,其中有一个功能必须能随时切断RabbitMq的coumser。第一时间写出来的代码如下:
伪代码:
while(flag){ QueueingConsumer.Delivery delivery=consumer.nextDelivery();
String message = new String(delivery.getBody());
//doing someting strange
//...... } //另外一个项目开始关闭 public void closeConsumer{
channel().close();
connection().close();
}
closeConsumer();
通过关闭channel,消费者自然会关闭。然而,项目开始报错:
channel关闭抛出ShutdownSignalException,抛出异常就表示这种关闭方式是不合理的。有必要去探索一下是否有更优雅的链接关闭方式。
先看consumer的源码:
public Delivery nextDelivery()throws InterruptedException, ShutdownSignalException, ConsumerCancelledException{
return handle(_queue.take());
}
而这里_queue其实是一个 LinkedBlockingQueue,LinkedBlockingQueue是一个单向链表实现的阻塞队列。nextDelivery()方法使用LinkedBlockingQueue的take方法实现了阻塞。这个地方感觉不好操作。但是QueueingConsumer还有另外一个读取数据的方法,源码如下:
public Delivery nextDelivery(long timeout)throws InterruptedException, ShutdownSignalException, ConsumerCancelledException{
return handle(_queue.poll(timeout, TimeUnit.MILLISECONDS));
}
这边设定了超时时间。虽然没想到优雅关闭消费者的方法,但是利用超时时间来修改一下读取数据的方法还是可以的。代码如下:
try{
while(flag){
QueueingConsumer.Delivery delivery = consumer.getDeliveryMessage(10000);
if(consumer.getState()==Consumer.Status.Stopped.getValue()){
break;
}
if(delivery==null){
continue;
}
String message = new String(delivery.getBody());
//....dosomething
}
}finally {
if(consumer!=null){
consumer.closeConnection();
}
} public void closeConnection{
channel().close();
connection().close();
} //另外一个线程关闭consumer
consumer.setStatus(Consumer.Status.Stopped);