RabbitMQ应用Demo:使用exchange广播消息

Posted 2023-02-26 Neo Yang

环境准备

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

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

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

安装后的界面截图：

消息生产者Demo代码

    private final static String EXCHANGE_NAME = "elon_exchange";

    /**
     * 生产消息发送到交换器.
     *
     * @param messageBody 消息体
     * @author elon
     */
    public void produceMessage2Exchange(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.exchangeDeclare(EXCHANGE_NAME, "fanout");
            channel.basicPublish(EXCHANGE_NAME, "", null, messageBody.getBytes(Charsets.UTF_8));
            LOGGER.info("Sent ", messageBody);
         catch (Exception e) 
            LOGGER.info("Produce message fail.", e);
        
    

消息消费者Demo代码

    private final static String EXCHANGE_NAME = "elon_exchange";

    /**
     * 消费交换器转发的消息
     */
    public void consumeMessageFromExchange()
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.*.*.*");
        factory.setPort(5672);
        factory.setUsername("***");
        factory.setPassword("*******");

        try 
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            
            String queueName = channel.queueDeclare().getQueue();
            LOGGER.info("Get queue name:", queueName);
            channel.queueBind(queueName, EXCHANGE_NAME, "");

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

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

核心逻辑是获取一个分配的队列名，绑定到交换器，再通过队列接收交换器转发的消息。多个队列可绑定到同一个交换器，从而实现广播消息。