RabbitMQ应用Demo:使用queue点对点通知消息

安装RabbitMq: 个人是先在Win10便携机上安装VMWare Workstation, 再安装Cent OS 操作系统，在此基础上安装RabbitMQ。

安装过程可以参考这篇博客： https://blog.csdn.net/hsxy123123/article/details/104006744

需要注意RabbitMQ官网提供的erlang与RabbitMQ的配套版本，按版本安装。

    private final static String QUEUE_NAME = "elon_queue";
 
    
    public void produceMessage(String messageBody) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.*.*.*");
        factory.setPort(5672);
        factory.setUsername("***");
        factory.setPassword("*******");

        try {
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.basicPublish("", QUEUE_NAME, null, messageBody.getBytes());
            LOGGER.info("Sent {}", messageBody);
        } catch (Exception e) {
            LOGGER.info("Produce message fail.", e);
        }
    }ren

发送消息使用默认交换器 “” ，指定队列名称elon_queue。

    private final static String QUEUE_NAME = "elon_queue";

    
    public void consumeMessage() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.*.*.*");
        factory.setPort(5672);
        factory.setUsername("***");
        factory.setPassword("*******");

        try {
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                LOGGER.info("Receive message from queue:{}", message);
            };

            channel.basicConsume(QUEUE_NAME, true, deliverCallback, tag->{});
        } catch (Exception e) {
            LOGGER.error("Consume message exception.", e);
        }
    }

消费者增加 channel.queueDeclare(QUEUE_NAME, false, false, false, null) 这行代码是考虑消费者进程可能先于生产者進程启动的情况。队列不存在的情况下调用basicConsume报错。

可以启动多个消费者进程接收队列elon_queue的消息，但只要有一个消费者收到消息应答后其它消费者将不会再收到该消息。这样做的目的是提升系统的可靠性，单个消费者进程挂了，其它消费者能正常处理业务，消息也不会重复消费。