SpringBoot 整合RabbitMQ

常用的交换机

DirectExchange

直连型交换机，根据消息携带的路由键，将消息转发给对应的队列

FanoutExchange

扇形交换机，接收到消息后会将消息转发到所有队列

TopicExchange

主题交换机，根据消息携带的路由键和交换机与队列绑定键的规则，将消息转发给对应的队列

规则：

*（星号）：表示一个字符必须出现

#（井号）：表示任意数量的字符

准备

两个Spring Boot 项目：

• rabbitmq-provider（生产者）
• rabbitmq-comsumer（消费者）

版本号：2.1.7.RELEASE

依赖：

<!--rabbitmq-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml：

server:
  port: 9000
spring:
  application:
    name: rabbitmq-provider
  rabbitmq:
    host: 192.168.1.45
    port: 5672
    username: admin
    password: admin
    #virtual-host: 
    # ------- 消息确认配置项 --------
    # 确认消息已发送到交换机
    #publisher-confirms: true
    # 确认消息已发送队列
    #publisher-returns: true

DirectExchange

rabbitmq-provider

1、创建DirectRabbitConfig.java

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectRabbitConfig {

    /**
     * 交换机
     */
    @Bean
    public DirectExchange myDirectExchange() {
        // 参数意义:
        // name: 名称
        // durable: true
        // autoDelete: 自动删除
        return new DirectExchange("myDirectExchange", true, false);
    }

    /**
     * 队列
     */
    @Bean
    public Queue myDirectQueue() {

        return new Queue("myDirectQueue", true);
    }

    /**
     * 绑定
     */
    @Bean
    public Binding bindingDirect() {
        return BindingBuilder.bind(myDirectQueue())
                             .to(myDirectExchange())
                             .with("my.direct.routing");
    }
}

2、控制器提供一个发送消息的方法

@Autowired
private RabbitTemplate rabbitTemplate;

@RequestMapping("/send")
public String send() {

    String msg = "hello";
    rabbitTemplate.convertAndSend("myDirectExchange", "my.direct.routing", msg);

    return "success";
}

3、启动项目，使用PostMan调用发送消息接口，观察RabbitMQ控制面板的队列和控制台输出

rabbitmq-consumer

1、创建消息处理者DirectReceiver.java

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
//@RabbitListener(queues = "myDirectQueue")
public class DirectReceiver{

    @RabbitHandler
    @RabbitListener(queues = "myDirectQueue")
    public void process(String msg) {
        System.out.println(msg);
    }
}

2、启动项目，观察控制台输出

监听的队列，在项目启动之前应该存在，否则会报错

解决方法：先启动生产者发送一次消息，这时RabbitMQ会自动创建队列，再启动消费者接收消息

FanoutExchange

创建多个队列绑定到扇形交换机，生产者发送一次消息，可以观察到多个处理者都收到了消息。

rabbitmq-provider

1、创建FanoutRabbitConfig.java

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutRabbitConfig {

    // ----- 交换机 -----
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange", true, false);
    }

    // ----- 队列 -----
    @Bean
    public Queue fanoutQueueA() {
        return new Queue("fanoutQueueA", true);
    }

    @Bean
    public Queue fanoutQueueB() {
        return new Queue("fanoutQueueB", true);
    }

    @Bean
    public Queue fanoutQueueC() {
        return new Queue("fanoutQueueC", true);
    }

    // ----- 绑定 -----
    @Bean
    public Binding bindingFanoutA() {
        return BindingBuilder.bind(fanoutQueueA()).to(fanoutExchange());
    }

    @Bean
    public Binding bindingFanoutB() {
        return BindingBuilder.bind(fanoutQueueB()).to(fanoutExchange());
    }

    @Bean
    public Binding bindingFanoutC() {
        return BindingBuilder.bind(fanoutQueueC()).to(fanoutExchange());
    }
}

2、控制器提供一个发送消息的方法

@Autowired
private RabbitTemplate rabbitTemplate;

@RequestMapping("/sendByFanout")
public String sendByFanout() {

    String msg = "hello fanout";
    rabbitTemplate.convertAndSend("fanoutExchange", null, msg);

    return "success";
}

3、启动项目，发送消息

rabbitmq-comsumer

1、创建消息处理者

