程序员经验分享-hwhale

RabbitMQ系列（十五）RabbitMQ进阶-SprintBoot集成RabbitMQ使用

2023-11-15

RabbitMQ进阶-SprintBoot集成RabbitMQ使用

1.构建项目
1.1 Spring Init创建项目

我们之选了RabbitMQ、Web、Lombok插件简单的几个组成
项目组成

SpringBoot 2.4.4 + RabbitMQ  AMQP版本 2.4.4

在这里插入图片描述

1.2 新建项目包

项目结构如下所示

-java
 --com.jzj.bootmqtest
   ---bean
   ---config
   --controller
   --rabbitmq
   --task

   --BootmqtestApplication
   
-resources
--static
--templates
 --application.yml

在这里插入图片描述

2.初始化RabbitMQ
2.1 项目配置Pom文件及application.yml

Pom文件

<?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 https://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.4.4</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.jzj</groupId>
    <artifactId>bootmqtest</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>bootmqtest</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>

        <!-- hutool 工具类-->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.6.1</version>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

application.yml 配置

#项目端口
server:
  port: 9672

spring:
  #应用名称
  application:
    name: jzj-sprintboot-mqtest
  #mq配置
  rabbitmq:
    addresses: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    #开启消息确认机制 confirms
    publisher-confirm-type: correlated
    publisher-returns: true
    #采用手动应答方式
    listener:
      simple:
        acknowledge-mode: manual
2.2 新建实体Bean

在com.jaj.bootmqtest.bean下新建Student实体Bean，用来作为消息体，发送Object的消息体

package com.jzj.bootmqtest.bean;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;

@Setter
@Getter
@AllArgsConstructor
public class Student {
    private Integer id;
    private String name;
    private Integer age;
}

在controller下面新建测试SiteController，用来测试消息

package com.jzj.bootmqtest.controller;

import com.jzj.bootmqtest.bean.Student;
import com.jzj.bootmqtest.config.QueueConfig;
import com.jzj.bootmqtest.rabbitmq.BaseMqProducer;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@ResponseBody
@RestController
@RequestMapping("test")
public class SiteController {


    @Resource
    private BaseMqProducer mqProducer;


    @RequestMapping("ping")
    public Object ping() {
        return "pong";
    }


    @RequestMapping("sendA")
    public Object mqtestA() {
        Student student = new Student(1, "jiazijie-AAA", 18);
        mqProducer.sendMsg(student, QueueConfig.RK_QUEUE_TEST_A);
        return "AAA";
    }

    @RequestMapping("sendB")
    public Object mqtestB() {
        Student student = new Student(2, "jiazijie-BBB", 30);
        mqProducer.sendMsg(student, QueueConfig.RK_QUEUE_TEST_B);
        return "BBB";
    }
}
2.3 RabbitMQ配置
2.3.1 Exchange交换机配置

ExchangeConfig.java

package com.jzj.bootmqtest.config;

import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class ExchangeConfig {

    public static final String MAIN_EXCHANGE = "mainExchange";
    public static final String DEAD_EXCHANGE = "deadExchange";

    /**
     * 声明一个主题交换机，作为默认交换机 ，名字： mainExchange
     *
     * @return
     */
    @Bean(name = MAIN_EXCHANGE)
    public DirectExchange mainExchange() {
        return new DirectExchange(MAIN_EXCHANGE);
    }

    /**
     * 声明一个直连交换机、作为死亡交换机，名字deadExchange，为了方便接收消息，我们声明fanout类型，有消息就转发
     *
     * @return
     */
    @Bean(name = DEAD_EXCHANGE)
    public FanoutExchange deadExchange() {
        return new FanoutExchange(DEAD_EXCHANGE);
    }
}
2.3.2 Queue绑定信息

QueueConfig.java

package com.jzj.bootmqtest.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

