rabbitmq生产者—消息发布时的权衡

水深无声 2022-10-09 04:50 120阅读 0赞

不做任何配置的情况下，生产者是不知道消息是否真正到达RabbitMQ，也就是说消息发布操作不返回任何消息给生产者。怎么保证我们消息发布的可靠性？有以下几种常用机制。

![在这里插入图片描述][20210623213003766.png]

# 准备 #

## 生产者 ##

package com.morris.rabbit.workqueue;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.MessageProperties;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    import java.util.stream.IntStream;
    
    /** * producer */
    public class NewTask { 
    
        private final static String QUEUE_NAME = "workQueue";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            try (Connection connection = factory.newConnection();
                 Channel channel = connection.createChannel()) { 
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                String message = "task";
                IntStream.rangeClosed(1, 9).forEach(i -> { 
                    try { 
                        channel.basicPublish("", QUEUE_NAME, null, (message + i).getBytes());
                        System.out.println(" [x] Sent '" + message +  i +"'");
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                });
            }
        }
    }

## 消费者 ##

package com.morris.rabbit.workqueue;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.DeliverCallback;
    
    import java.util.concurrent.TimeUnit;
    
    /** * consumer */
    public class Worker { 
    
        private final static String QUEUE_NAME = "workQueue";
    
        public static void main(String[] argv) throws Exception { 
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.80.205");
            factory.setPort(5672);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            DeliverCallback deliverCallback = (consumerTag, delivery) -> { 
                String message = new String(delivery.getBody(), "UTF-8");
    
                System.out.println(" [x] Received '" + message + "'");
                try { 
                    int i = (int) (Math.random() * 30);
                    TimeUnit.SECONDS.sleep(i);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    System.out.println(" [x] Done");
                }
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { 
    });
        }
    }

# 发送消息的确认机制 #

消息的确认机制有以下几种：

*  自动确认：也就是发完消息就不管了，不进行确认。
 *  同步手动确认
 *  批量手动确认
 *  异步确认

## 同步手动确认 ##

// 启用发送者确认模式
    channel.confirmSelect();
    String message = "task";
    IntStream.rangeClosed(1, 9).forEach(i -> { 
        try { 
            channel.basicPublish("", QUEUE_NAME, null, (message + i).getBytes());
            System.out.println(" [x] Sent '" + message + i + "'");
    
            //确认是否成功(true成功)
            if (channel.waitForConfirms()) { 
                System.out.println("send success");
            } else { 
                System.out.println("send failure");
            }
        } catch (IOException | InterruptedException e) { 
            e.printStackTrace();
        }
    });

## 批量手动确认 ##

// 启用发送者确认模式
    channel.confirmSelect();
    String message = "task";
    IntStream.rangeClosed(1, 9).forEach(i -> { 
        try { 
            channel.basicPublish("", QUEUE_NAME, null, (message + i).getBytes());
            System.out.println(" [x] Sent '" + message + i + "'");
        } catch (IOException e) { 
            e.printStackTrace();
        }
    });
    // 启用发送者确认模式（批量确认）
    channel.waitForConfirmsOrDie();

## 异步确认 ##

// 启用发送者确认模式
    channel.confirmSelect();
    
    // 确认回调
    channel.addConfirmListener(new ConfirmListener() { 
        // 成功
        public void handleAck(long deliveryTag, boolean multiple)
                throws IOException { 
            System.out.println("send_ACK:" + deliveryTag + ", multiple:" + multiple);
        }
    
        // 失败
        public void handleNack(long deliveryTag, boolean multiple)
                throws IOException { 
            System.out.println("Error----send_NACK:" + deliveryTag + ", multiple:" + multiple);
        }
    });
    
    IntStream.rangeClosed(1, 2).forEach(i -> { 
        try { 
            channel.basicPublish("", QUEUE_NAME, null, (message + i).getBytes());
            System.out.println(" [x] Sent '" + message + i + "'");
        } catch (IOException e) { 
            e.printStackTrace();
        }
    });

注意：

*  不能使用try-with-resource，否则消息发完了，channel就关闭了，消息确认的回调有可能不会执行。
 *  多条消息有时回调只会执行一次，multiple=true

# 消息的持久化机制 #

消息要真正做到持久化，需满足以下条件：

*  exchange持久化（若使用了）
 *  queue持久化
 *  message持久化

// queue持久化 durable=true
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    String message = "task";
    // message持久化 MessageProperties.PERSISTENT_TEXT_PLAIN
    channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");

# 消息的失败通知 #

在发送消息时设置mandatory标志，告诉RabbitMQ，如果消息不可路由，应该将消息返回给发送者，并通知失败。可以这样认为，开启mandatory  
是开启故障检测模式。

注意：它只会让RabbitMQ向你通知失败，而不会通知成功。如果消息正确路由到队列，则发布者不会受到任何通知。带来的问题是无法确保发布消  
息一定是成功的，因为通知失败的消息可能会丢失。

channel.addReturnListener((replycode, replyText, exchange, routeKey, basicProperties, bytes) -> { 
        System.out.println("返回的replycode:" + replycode);
        System.out.println("返回的replyText:" + replyText);
    });
    channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    String message = "task";
    // mandatory=true
    channel.basicPublish(EXCHANGE_NAME, QUEUE_NAME, true, null, message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");

[20210623213003766.png]: /images/20221005/78d13dc599b54763811ddd62fa592543.png