SpringBoot整合RabbitMQ 手动应答简单demo / 开普饭

版本说明

JDK 1.8
RabbitMQ 3.7.15 Erlang 22.0
SpringBoot 2.3.3.RELEASE

// Update 2021年1月19日16:50:16 CentOS搭建RabbitMQ 完整可操作
https://www.cnblogs.com/fengyumeng/p/11133924.html

1. 在RabbitMQ的Web管理界面，创建test队列

参数的含义
durability：是否持久化（重启或宕机后消息依然保存）

durable 持久
transient 暂时

新建maven项目。

2. pom.xml

<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0"         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <parent>        <groupId>org.springframework.boot</groupId>        <artifactId>spring-boot-starter-parent</artifactId>        <version>2.3.3.RELEASE</version>        <relativePath/> <!-- lookup parent from repository -->    </parent>    <groupId>com.demo</groupId>    <artifactId>rabbitmq-demo</artifactId>    <version>1.0.0</version>    <properties>        <lombok.version>1.18.12</lombok.version>    </properties>    <dependencies>        <!--web 模块-->        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-web</artifactId>        </dependency><!-- AMQP -->        <dependency>            <groupId>org.springframework.boot</groupId>            <artifactId>spring-boot-starter-amqp</artifactId>        </dependency>        <!--  lombok-->        <dependency>            <groupId>org.projectlombok</groupId>            <artifactId>lombok</artifactId>            <optional>true</optional>            <scope>provided</scope>            <version>${lombok.version}</version>        </dependency>    </dependencies>    <build>        <plugins>            <plugin>                <groupId>org.springframework.boot</groupId>                <artifactId>spring-boot-maven-plugin</artifactId>            </plugin>        </plugins>    </build></project>

3. application.yaml

server:  port: 20002spring:  rabbitmq:  # 这里我改了本地的hosts，实际地址是192.168.0.121    host: vm.com    port: 5672    virtual-host: /    username: admin    password: admin    # 开启消息确认模式    # 消息发送到交换机确认机制，是否确认回调    # publisher-confirms: true    # 是否返回回调    publisher-returns: true    template:      #开启mandatory: true, basic.return方法将消息返还给生产者      mandatory: true    listener:      simple:      # 手动应答        acknowledge-mode: manual        # 最少消费者数量        concurrency: 1         # 最多消费者数量        max-concurrency: 10        # 支持重试        retry:          enabled: true

端口

5672：RabbitMQ的通信端口
15672：Web管理界面端口

4. RabbitmqDemo.java

@SpringBootApplication@EnableRabbitpublic class RabbitmqDemoApplication {        public static void main(String[] args) {        SpringApplication.run(RabbitmqDemoApplication.class, args);    }}

5. RabbitConfig.java

@Configuration@Slf4jpublic class RabbitConfig {    private RabbitTemplate rabbitTemplate;    @Bean    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {        rabbitTemplate = new RabbitTemplate(connectionFactory);        return rabbitTemplate;    }}

配置RabbitMQ的消息模板。

6. 消息生产者 produce.java

@Componentpublic class Producer {        // @Qualifier("rabbitTemplate")    @Autowired    private RabbitTemplate rabbitTemplate;    public void send() {        for (int i = 0; i < 5; i++) {            System.out.println("生产者发送消息，序号为: " + i);            rabbitTemplate.convertAndSend("test", String.valueOf(i));        }    }}

初始化消息发送模板RabbitTemplate，@Qualifier注解用于限定具体的实现类，这里可以不指定。

7. 消息消费者 consumer.java

消费者1和消费者2均监听test队列。

不同的是，消费者1收到消息后返回确认应答basicAck。

而消费者2收到消息后返回拒绝应答basicRegect，消息被消费者拒绝后重新回到test队列中，等待下次发送给消费者。

@Component@Slf4jpublic class Consumer {    /**     * 消费者1 模拟正常处理消息的情况，消息处理完毕发送确认应答     * @param message     * @param channel     * @throws IOException     */    @RabbitListener(queues = "test")    public void process1(Message message, Channel channel) throws IOException {        log.info("消费者1 接收消息: " + new String(message.getBody()));        log.info("消费者1 确认应答消息：" + new String(message.getBody()));        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);    }    /**     * 消费者2 模拟处理消息出错的情况，消费者2向rabbitmq发送拒绝应答。     * 处理失败的消息会被重新放入ready中，再次发送给消费者，直至收到确认应答     * @param message     * @param channel     * @throws IOException     */    @RabbitListener(queues = "test")    public void process2(Message message, Channel channel) throws IOException {        log.info("消费者2 接收消息：" + new String(message.getBody()));        log.info("消费者2 拒绝应答消息：" + new String(message.getBody()));        channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);    }}

8. 测试RabbitMqController.java

@RestController@RequestMapping("")public class RabbitMqController {    @Autowired    private Producer producer;    @GetMapping("/send")    public String send() {        producer.send();        return "发送完成";    }}

9. 测试

使用postman或浏览器使用Get方法请求http://localhost:20001/send，生产者会向RabbitMQ的test队列发送5条消息：

生产者发送消息，序号为: 0生产者发送消息，序号为: 1生产者发送消息，序号为: 2生产者发送消息，序号为: 3生产者发送消息，序号为: 4

可以看出序号为2的消息3次被消费者2接收，消费者2也3次发送拒绝应答，直到第4次才被消费者1接收，并返回确认应答。

END

// 2021年1月12日更新Demo
Demo from Gitee

SpringBoot整合RabbitMQ 手动应答 简单demo