import static com.jzj.bootmqtest.config.ExchangeConfig.DEAD_EXCHANGE;
import static com.jzj.bootmqtest.config.ExchangeConfig.MAIN_EXCHANGE;

@Configuration
public class QueueConfig {

    public static final String QUEUE_TEST_A = "queue_test_AAA";
    public static final String RK_QUEUE_TEST_A = "rk.queue_test_AAA";

    public static final String QUEUE_TEST_B = "queue_test_BBB";
    public static final String RK_QUEUE_TEST_B = "rk.queue_test_BBB";

    @Autowired
    @Qualifier(MAIN_EXCHANGE)
    DirectExchange mainExchange;
    @Autowired
    @Qualifier(DEAD_EXCHANGE)
    FanoutExchange deadExchange;

    /**
     * 声明一个
     *
     * @return
     */
    @Bean
    @Qualifier("queueAAA")
    public Queue queueAAA() {
        //声明 队列名称及队列持久化
        return new Queue(QUEUE_TEST_A);
    }

    /**
     * 声明队列，绑定死信队列交换机参数
     */
    @Bean
    @Qualifier("queueBBB")
    public Queue queueBBB() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        args.put("x-dead-letter-routing-key", RK_QUEUE_TEST_B);
        return new Queue(QUEUE_TEST_B, true, false, false, args);
    }


    /**
     * 声明对立绑定的Exchange 及 Routingkey
     *
     * @return
     */
    @Bean
    public Binding bindingQueueTest1() {
        return BindingBuilder.bind(queueAAA()).to(mainExchange).with(RK_QUEUE_TEST_A);
    }


    /**
     * 声明队列绑定，绑定 mainExchange 及 RoutingKey
     */
    @Bean
    public Binding bindingQueueTest2() {
        return BindingBuilder.bind(queueBBB()).to(mainExchange).with(RK_QUEUE_TEST_B);
    }


    /**
     * 声明死信队列，用于接收 死信交换机过来的消息
     */
    @Bean
    @Qualifier("deadQueue")
    public Queue deadQueue() {
        return new Queue("dead_queue", true);
    }

    /**
     * 声明死信队列绑定，只要是rk开头的全都接收
     */
    @Bean
    public Binding bindingDeadQueue() {
        return BindingBuilder.bind(deadQueue()).to(deadExchange);
    }


}
2.3.3 生产者

在此rabbitmq包下，新建生产者，用基类生产者来定义MQ生产者，消息发送时候只需要注入 BaseMqConsumer就可以了
BaseMqProducer.java

package com.jzj.bootmqtest.rabbitmq;

import cn.hutool.json.JSONUtil;
import com.jzj.bootmqtest.task.SendMqTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskRejectedException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.UUID;

/**
 * <mq消息发送>.
 *
 * @author peng cheng
 * @date 2019/12/23
 * @since
 */
@Component
public class BaseMqProducer {

    private static final Logger LOGGER = LoggerFactory.getLogger(BaseMqProducer.class);

    /**
     * 最大线程数目
     */
    private static final Integer RETRY_TIMES = 50;

    /**
     * 消息优先级设置
     */
    private Integer msgPriority;


    @Autowired
    private AmqpTemplate amqpTemplate;

    @Autowired
    private ThreadPoolTaskExecutor mqSenderExecutor;

    @Autowired
    private DirectExchange mainExchange;


    /**
     * 发送消息  采用默认的交换机
     *
     * @param msg
     * @param routingKey
     */
    public void sendMsg(Object msg, String routingKey) {
        sendMsg(msg, mainExchange.getName(), routingKey);
    }


