【rabbitmq】Queueingconsumer被废止后老代码如…
2020-02-05 16:03:50来源:博客园 阅读 ()
【rabbitmq】Queueingconsumer被废止后老代码如何做的解决方案
amqp-client 3.x之前的rabbitmq版本有个消费者的写法是借助于Queueingconsumer的:
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicQos(1);
channel.basicConsume(QUEUE_NAME, false, "consumer_test",consumer);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [X] Received '" + message + "'");
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
break;
}
这个应该是5.x之前的经典写法。但是在4.x的版本QueueingConsumer被标记废止5.x被移除。移除的原因是什么呢?
原来QueueingConsumer内部用LinkedBlockingQueue来存放消息的内容,而LinkedBlockingQueue:一个由链表结构组成的有界队列,照先进先出的顺序进行排序 ,未指定长度的话,默认 此队列的长度为Integer.MAX_VALUE,那么问题来了,如果生产者的速度远远大于消费者的速度,也许没等到队列阻塞的条件产生(长度达到Integer.MAX_VALUE)内存就完蛋了,在老的版本你可以通过设置 rabbitmq的prefetch属性channel.basicQos(prefetch)来处理这个问题如果不设置可能出现内存问题(比如因为网络问题只能向rabbitmq生产不能消费,消费者恢复网络之后就会有大量的数据涌入,出现内存问题,oom fgc等)。
而且最上面的写法很不合理不符合事件驱动,什么时候停止while循环也不能写的很优雅,所以在更高的版本直接被移除。取而代之的是DefaultConsumer,你可以通过扩展DefaultConsumer来实现消费者:
消费的代码:(RabbitMqMessageConsumer是对DefaultConsumer的扩展)
RabbitMqMessageConsumer rpcMessageConsumer = new RabbitMqMessageConsumer(channel,cores);
channel.basicQos(cores);
channel.basicConsume(QUEUE_NAME, true, rpcMessageConsumer);
RabbitMqMessageConsumer代码:
public class RabbitMqMessageConsumer extends DefaultConsumer{
public RabbitMqMessageConsumer(Channel channel) {
super(channel);
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
//TODO someting
}
}
其中handleDelivery是处理消息的逻辑。
高版本的解决方案给出了,那么回到我们题目的问题,老代码是按照4.x之前的写的,由于某种原因升级到了5.x了如何做?釜底抽薪的办法就是按照上面的事件驱动的方式重写消费者。折中的办法呢(不想改变老代码的逻辑和结构)。
我就碰到了这样的问题,老代码写了很多的轮子,导致这块代码很难重写。那就只能按照原来QueueingConsumer的写法继续做。解决思路如下:
首先消费的过程还是按照最开始那样:
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicQos(1);
channel.basicConsume(QUEUE_NAME, false, "consumer_test",consumer);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [X] Received '" + message + "'");
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
break;
}
然后实现自己的QueueingConsumer(QueueingConsumer已经移除):
public class QueueingConsumer extends DefaultConsumer{
private LinkedBlockingQueue<Delivery> queue;
public QueueingConsumer(Channel channel) {
super(channel);
queue = new LinkedBlockingQueue<RabbitMqMessageConsumer.Delivery>();
}
public QueueingConsumer(Channel channel,int size) {
super(channel);
queue = new LinkedBlockingQueue<RabbitMqMessageConsumer.Delivery>(size);
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
Delivery delivery = new Delivery();
delivery.setBody(body);
delivery.setProperties(properties);
delivery.setEnvelope(envelope);
try {
queue.put(delivery);
} catch (InterruptedException e) {
LogUtils.error(e);
}
}
public Delivery nextDelivery()throws InterruptedException, ShutdownSignalException, ConsumerCancelledException{
return queue.take();
}
public Delivery nextDelivery(long timeout)throws InterruptedException, ShutdownSignalException, ConsumerCancelledException{
return queue.poll(timeout, TimeUnit.MILLISECONDS);
}
public class Delivery{
private BasicProperties properties;
private byte[] body;
private Envelope envelope;
public BasicProperties getProperties() {
return properties;
}
public void setProperties(BasicProperties properties) {
this.properties = properties;
}
public byte[] getBody() {
return body;
}
public void setBody(byte[] body) {
this.body = body;
}
public Envelope getEnvelope() {
return envelope;
}
public void setEnvelope(Envelope envelope) {
this.envelope = envelope;
}
}
}
这样你就能在不修改之前的老代码的情况下升级版本了,当然最好还是重写,这个只能起到个过度
原文链接:https://www.cnblogs.com/nfsnyy/p/12264590.html
如有疑问请与原作者联系
标签:
版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有
- SpringBoot2.3整合RabbitMQ实现延迟消费消息 2020-05-26
- Docker中使用RabbitMQ 2020-05-21
- 消息中间件—SpringBoot下RabbitMQ实战 2020-05-15
- 一次奇怪的RabbitMQ超时异常 2020-05-04
- SpringBoot+RabbitMQ学习笔记(六)RabbitMQ之ACK机制 2020-04-29
IDC资讯: 主机资讯 注册资讯 托管资讯 vps资讯 网站建设
网站运营: 建站经验 策划盈利 搜索优化 网站推广 免费资源
网络编程: Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它
服务器技术: Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护
软件技巧: 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷
网页制作: FrontPages Dreamweaver Javascript css photoshop fireworks Flash