package com.rabbitmq.demo.mq;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class DirectReceiver{

    @RabbitHandler
    @RabbitListener(queues = "fanoutQueueA")
    public void processA(String msg) {
        System.out.println("fanoutQueueA " + msg);
    }

    @RabbitHandler
    @RabbitListener(queues = "fanoutQueueB")
    public void processB(String msg) {
        System.out.println("fanoutQueueB " + msg);
    }

    @RabbitHandler
    @RabbitListener(queues = "fanoutQueueC")
    public void processC(String msg) {
        System.out.println("fanoutQueueC " + msg);
    }
}

2、启动项目，观察控制台输出

TopicExchange

创建两个队列，并使用通配符绑定到主题交换机

rabbitmq-provider

1、创建TopicRabbitConfig.java

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TopicRabbitConfig {

    // 交换机
    @Bean
    public TopicExchange myTopicExchange() {
        return new TopicExchange("myTopicExchange", true, false);
    }

    // ----- 队列 -----
    @Bean
    public Queue myTopicQueue_01() {
        return new Queue("myTopicQueue_01", true);
    }

    @Bean Queue myTopicQueue_02() {
        return new Queue("myTopicQueue_02", true);
    }

    /**
     * 绑定路由键为topic.01
     */
    @Bean
    public Binding binding_01() {
        return BindingBuilder.bind(myTopicQueue_01()).to(myTopicExchange()).with("topic.01");
    }

    /**
     * 绑定路由键为topic.#规则
     */
    @Bean
    public Binding binding_02() {
        return BindingBuilder.bind(myTopicQueue_02()).to(myTopicExchange()).with("topic.#");
    }
}

2、控制器提供一个发送消息的方法

@Autowired
private RabbitTemplate rabbitTemplate;

@RequestMapping("/sendByTopic")
public String sendByTopic() {

    String msg = "hello topic";
    rabbitTemplate.convertAndSend("myTopicExchange", "topic.01", msg + " topic.01");
    rabbitTemplate.convertAndSend("myTopicExchange", "topic.xxx", msg + " topic.xxx");

    return "success";
}

3、启动项目，发送消息

rabbitmq-consumer

1、创建消息处理者

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class DirectReceiver{
    @RabbitHandler
    @RabbitListener(queues = "myTopicQueue_01")
    public void process_01(String msg) {
        System.out.println("myTopicQueue_01 " + msg);
    }

    @RabbitHandler
    @RabbitListener(queues = "myTopicQueue_02")
    public void process_02(String msg) {
        System.out.println("myTopicQueue_02 " + msg);
    }
}

2、启动项目，观察控制台输出

消息确认

1、application.yml文件开启rabbitmq消息确认

#确认消息已发送到交换机
publisher-confirms: true
#确认消息已发送到队列
publisher-returns: true

高版本报错可能得使用： publisher-confirm-type: correlated

2、生产者配置回调函数

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);

        // 开启Mandatory, 才能触发回调函数，无论消息推送结果如何都强制调用回调函数
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback: "+"相关数据：" + correlationData);
                System.out.println("ConfirmCallback: "+"确认情况：" + ack);
                System.out.println("ConfirmCallback: "+"原因：" + cause);
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback: "+"消息：" + message);
                System.out.println("ReturnCallback: "+"回应码：" + replyCode);
                System.out.println("ReturnCallback: "+"回应信息：" + replyText);
                System.out.println("ReturnCallback: "+"交换机：" + exchange);
                System.out.println("ReturnCallback: "+"路由键：" + routingKey);
            }
        });

        return rabbitTemplate;
    }
}

3、消费者配置监听

import com.rabbitmq.demo.mq.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MessageListenerConfig {

    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;

    @Autowired
    private MyAckReceiver myAckReceiver;

    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new  SimpleMessageListenerContainer(cachingConnectionFactory);

        // 监听队列名
        container.setQueueNames("myDirectQueue");
        // 当前消费者数量
        container.setConcurrentConsumers(1);
        // 最大消费者数量
        container.setMaxConcurrentConsumers(1);
        // 手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置监听器
        container.setMessageListener(myAckReceiver);

        return container;
    }
}

4、消费者创建监听器

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        // 消息的唯一性ID
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        try {
            String msg = message.toString();
            System.out.println("消息: " + msg);
            System.out.println("消息来自: "+message.getMessageProperties().getConsumerQueue());

            // 手动确认
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            // 拒绝策略
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
}

5、启动生产者发送消息

6、启动消费者

7、观察生产者和消费者控制台输出