    /**
     * 发送消息 指定交换机和路由键
     *
     * @param exchange
     * @param routingKey
     * @param mqMsg
     */
    public void sendMsg(Object mqMsg, String exchange, String routingKey) {
        if (null == mqMsg) {
            return;
        }
        try {
            Message message = generateMessage(mqMsg);
            LOGGER.info("RabbitMQ sendMsg exchange={},routingKey={},messageId={},msgBody={}", exchange, routingKey, message.getMessageProperties().getMessageId(),
                    new String(message.getBody()));

            threadExecute(new SendMqTask(exchange, routingKey, message, amqpTemplate));
        } catch (AmqpException e) {
            LOGGER.error("booster-notify sendMsq failed exchange=" + exchange + ",routingKey=" + routingKey + ",mqMsg=" + JSONUtil.toJsonStr(mqMsg), e);
        }
    }


    /**
     * 消息处理
     *
     * @param msg
     * @return
     */
    public Message generateMessage(Object msg) {
        String msgBody = null;
        try {
            msgBody = JSONUtil.toJsonStr(msg);

            return MessageBuilder.withBody(msgBody.getBytes())
                    .setMessageId(UUID.randomUUID().toString())
                    .setTimestamp(new Date())
                    .setPriority(msgPriority)
                    .build();
        } catch (Exception e) {
            LOGGER.error("generateMessage error :" + e.getMessage(), e);
        }
        return null;
    }


    /**
     * thread execute
     *
     * @param
     */
    private void threadExecute(Runnable var) {
        for (int i = 0; i < RETRY_TIMES; i++) {
            try {
                mqSenderExecutor.execute(var);
                return;
            } catch (TaskRejectedException e) {
                LOGGER.error("RabbitMQ线程池满请等待，等待次数:" + i + " 错误信息:" + e.getMessage());

                // sleep 100 ms
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e2) {
                    LOGGER.error("sleep error! " + e2.getMessage(), e2);
                }
            }
        }

        mqSenderExecutor.execute(var);
    }


    public void setMsgPriority(Integer msgPriority) {
        this.msgPriority = msgPriority;
    }

}
2.3.4 消费者

我们模板设计模式、制定消费者抽象基类，所有的消费者都继承实现基类就可以了
BaseMqConsumer.java

package com.jzj.bootmqtest.rabbitmq;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * 封装消费者抽象类
 * 每个消费者只需要关注自己的Listener 及 消息的处理，其余的工作都放在抽象模板中做
 */
public abstract class BaseMqConsumer {

    private static final Logger LOGGER = LoggerFactory.getLogger(BaseMqConsumer.class);

    /**
     * @param message
     * @return
     * @throws Exception
     */
    protected abstract boolean handlerMessage(Message message) throws Exception;

    /**
     * 获取消息消费异常后，重试的次数
     */
    protected abstract int initMaxRetryTimes();

    /**
     * 处理公共信息
     *
     * @param channel 通道
     * @param message 消息体
     */
    public void onMessage(Channel channel, Message message) throws Exception {
        String msgBody = new String(message.getBody(), StandardCharsets.UTF_8);
        // 获取重试次数
        final int maxRetryTimes = initMaxRetryTimes();

        String messageId = message.getMessageProperties().getMessageId();
        LOGGER.info(String.format("RabbitMQ receive msg, messageId:%s message:%s", messageId, msgBody));

        //先设置 处理结果为False
        boolean processResult = false;
        try {
            //开始处理消息
            processResult = handlerMessage(message);
        } catch (Exception e) {
            String errorMsg = e.getMessage() == null ? "" : e.getMessage();
            LOGGER.error(String.format("Rabbitmq Comsumer process error, messageId:%s,messageBody:%s, errorMsg:%s", messageId, msgBody, errorMsg), e);
        } finally {
            Long retryNum = getRetryCountFromMessage(message.getMessageProperties());
            if (retryNum > maxRetryTimes) {
                processResult = true;
                LOGGER.error(String.format("Rabbitmq 当前消息体重试操作超过[" + maxRetryTimes + "]次, 放弃重试!, messageId:%s retryNum:%s message:%s", messageId, retryNum, msgBody));
            }
            ackHandler(channel, message, processResult);
        }
    }

