RabbitMQ消息Unacked的可能原因

作者：AllinToyou | 2024-05-01 22:51:32

踩

项目场景：

最近在写Android应用时，使用到了RabbitMQ实现评论消息与聊天消息的实时收发，这两种消息分别通过两个阻塞线程来接收，两个线程监听同一个队列(每个用户拥有一个队列)，根据发送消息的exchange进行区分。在接收消息并保存至本地数据库后会手动应答，阻塞线程销毁时会关闭Connection连接。代码如下：

@Override
       private void consumeMessage(final String queueName){
        Thread thread = new Thread() {
            @Override
            public void run() {
                super.run();
                try {
                    connection = factory.newConnection();
                    channel=connection.createChannel();
                } catch (Exception e){
                    e.printStackTrace();
                }
                while (true) {  //阻塞线程，持续监听RabbitMQ消息，注意：Service销毁并不会销毁该线程
                    try {
                        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
                            @Override
                            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                                Log.e("TAG", "handleDelivery()方法执行");
                                String messageJson = new String(body);
                                String exchange = envelope.getExchange();
                                HashMap<String, Object> headers = (HashMap<String, Object>) properties.getHeaders();
                                Object obj = headers.get("messageType");
                                String messageType = String.valueOf(obj);
                                Message msg = handler.obtainMessage();
                                if (exchange.equals("myforum_interaction_exchange")){
                                    if ("praise".equals(messageType)){  //点赞消息
                                        Log.e("TAG", "监听到点赞消息，Json格式为"+messageJson);
                                        msg.what=1;
                                        msg.obj=messageJson;
                                        handler.sendMessage(msg);
                                    }else if ("comment".equals(messageType)){   //评论消息
                                        Log.e("TAG", "监听到评论消息，Json格式为："+messageJson);
                                        msg.what=2;
                                        msg.obj=messageJson;
                                        handler.sendMessage(msg);
                                    }
                                    //消息应答应该在if判断里面，否则可能会应答聊天消息，造成消息丢失
                                    channel.basicAck(envelope.getDeliveryTag(), false);
                                }
                            }
                        });
                        sleep(500);
                        if (isServiceDestroy) { //监听Service销毁标志位，当Service销毁时跳出阻塞循环，线程随之结束
                            if (isThreadCompleteTheLastWork){
                                try {
                                    if (channel!=null){
                                        channel.close();
                                    }
                                    if (connection!=null){
                                        connection.close();
                                    }
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                                break;
                            }else {
                                isThreadCompleteTheLastWork=true;
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        thread.start();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

private void consumeMessage(final String queueName){
        Thread thread = new Thread() {
            @Override
            public void run() {
                super.run();
                try {
                    connection = factory.newConnection();
                    channel=connection.createChannel();
                } catch (Exception e){
                    e.printStackTrace();
                }
                while (true) {  //阻塞线程，持续监听RabbitMQ消息，注意：Service销毁并不会销毁该线程
                    if (isServiceDestroy) { //监听Service销毁标志位，当Service销毁时跳出阻塞循环，线程随之结束
                        try {
                            if (channel!=null){
                                channel.close();
                            }
                            if (connection!=null){
                                connection.close();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        break;
                    }
                    try {
                        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
                            @Override
                            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                                String exchange = envelope.getExchange();
                                if ("myforum_communication_exchange".equals(exchange)){
                                    String chatMessageJson = new String(body);
                                    Message message = handler.obtainMessage();
                                    message.obj=chatMessageJson;
                                    message.what=1;
                                    handler.sendMessage(message);
                                    channel.basicAck(envelope.getDeliveryTag(), false);
                                }
                            }
                        });
                        sleep(500);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        thread.start();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

问题描述：

但是，RabbitMQ Manager界面时不时就会显示评论消息Unacked，Android端表现为收不到消息，没有UI显示，且重启后便能接收到消息 (RabbitMQ未应答的消息在断开连接后会重新入队)

原因：

由于两个阻塞线程监听同一队列，它们会竞争处理消息，当接收聊天消息的线程获取到评论消息时，虽然会进入handleDelivery()方法进行消费，但是方法中检查到该消息并不是来自聊天exchange，不会应答。也就是说消息虽然被送达到客户端，但是被不对应的消费者线程处理而被丢弃，最终造成未应答。

解决：

将两个阻塞线程合二为一，即一个消费者线程监听一个队列，在线程内部通过逻辑对消息进行区分

private void consumeMessage(final String queueName){
        Thread thread = new Thread() {
            @Override
            public void run() {
                super.run();
                try {
                    connection = factory.newConnection();
                    channel=connection.createChannel();
                } catch (Exception e){
                    e.printStackTrace();
                }
                while (true) {  //阻塞线程，持续监听RabbitMQ消息，注意：Service销毁并不会销毁该线程
                    try {
                        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
                            @Override
                            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                                Message msg = handler.obtainMessage();
                                String messageJson = new String(body);
                                String exchange = envelope.getExchange();
                                if ("myforum_communication_exchange".equals(exchange)){ //聊天消息
                                    String chatMessageJson = new String(body);
                                    Message message = handler.obtainMessage();
                                    message.obj=chatMessageJson;
                                    message.what=3;
                                    handler.sendMessage(message);
                                    channel.basicAck(envelope.getDeliveryTag(), false);
                                }else { //互动消息
                                    HashMap<String, Object> headers = (HashMap<String, Object>) properties.getHeaders();
                                    Object obj = headers.get("messageType");
                                    String messageType = String.valueOf(obj);
                                    if (exchange.equals("myforum_interaction_exchange")){
                                        if ("praise".equals(messageType)){  //点赞消息
                                            msg.what=1;
                                            msg.obj=messageJson;
                                            handler.sendMessage(msg);
                                        }else if ("comment".equals(messageType)){   //评论消息
                                            msg.what=2;
                                            msg.obj=messageJson;
                                            handler.sendMessage(msg);
                                        }
                                        //消息应答应该在if判断里面，否则可能会应答聊天消息，造成消息丢失
                                        channel.basicAck(envelope.getDeliveryTag(), false);
                                    }
                                }
                            }
                        });
                        sleep(500);
                        if (isServiceDestroy) { //监听Service销毁标志位，当Service销毁时跳出阻塞循环，线程随之结束
                            try {
                                if (channel!=null){
                                    channel.close();
                                }
                                if (connection!=null){
                                    connection.close();
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                            break;
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        thread.start();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

总结：

两个消费者监听同一队列时会存在竞争，即使调用了消费消息的回调方法，也并不代表者消费成功，RabbitMQ只会在消息应答后才会确认消息消费成功。
不要图省事，直接使用消息自动应答，自动应答会在消费者调用回调方法时进行应答。假如在本案例使用了自动应答，那么在聊天消费者线程接收到了消息的那一刻，RabbitMQ就会接收到消息应答从而删除消息，但是该消息在客户端已被丢弃，从而造成消息丢失。而使用手动应答，即使聊天消费者线程丢弃了消息，RabbitMQ由于未接收到消息应答，会将消息重新入队直到接收到消息应答为止 (即聊天消费者线程接收到聊天消息并应答)。换句话说，本案例未修改前虽然有UI上的不友好(多次重新启动才可能接收到消息)，但是保证了消息不丢失

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/AllinToyou/article/detail/520921