RabbitMQ踩坑记
Posted little-sheep
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ踩坑记相关的知识,希望对你有一定的参考价值。
之前我们给我们的系统加了一个使用SpringAOP+RabbitMQ+WebSocket进行实时消息通知功能(https://www.cnblogs.com/little-sheep/p/9934887.html)。在测试环境下没有问题,但上到生产环境后部分用户反映出现了丢消息的情况,针对这个问题我们进行了排查,发现,原本我们的系统是单机的,但用户在之前做了调整,在内外网服务器分别部署了系统,两个服务器都公用一个RabbitMQ。那么问题就来了。
之前生产者向Exchange生产消息,消费者从queue消费消息使用的是direct模式,通过routing-key保证生产的消息只有指定的消费者可消费。但当两台服务器共用一个MQ时即有了两个消费者连接同一个queue,此时rabbitmq并不是一个消息两个消费者都能消费,而是采用默认的轮询发送方式,A服务器收到消息1、3、5 。。。而B服务器收到2、4、6 。。。
这就出现了用户感觉的丢消息现象。所以我们考虑将通知改为广播形式即fanout。
RabbitMQ将消息中间件的实现分成了Exchange+Queue的形式,Exchange和Queue使用Binding;生产者向Exchange生产消息,消息根据指定的binding进入Queue,消费者从Queue取消息。Fanout模式如图:
一个消费者会对应一个queue,那么多个消费者要有多个queue。
修改后代码如下:
RabbitMQConfig.java
//声明exchange Connection connection = factory.createConnection(); Channel channel = connection.createChannel(false); //生产环境中有多个server,每个server都是一个消费者,对同一个消息都要进行处理。选用广播模式 channel.exchangeDeclare("exchange.websocket.msg", BuiltinExchangeType.FANOUT);
RabbitMessageQueue.java
@Override public void send(WebSocketMsgEntity entity) { logger.warn("::product msg to MQ-websocket_msg_queue!"); //若没有指定exchange,则使用默认名为“”的exchange,binding名与queue名相同 rabbitTemplate.convertAndSend("exchange.websocket.msg","", entity); }
RabbitMQListener.java
@Component public class RabbitMQListener { private static Logger logger = LoggerFactory.getLogger(RabbitMQListener.class); @Autowired private RabbitMQService mqService; /** * WebSocket推送监听器 * @param socketEntity * @param deliveryTag * @param channel */ @RabbitListener(bindings ={@QueueBinding(value = @Queue(exclusive = "true"), exchange = @Exchange(value = "exchange.websocket.msg", type = ExchangeTypes.FANOUT))}) public void webSocketMsgListener(@Payload WebSocketMsgEntity socketMsgEntity, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException { logger.warn("::consume msg from MQ-websocket_msg_queue!"); mqService.handleWebSocketMsg(socketMsgEntity, deliveryTag, channel); } }
原本的客户端监听是监听Queue,而现在改成监听Binding。并不显式的指定一个Queue,而是将Queue设置成exclusive = true,这样每个消费者在监听Binding时都会默认创建一个Queue与指定的Exchange绑定,在消费者断开连接后Queue自动删除。若有两个消费者,则创建两个Queue,他们绑定的Exchange相同,当生产者有消息时会向两个Queue各插入一条,那么两个系统的用户就都能收到通知啦!
以上是关于RabbitMQ踩坑记的主要内容,如果未能解决你的问题,请参考以下文章