    protected void ackHandler(Channel channel, Message message, boolean processResult) throws IOException {
        if (processResult) {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            // 不成功, 丢到dead letter (second param true:requeue false:dead letter)
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    private Long getRetryCountFromMessage(MessageProperties properties) {
        //此处最好把 Message中的消息ID放入Redis，用分布式锁获取Redis的值 来判断当前的MessageId来到过几次了，用来解决幂等性
        // 方便测试，后面我们测试时候，设置为3次,表示这是第3次到来，防止消息无线消费，导致CPU暴增
        return 0L;
    }

}

消费者A，对应A队列 QueueAListener.java

package com.jzj.bootmqtest.rabbitmq;


import cn.hutool.json.JSONUtil;
import com.jzj.bootmqtest.bean.Student;
import com.jzj.bootmqtest.config.QueueConfig;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class QueueAListener extends BaseMqConsumer {

    private static final Logger LOGGER = LoggerFactory.getLogger(QueueAListener.class);


    /**
     * 绑定队列
     */
    @Override
    @RabbitHandler
    @RabbitListener(queues = QueueConfig.QUEUE_TEST_A)
    public void onMessage(Channel channel, Message message) throws Exception {
        super.onMessage(channel, message);
    }

    @Override
    protected int initMaxRetryTimes() {
        return 1;
    }


    @Override
    protected boolean handlerMessage(Message message) {
        boolean result = false;
        try {
            result = processSendResult(message);
        } catch (Exception e) {
            LOGGER.error("邮件消息 处理 异常 ", e);
        }
        return result;
    }

    /**
     * 解析邮件发送结果，且尝试计数
     * <p>
     * 此处 返回的都是 正确发送的消息，失败重试后 避免重新路由
     *
     * @param message 邮件消息
     * @return 处理结果
     */
    private boolean processSendResult(Message message) {

        // 解析MQ 邮件消息
        String msgBody = new String(message.getBody());

        Student msg = JSONUtil.toBean(msgBody, Student.class);
        if (null == msg) {
            LOGGER.error("转化失败 msg:{}", msgBody);
            throw new RuntimeException("邮件消息解析失败");
        }

        LOGGER.info("AAA处理消息:" + msgBody);
        return true;
    }


}

消费者B，对应B队列 QueueBListener.java

package com.jzj.bootmqtest.rabbitmq;


import cn.hutool.json.JSONUtil;
import com.jzj.bootmqtest.bean.Student;
import com.jzj.bootmqtest.config.QueueConfig;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class QueueBListener extends BaseMqConsumer {

    private static final Logger LOGGER = LoggerFactory.getLogger(QueueBListener.class);


    /**
     * 绑定队列
     */
    @Override
    @RabbitHandler
    @RabbitListener(queues = QueueConfig.QUEUE_TEST_B)
    public void onMessage(Channel channel, Message message) throws Exception {
        super.onMessage(channel, message);
    }

    @Override
    protected int initMaxRetryTimes() {
        return 2;
    }


    @Override
    protected boolean handlerMessage(Message message) {
        boolean result = false;
        try {
            result = processSendResult(message);
        } catch (Exception e) {
            LOGGER.error("邮件消息 处理 异常 ", e);
        }
        return result;
    }

    /**
     * 解析邮件发送结果，且尝试计数
     * <p>
     * 此处 返回的都是 正确发送的消息，失败重试后 避免重新路由
     *
     * @param message 邮件消息
     * @return 处理结果
     */
    private boolean processSendResult(Message message) {

        // 解析MQ 邮件消息
        String msgBody = new String(message.getBody());

        Student msg = JSONUtil.toBean(msgBody, Student.class);
        if (null == msg) {
            LOGGER.error("转化失败 msg:{}", msgBody);
            throw new RuntimeException("邮件消息解析失败");
        }

        //测试异常信息，直接抛出异常
//        throw new RuntimeException("邮件消息BBB解析失败");

        LOGGER.info("BBB 处理消息:" + msgBody);
        return true;
    }


}
2.3.5 MQ任务

设计SendMqTask 用来接收MQ消息发送者的消息，实现消息的发送

package com.jzj.bootmqtest.task;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;

/**
 * 单独发送的线程
 */
public class SendMqTask implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(SendMqTask.class);

    /**
     * 交换机
     */
    private String exchange;
    /**
     * Routingkey
     */
    private String rk;
    /**
     * 发送的消息
     */
    private Message message;
    /**
     * amqpTemplate
     */
    private AmqpTemplate amqpTemplate;

    public SendMqTask(String exchange, String rk, Message message, AmqpTemplate amqpTemplate) {
        this.exchange = exchange;
        this.rk = rk;
        this.message = message;
        this.amqpTemplate = amqpTemplate;
    }


    @Override
    public void run() {
        amqpSend(exchange, rk, message);
    }

    /**
     * amqp发送封装逻辑
     *
     * @param rk
     * @param message
     */
    private void amqpSend(final String exchange, final String rk, final Message message) {
        try {
            amqpTemplate.convertAndSend(exchange, rk, message);
        } catch (AmqpException e) {
            LOGGER.info(String.format("send error!!! messageId:%s message:%s",
                    message.getMessageProperties().getMessageId(), new String(message.getBody())));
            LOGGER.error("send message error: " + e.getMessage(), e);
        }
    }
}
2.4 启动类

启动类BootmqtestApplication.java

package com.jzj.bootmqtest;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class BootmqtestApplication {

	public static void main(String[] args) {
		SpringApplication.run(BootmqtestApplication.class, args);
	}

}
3.启动测试
3.1 项目启动

项目成功启动
在这里插入图片描述

3.2 Rabbitmq界面

交换机信息如下：
在这里插入图片描述
3个队列信息
在这里插入图片描述

3.3 执行结果

执行Controller sendA 给A队列发一条消息 curl 127.0.0.1:9672/test/sendA
执行结果可以看到，先发送消息，然后Listener接收消息，然后处理消息，处理成功消费完毕，B队列也一样
在这里插入图片描述

3.4 异常测试
3.4.1 basicReject 拒绝测试

现在基类BaseMqConsumer中 逻辑是消息处理成功，就确认消息，不成功就Reject消息
在这里插入图片描述
而且Reject中的参数requeue是false，表明拒绝后到私信队列，我们制造个异常，让他处理失败，看看

修改B的Listener的processSendResult方法，因为B有死信队列配置，让他抛出异常，处理失败，走下面的else分支

    private boolean processSendResult(Message message) {

        // 解析MQ 邮件消息
        String msgBody = new String(message.getBody());

        Student msg = JSONUtil.toBean(msgBody, Student.class);
        if (null == msg) {
            LOGGER.error("转化失败 msg:{}", msgBody);
            throw new RuntimeException("邮件消息解析失败");
        }

        //测试异常信息，直接抛出异常
        throw new RuntimeException("邮件消息BBB解析失败");

//        LOGGER.info("BBB 处理消息:" + msgBody);
//        return true;
    }

重启项目，再次发送sendB消息，发送完、接收消息，处理失败，抛出异常
在这里插入图片描述
死信队列中接收到消息
在这里插入图片描述

3.4.2 basicReject 拒绝、重新路由测试

我们可以看到 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
消息拒绝后是 requeue是false，表明不再路由，我们改为true看看，是否会重新路由
！！！！！！！ 注意，此处会有死循环，抛出异常->basicReject->重新路由->再次消费->再抛出异常…无限循环

    protected void ackHandler(Channel channel, Message message, boolean processResult) throws IOException {
        if (processResult) {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            // 不成功, 丢到dead letter (second param true:requeue false:dead letter)
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        }
    }

重启项目再次发送B
在这里插入图片描述

我们现在把循环次数变为3尝试一下，重启，可以看到，消息消费了3次后，就自动丢弃了，不会再无限循环

    private Long getRetryCountFromMessage(MessageProperties properties) {
        //此处最好把 Message中的消息ID放入Redis，用分布式锁获取Redis的值 来判断当前的MessageId来到过几次了，用来解决幂等性
        // 方便测试，后面我们测试时候，设置为3次,表示这是第3次到来，防止消息无线消费，导致CPU暴增
        return 3L;
    }

重试结果，Rabbitmq 当前消息体重试操作超过[2]次, 放弃重试!
在这里插入图片描述

4.常见错误
4.1 reply-text=PRECONDITION_FAILED - unknown delivery tag 1

2021-04-06 21:46:41.708 ERROR 7560 — [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory : Shutdown Signal: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)

该错误是因为 没有配置手动消息确认导致的，需要在配置文件application.yml中新增
publisher-confirm-type: correlated
publisher-returns: true
#采用手动应答方式
listener:
simple:
acknowledge-mode: manual

spring:
  #应用名称
  application:
    name: jzj-sprintboot-mqtest
  #mq配置
  rabbitmq:
    addresses: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    #开启消息确认机制 confirms
    publisher-confirm-type: correlated
    publisher-returns: true
    #采用手动应答方式
    listener:
      simple:
        acknowledge-mode: manual
4.2 spring.rabbitmq.publisher-confirms过时解决

该错误是因为 rabbitmq升级后，这对消息确认的方式变了，将原来的publisher-confirms 修改为 publisher-confirm-type 配置

    publisher-confirm-type: correlated

在这里插入图片描述

至此 我们的 Rabbitmq结合SpringBoot 简单完毕，下一节我们讲一下 批量发送消息

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

MQ中间件

Springboot整合MQ

Rabbitmq 实战

Springboot MQ实战

Springboot集成MQ

RabbitMQ系列（十五）RabbitMQ进阶-SprintBoot集成RabbitMQ使用 的相关文章

  • linux移动文件到另一个文件夹命令_Linux 使用 mv 命令重命名或移动文件

    使用方法 split help 用法 mv 选项 T 源 目标 mv 选项 源 目标 mv 选项 t 目标 源 重命名或者移动文件 目录 长选项必须使用的参数对于短选项时也是必需使用的 b 类似 backup 选项 但不接受参数 覆盖前先备

  • ZooKeeper（八）伸缩性

    一 ZooKeeper中Observer 1 1 ZooKeeper角色 经过前面的介绍 我想大家都已经知道了在ZooKeeper集群当中有两种角色Leader和Follower Leader可以接受client 请求 也接收其他Serve

  • 记一次JVM参数调优经历

    1 背景 定时任务每秒向客户端推送10w全量数据 预计最大有30M 两个用户系统勉强能用 六个 几分钟后系统不可用 频发full gc 导致系统不可用 2 最终方案 3 常见参数介绍 3 1 jps jps查看所有java进程编号 相当与

  • 顺序栈的基本操作（初始化、销毁、判空等）

    1 实验目的 通过该实验 让学生掌握栈的相关基本概念 认识栈是插入和删除集中在一端进行的线性结构 掌握栈的 先入后出 操作特点 栈在进行各类操作时 栈底指针固定不动 掌握栈空 栈满的判断条件 2 实验内容 用顺序存储结构 实现教材定义的栈的

  • 最大子数组之和对应的子数组

    给定一个整数数组 包含正负数 找到一个具有最大和的子数组 返回其最大的子数组 public static List

  • [译]理解HTTP/304响应

    原文 http www telerik com automated testing tools blog eric lawrence 12 11 06 understanding http 304 responses aspx 刚刚开始使用

  • 【JetBrains】安装使用技巧

    JetBrains 使用 JetBrains Toolbox 管理 IDE 远程开发 Gateway 通过 SSH 连接 疑难杂症 1 部署失败 使用 JetBrains Toolbox 管理 IDE 下载 Toolbox 工具 解压运行

  • Wireshark常用命令

    目录 页面 命令 不定期更新我自己遇到的语法 页面 命令 数据链路层 筛选mac地址为04 f9 38 ad 13 26的数据包 eth src 04 f9 38 ad 13 26 筛选源mac地址为04 f9 38 ad 13 26的数据

  • 模板特化

    上一篇 模板与重载 里 我遇见了想同时使用模板函数与非模板函数的情况 后来才知道 其实并不需要 当我想对某些特定的类型进行特殊操作时 只需要使用模板特化就可以 所谓特化 就是说对于模板函数 对于某些类型可能需要特殊处理 所以进行特殊化 可以

  • OpenApi-Generator：简化RESTful API开发流程

    目录 1 OpenAPI Generator简介 1 1 OpenAPI Generator是什么 1 2 为什么选择OpenAPI Generator 1 3 谁需要 OpenAPI Generator 2 OpenAPI 2 0规范 2

随机推荐

  • 单臂路由实现原理

    一 概述 单臂路由 router on a stick 是指在路由器的一个接口上通过配置子接口 或 逻辑接口 并不存在真正物理接口 的方式 实现原来相互隔离的不同VLAN 虚拟局域网 之间的互联互通 单臂路由的子接口 路由器的物理接口可以被

  • Python Pandas 处理空数据/缺失数据 dropna fillna，增加/更新列 assign，分层 qcut，向量函数

    Pandas 处理空数据 缺失数据 增加 更新列 分层 向量函数 数据准备 一 处理缺失数据 1 1 去除有缺失数据的行 dropna 1 2 替换缺失数据 fillna 二 增加 更新列 2 1 指定生成列的方式 2 2 复制现有的列生成

  • dataframe的索引遍历_pandas

    今天是pandas数据处理专题第三篇文章 我们来聊聊DataFrame中的索引 上篇文章当中我们简单介绍了一下DataFrame这个数据结构的一些常见的用法 从整体上大概了解了一下这个数据结构 今天这一篇我们将会深入其中索引相关的应用方法

  • 开发一个APP需要多少钱？

    作为一个移动端开发人员 我们可能被外行朋友或者被客户问及最多的一个问题就是 开发一个APP需要多少钱 不错 这个是大家特别关心的问题 也是互联网公司非常重视的一个问题 因为涉及到自己的成本问题 作为APP开发人员 站在产品经理的角度来给大家

  • windows 2008 32位IIS 服务器转到64位后的各种错误，以及解决方法

    之前在32位IIS服务器上没有问题 发布到64位出现各种错误 请检查以下几项 因各系统不一样 有则检查 无则跳过 重点第4点 1 先安装IIS 后安装 net 4 0环境 否则要重新注册iis windir Microsoft NET Fr

  • 机器学习可视化：模型评估和参数调优

    本篇文章详细阐述机器学习模型评估和参数调优 将主要围绕两个问题来阐述 知其所以然 当你选择的一个机器学习模型运行时 你要知道它是如何工作的 青出于蓝 更进一步 你得知道如何让此机器学习模型工作的更优 模型评估的方法 一般情况来说 F1评分或

  • 第四届蓝桥杯省赛JavaB组第六题三部排序

    标题 三部排序 一般的排序有许多经典算法 如快速排序 希尔排序等 但实际应用时 经常会或多或少有一些特殊的要求 我们没必要套用那些经典算法 可以根据实际情况建立更好的解法 比如 对一个整型数组中的数字进行分类排序 使得负数都靠左端 正数都靠

  • 在阿里云上运行hadoop遇到的50070，9000无法访问问题

    问题 我在阿里云上运行namenode和腾讯云上运行datanode 在hadooop配置完之后 运行hdfs 发现没有namenode 然后查看namenode的日志 日志显示50070端口被占用 9000端口拒绝服务 但是通过natst

  • vue - 实现页面全屏文字水印效果，类似 word 插入的自定义水印（支持单页或整个项目全部页面 “选择性“ 插入，可自定义水印文字、大小样式等，也能动态设置文字）和页面一同渲染，无任何卡顿示例源码

    效果图 代码干净简洁 示例源码注释详细 无任何乱七八糟的代码 本文实现了 单页或整个项目所有页面的全屏水印效果 支持自定义水印文字 可 动态 设置文字内容 你只需要复制本文提供的封装方法 直接在页面中或 App vue 中引入即可生效 只需

  • vue3+element-plus实现表格多选功能（可以清除选项或分页保留选项）

    如图所示 在实际开发中 数据量大的表格基本都添加上了分页功能 每个页面请求的数据回交换更新 第一页的选中效果在跳转至第二页后 如果没有做相关处理 选中项会被清空 具体解决方法如下 在需要处理的表格标签中加上 row key getRowKe

  • 第五章-CSRF漏洞

    第五章 CSRF漏洞 第一节 CSRF原理介绍 1 1 CSRF漏洞定义 CSRF Cross site request forery 跨站请求伪造 也被称为One Click Attack或者Session Riding 通常缩写为CSR

  • k8s组件理解

    一 k8s组件交互关系由下图可大致体现 二 k8s master组件理解 1 kube apiserver组件 kube apiserver Kubernetes kubernets API server 提供了k8s各类资源对象的增删改查

  • EasyTalking微博系统

    EasyTalking微博系统 摘要 随着互联网的迅猛发展 人们的日常生活 学习工作已经离不开网络 人们的生活和工作在未来的生活中将越来越依赖于计算机网络技术的发展 越来越网络化 电子化 虚拟化 便捷化 Internet目前的应用历程和发展

  • 如何const定义一个不可变数组

    有个常见的面试题 我们知道 const是es6中新增用于定义常量 但是对于引用类型来说 const 所说的常量 是指 对应的指针或者说地址是常量 那么 如果我们要求 我们定义的数组里面的元素也是不可改变的呢 先来看现象 const a 1

  • webgl--attribute相关使用

    attribute 是存储限定符 是专门用于向外部导出与点位相关的对象的 这类似于es6模板语法中export vec4 是变量类型 vec4是4维矢量对象 a position 是变量名 之后在js中会根据这个变量名导入变量 这个变量名是

  • [CTSC2008]网络管理Network【树状数组+主席树】

    题目链接 题意 有一棵N个点的树 每个点有对应的权值 现在有这样的操作 0 a b 将a点的权值改成为b k a b 询问a到b的链上第k大的权值是几 我们可以用dfs序的树上差分的方式来解决这个问题 可以发现 求u到v的信息 其实就是求u

  • 【c++】角度与弧度转换

    角度转换为弧度 double angle to radian double degree double min double second double flag degree lt 0 1 0 1 0 判断正负 if degree lt

  • matlab与机器学习（二）程序基本操作（含代码解析）

    备注 之间的为注释内容 注释后面的内容 matlab形成自己的编程风格更有利于简洁的编程界面 图像的初步处理可应用于论文撰写上 文章更有说服力 程序调试有利于发现逻辑问题 都是必备的基础知识 I 清空环境变量及命令 clear all cl

  • Python selenium基础用法详解

    活动地址 CSDN21天学习挑战赛 学习的最大理由是想摆脱平庸 早一天就多一份人生的精彩 迟一天就多一天平庸的困扰 学习日记 目录 学习日记 一 Selenium库介绍 1 Selenium简介 2 Selenium的安装 3 安装浏览器驱

  • RabbitMQ系列（十五）RabbitMQ进阶-SprintBoot集成RabbitMQ使用

    RabbitMQ进阶 SprintBoot集成RabbitMQ使用 文章目录 RabbitMQ进阶 SprintBoot集成RabbitMQ使用 1 构建项目 1 1 Spring Init创建项目 1 2 新建项目包 2 初始化Rabbi

热门标签