谷粒商城-消息队列Rabbitmq

RabbitMq参考文档

在谷粒商城项目中使用消息队列主要有以下几个重要原因：

1. 异步处理提高性能

• • 场景示例：在订单系统中，当用户提交订单后，系统需要完成多个操作，如更新库存、生成订单记录、发送订单通知等。如果这些操作都同步进行，用户需要等待所有操作完成才能得到响应，这会导致响应时间变长。
  • 使用消息队列后的优化：通过消息队列，如 RabbitMQ，可以将这些操作中的部分任务异步处理。例如，当订单创建成功后，将更新库存和发送通知的任务以消息的形式发送到消息队列。库存服务和通知服务可以从消息队列中获取消息并进行处理，而订单系统不需要等待这些操作完成就可以返回给用户订单创建成功的信息，大大提高了系统的响应速度和吞吐量。

1. 流量削峰填谷

• • 高并发场景示例：在电商平台的促销活动期间，如 “双 11”“618” 等，会出现大量的订单请求。如果没有消息队列，这些请求直接涌向数据库和后端服务，可能会导致数据库和服务器不堪重负，出现系统崩溃或者响应缓慢的情况。
  • 消息队列的缓冲作用：消息队列可以作为一个缓冲层。在流量高峰时，订单请求等消息先堆积在消息队列中，后端服务按照自己的处理能力从消息队列中获取消息并处理。这样就避免了瞬间的高流量对系统造成过大的压力，实现了流量的削峰。而在流量低谷时，消息队列中的消息可以继续被处理，达到填谷的效果。

1. 解耦服务之间的依赖关系

• • 服务耦合问题示例：假设谷粒商城中的商品服务和库存服务紧密耦合，商品服务在更新商品信息后直接调用库存服务来更新库存。如果库存服务出现故障或者需要升级，商品服务可能会受到影响。
  • 通过消息队列解耦：引入消息队列后，商品服务只需要将更新库存的消息发送到消息队列，而不需要关心库存服务的具体实现和状态。库存服务从消息队列中获取消息并进行处理，这样两个服务之间的依赖关系得到了解耦。即使库存服务出现故障，商品服务仍然可以正常运行，消息会在库存服务恢复后继续被处理。

1、MQ简介

1)、为什么要用MQ：

• 异步处理

• 应用解耦

• 流量控制

2)、概述：

1.大多应用中，可通过消息服务中间件来提升系统异步通信、扩展解耦能力
2.消息服务中两个重要概念：
2.1消息代理（message broker）和目的地（destination）
2.2当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地。
3.消息队列主要有两种形式的目的地
3.1队列（queue）：点对点消息通信（point-to-point）
3.2主题（topic）：发布（publish）/订阅（subscribe）消息通信
4.点对点式：
4.1消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获取消息内容，消息读取后被移出队列
4.2消息只有唯一的发送者和接受者，但并不是说只能有一个接收者
5.发布订阅式
5.1发送者（发布者）发送消息到主题，多个接收者（订阅者）监听（订阅）这个主题，那么就会在消息到达时同时收到消息
6.JMS（Java Message Service）JAVA消息服务：
6.1基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现
7.AMQP（Advanced Message Queuing Protocol）
7.1高级消息队列协议，也是一个消息代理的规范，兼容JMS
7.2RabbitMQ是AMQP的实现

8.Spring支持
8.1spring-jms提供了对JMS的支持
8.2spring-rabbit提供了对AMQP的支持
8.3需要ConnectionFactory的实现来连接消息代理
8.4提供JmsTemplate、RabbitTemplate来发送消息
8.5@JmsListener（ JMS ）、 @RabbitListener （ AMQP）注解在方法上监听消息(代理发布的消息)
8.6@EnableJms、 @EnableRabbit开启支持
9.Spring Boot自动配置
9.1JmsAutoConfiguration
9.2RabbitAutoConfiguration
10.市面的MQ产品
ActiveMQ、RabbitMQ、RocketMQ、Kafka

2、RabbitMQ概念

1)、RabbitMQ简介：

RabbitMQ 是一个由 erlang 开发的 AMQP(Advanved Message Queue Protocol) 的开源实现。

2)、核心概念

Message
消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key （路由键）、 priority （相对于其他消息的优先权）、 delivery-mode （指出该消息可能需要持久性存储）等。

Publisher
消息的生产者，也是一个向交换器发布消息的客户端应用程序。

Exchange
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Exchange 有 4 种类型： direct( 默认 ) ，fanout，topic和 headers ，不同类型的 Exchange 转发消息的策略有所区别。

Queue
消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

Binding
绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
Exchange 和 Queue 的绑定可以是多对多的关系。

Connection
网络连接，比如一个 TCP 连接。

Channel
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的 TCP 连接内的虚拟连接， AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。

Consumer
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

Virtual Host
虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。

Broker
表示消息队列服务器实体

3、docker安装RabbitMQ

docker pull rabbitmq:3.8.2

不下载镜像，直接安装。默认会帮你下载

docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p 25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:3.8.2

• 4369, 25672 (Erlang发现&集群端口)
• 5672, 5671 (AMQP端口)
• 15672 (web管理后台端口)
• 61613, 61614 (STOMP协议端口)
• 1883, 8883 (MQTT协议端口)
• 官网地址：https://www.rabbitmq.com/networking.html

修改只要启动docker自动重启rabbitMQ

docker update rabbitmq --restart=always

登录rabbitmq控制台： http://192.168.119.127:15672

账号:guest
密码:guest

若没反应

进入rabbitmq的插件管理中

docker exec -it xxx /bin/bash

rabbitmq-plugins list

故障排除

[root@VM-8-12-centos ~]# docker exec -it 3e9 /bin/bash
root@3e959345fd18:/# rabbitmq-plugins list
Listing plugins with pattern ".*" ...
 Configured: E = explicitly enabled; e = implicitly enabled
 | Status: * = running on rabbit@3e959345fd18
 |/
[  ] rabbitmq_amqp1_0                  3.8.2
[  ] rabbitmq_auth_backend_cache       3.8.2
[  ] rabbitmq_auth_backend_http        3.8.2
[  ] rabbitmq_auth_backend_ldap        3.8.2
[  ] rabbitmq_auth_backend_oauth2      3.8.2
[  ] rabbitmq_auth_mechanism_ssl       3.8.2
[  ] rabbitmq_consistent_hash_exchange 3.8.2
[  ] rabbitmq_event_exchange           3.8.2
[  ] rabbitmq_federation               3.8.2
[  ] rabbitmq_federation_management    3.8.2
[  ] rabbitmq_jms_topic_exchange       3.8.2
[  ] rabbitmq_management               3.8.2
[  ] rabbitmq_management_agent         3.8.2
[  ] rabbitmq_mqtt                     3.8.2
[  ] rabbitmq_peer_discovery_aws       3.8.2
[  ] rabbitmq_peer_discovery_common    3.8.2
[  ] rabbitmq_peer_discovery_consul    3.8.2
[  ] rabbitmq_peer_discovery_etcd      3.8.2
[  ] rabbitmq_peer_discovery_k8s       3.8.2
[  ] rabbitmq_prometheus               3.8.2
[  ] rabbitmq_random_exchange          3.8.2
[  ] rabbitmq_recent_history_exchange  3.8.2
[  ] rabbitmq_sharding                 3.8.2
[  ] rabbitmq_shovel                   3.8.2
[  ] rabbitmq_shovel_management        3.8.2
[  ] rabbitmq_stomp                    3.8.2
[  ] rabbitmq_top                      3.8.2
[  ] rabbitmq_tracing                  3.8.2
[  ] rabbitmq_trust_store              3.8.2
[  ] rabbitmq_web_dispatch             3.8.2
[  ] rabbitmq_web_mqtt                 3.8.2
[  ] rabbitmq_web_mqtt_examples        3.8.2
[  ] rabbitmq_web_stomp                3.8.2
[  ] rabbitmq_web_stomp_examples       3.8.2

发现[ ] rabbitmq_management ==》代表没开启web页面访问权限

解决：

#容器内执行
rabbitmq-plugins enable rabbitmq_management

执行完后，管理页面就可以成功访问了。

不想每次都执行？那就自己加工一下即可

vim Dockerfile
FROM rabbitmq:latest
# 启用 RabbitMQ 管理插件
RUN rabbitmq-plugins enable rabbitmq_management
然后，在构建 Docker 镜像时使用该 Dockerfile：


docker build -t my-rabbitmq-image .

1)、创建一个交换机

2)、创建队列

3)、交换机绑定队列

删除交换机，先双击点击要删除的交换机，接着

4、Exchange类型

1)、RabbitMQ运行机制

AMQP中消息的路由过程和 Java 开 发者熟悉的 JMS 存在一些差别，AMQP中增加了 Exchange和 Binding 的角色。生产者把消息发布 到 Exchange 上，消息最终到达队列并被消费者接收，而 Binding 决定交换器的消息应该发送到那个队列。

2)、Exchange 类型

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、 fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键， headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了，所以直接 看另外三种类型：

Direct Exchange（点对点）

消息中的路由键（ routing key ）如果和Binding 中的 binding key 一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog” ，则只转发 routing key 标记为“ dog” 的消息，不会转发“dog.puppy” ，也不会转发“ dog.guard”等等。它是完全匹配、单播的模式。

Fanout Exchange

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

Topic Exchange

topic 交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。 它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开 。它同样也会识别两个通配符：符号“#” 和符号 “* ”。

# 匹配 0 个或多个单词， * 匹配一个单词。

发消息是发给exchange（交换机），

接收信息是接收queue（队列）

3)、测试

根据下图要求我们创建交换机和队列

1、exchange.direct

创建交换机

创建队列

绑定队列

发送消息

消费消息

2、exchange.fanout

创建交换机

绑定队列

发送消息

消费消息

3、exchange.topic

创建交换机

绑定队列

发送消息

（1）发送hello.news只匹配*.news的队列

（2）发送atguigu.news的匹配atguigu.#和*.news的队列

5、SpringBoot整合RabbitMQ

RabbitMQ的使用

1、引入amqp；RabbitAutoConfiguration就会自动生效
2、给容器中自动配置了RabbitTemplate、AmqpAdmin、CachingConnectionFactory、RabbitMessagingTemplate
    所有的属性都是
    @ConfigurationProperties(prefix = "spring.rabbitmq")
    public class RabbitProperties
3、给配置文件中配置 spring.rabbitmq 信息
4、@EnableRabbit 开启功能

gulimall-order

1)、pom导入amqp依赖

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

2)、添加配置（@ConfigurationProperties(prefix = "spring.rabbitmq")）注意配置前缀一定是spring.rabbitmq

spring.rabbitmq.host=192.168.119.127
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/

3)、主启动类添加@EnableRabbit注解

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

6、AmqpAdmin使用

作用：

创建exchange 交换机

创建queue 队列

创建binding 绑定

durable 持久化

autoDelete 交换机是否自动删除

@Slf4j
@SpringBootTest
class GulimallOrderApplicationTests {


@Autowired
AmqpAdmin amqpAdmin;

/**
 * 1、如何创建Exchange[hello-java-exchange]、Queue、Binding
 *      1)、使用AmqpAdmin进行创建
 * 2、如何收发消息
 */
@Test
void contextLoads() {
  
    DirectExchange directExchange = new DirectExchange("hello-java-exchange",true,false);
    amqpAdmin.declareExchange(directExchange);
    log.info("Exchange[{}]创建成功","hello-java-exchange");
}

@Test
public void createQueue(){
    Queue queue = new Queue("hello-java-queue",true,false,false);
    amqpAdmin.declareQueue(queue);
    log.info("Queue[{}]创建成功","hello-java-queue");
}

@Test
public void createBinding(){
    Binding binding = new Binding("hello-java-queue", Binding.DestinationType.QUEUE,"hello-java-exchange","hello.java",null);
    amqpAdmin.declareBinding(binding);
    log.info("Binding[{}]创建成功","hello-java-binding");
}
}

7、RabbitTemplate使用

作用：发送消息

@Slf4j
@SpringBootTest
class GulimallOrderApplicationTests {
@Autowired
AmqpAdmin amqpAdmin;

@Autowired
RabbitTemplate rabbitTemplate;


/**
 * 发送消息
 */
@Test
public void sendMessageTest() {
    OrderReturnApplyEntity orderReturnApplyEntity = new OrderReturnApplyEntity();
    orderReturnApplyEntity.setId(1L);
    orderReturnApplyEntity.setCreateTime(new Date());
    orderReturnApplyEntity.setReturnName("哈哈哈");
    //1、发送消息，如果发送的消息是个对象，
    我们会使用序列化机制，将对象写出去。对象必须实现Serializable
    String msg = "hello word";

    //2、配置MyRabbitConfig，让发送的对象类型的消息，可以是一个json
    rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", orderReturnApplyEntity);
    log.info("消息发送完成{}", orderReturnApplyEntity);
}
}

注意：

配置MyRabbitConfig，让发送的对象类型的消息，可以是一个json

添加“com.atguigu.gulimall.order.config.MyRabbitConfig”类，代码如下：

序列化

@Configuration
public class MyRabbitConfig {
/**
 * 使用JSON序列化机制，进行消息转换
 * 
 * @return
 */
@Bean
public MessageConverter messageConverter(){
    return new Jackson2JsonMessageConverter();
}
}

8、RabbitListener&RabbitHandler

作用：

接收消息

监听消息：使用@RabbitListener; 主启动类必须有@EnableRabbit

@RabbitListener：类+方法上（监听哪些队列即可）
@RabbitHandler：标在方法上(重载区分不同的消息)

RabbitListener用法

queues:声明需要监听的所有队列

org.springframework.amqp.core.Message

参数可以写以下内容：

1、Message message：原生消息详细信息。头+体

2、T<发送的消息类型> OrderReturnApplyEntity content

3、Channel channel 当前传输数据的通道

Queue：可以很多人都来监听。只要收到消息，队列删除消息，而且只能有一个收到此消息

场景：

1)、订单服务启动多个；同一个消息，只能有一个客户端收到

2)、只有一个消息完全处理完，方法运行结束，我们就可以接收到下一个消息

修改“com.atguigu.gulimall.order.GulimallOrderApplicationTests”类，代码如下：

/**
 * 同一个队列发送不同消息
 */
@Test
public void sendUniqueMessageTest() {
    for (int i = 0; i < 10; i++) {
        if (i % 2 == 0) {
            OrderReturnApplyEntity orderReturnApplyEntity = new OrderReturnApplyEntity();
            orderReturnApplyEntity.setId(1L);
            orderReturnApplyEntity.setCreateTime(new Date());
            orderReturnApplyEntity.setReturnName("哈哈哈");
            // 配置MyRabbitConfig，让发送的对象类型的消息，可以是一个json
            rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", orderReturnApplyEntity, new CorrelationData(UUID.randomUUID().toString()));
        } else {
            OrderEntity entity = new OrderEntity();
            entity.setOrderSn(UUID.randomUUID().toString());
            rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", entity, new CorrelationData(UUID.randomUUID().toString()));
        }
    }
}

修改“com.atguigu.gulimall.order.service.impl.OrderItemServiceImpl”类，代码如下：

package com.atguigu.gulimall.order.service.impl;
 
import com.atguigu.gulimall.order.entity.OrderEntity;
import com.atguigu.gulimall.order.entity.OrderReturnApplyEntity;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
import java.util.Map;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.atguigu.common.utils.PageUtils;
import com.atguigu.common.utils.Query;
 
import com.atguigu.gulimall.order.dao.OrderItemDao;
import com.atguigu.gulimall.order.entity.OrderItemEntity;
import com.atguigu.gulimall.order.service.OrderItemService;
 
import javax.swing.*;
 
@RabbitListener(queues = {"hello-java-queue"})
@Service("orderItemService")
public class OrderItemServiceImpl extends ServiceImpl<OrderItemDao, OrderItemEntity> implements OrderItemService {
 
    /**
     * queues:声明需要监听的所有队列
     *
     * org.springframework.amqp.core.Message
     * @param message
     *
     * 参数可以写以下内容
     * 1、Message message：原生消息详细信息。头+体
     * 2、T<发送的消息类型> OrderReturnApplyEntity content
     * 3、Channel channel 当前传输数据的通道
     *
     * Queue:可以很多人都来监听。只要收到消息，队列删除消息，而且只能有一个收到此消息
     * 场景：
     *       1）、订单服务启动多个；同一个消息，只能有一个客户端收到
     *       2)、只有一个消息完全处理完，方法运行结束，我们就可以接收到下一个消息
     *
     */
    //@RabbitListener(queues = {"hello-java-queue"})
    @RabbitHandler
    public void receiverMessage(Message message,OrderReturnApplyEntity content,
                                Channel channel) throws InterruptedException {
        //消息体
        byte[] body = message.getBody();
        //消息头属性信息
        MessageProperties properties = message.getMessageProperties();
        System.out.println("接收到消息...内容:" + content);
//        Thread.sleep(3000);
        System.out.println("消息处理完成=》"+content.getReturnName());
    }
 
    @RabbitHandler
    public void receiverMessage(OrderEntity orderEntity){
        System.out.println("接收到消息...内容:" + orderEntity);
 
    }
}

创建contorller

package com.atguigu.gulimall.order.controller;

import com.atguigu.gulimall.order.entity.OrderEntity;
import com.atguigu.gulimall.order.entity.OrderReturnReasonEntity;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

@RestController
public class RabbitmqController {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @GetMapping("/sendMessage")
    public String sendMessage(@RequestParam(value = "num",defaultValue = "10") Integer num){
        for (int i = 0; i < 10; i++) {
            if (i % 2 == 0){
                // 测试返回实体类===>引入OrderReturnReasonEntity
                OrderReturnReasonEntity reasonEntity = new OrderReturnReasonEntity();
                reasonEntity.setId(1L);
                reasonEntity.setName("哈哈哈");
                rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", reasonEntity,new CorrelationData(UUID.randomUUID().toString()));

            }else {
                OrderEntity orderEntity = new OrderEntity();
                orderEntity.setOrderSn("12345678");
                rabbitTemplate.convertAndSend("hello-java-exchange", "hello.java", orderEntity,new CorrelationData(UUID.randomUUID().toString()));
            }
        }
        return "ok";
    }
}

结果（重载实现）==》不同参数类型对应到不同的方法，返回不同的值

9、消息确认机制-发送端确认

RabbitMQ消息确认机制：

保证消息不丢失，可靠抵达，可以使用事务消息，性能下降250倍，为此引入确认机制
publisher confirmCallback 确认模式
publisher returnCallback 未投递到 queue 退回模式
consumer ack机制
 

1)、ConfirmCallback

spring.rabbitmq.publisher-confirms=true
在创建 connectionFactory 的时候设置 PublisherConfirms(true) 选项，开启confirmcallback 。
CorrelationData：用来表示当前消息唯一性。
消息只要被 broker 接收到就会执行 confirmCallback，如果是 cluster 模式，需要所有 broker 接收到才会调用 confirmCallback。
被 broker 接收到只能表示 message 已经到达服务器，并不能保证消息一定会被投递到目标 queue 里。所以需要用到接下来的 returnCallback 。

2)、ReturnCallback

• spring.rabbitmq.publisher-returns=true
• spring.rabbitmq.template.mandatory=true

• • confrim 模式只能保证消息到达 broker，不能保证消息准确投递到目标 queue 里。在有 些业务场景下，我们需要保证消息一定要投递到目标 queue 里，此时就需要用到return 退回模式。
  • 这样如果未能投递到目标 queue 里将调用 returnCallback ，可以记录下详细到投递数据，定期的巡检或者自动纠错都需要这些数据。

定制RabbitTemplate

服务器收到消息就回调

1、开启发送端确认
        1、spring.rabbitmq.publisher-confirms=true
        2、设置确认回调

2、消息抵达队列就回调

        1、#开启发送端抵达队列确认
         spring.rabbitmq.publisher-returns=true
             #只要抵达队列，以异步发送优先回调我们这个returnConfirm
         spring.rabbitmq.template.mandatory=true
       2、设置确认回调ReturnCallback
3、消费端确认（保证每个消息被正确消费，此时才可以保证broker删除这个消息）

修改application.properties

#开启发送端确认
#spring.rabbitmq.publisher-confirms=true #老版本springboot配置写法
spring.rabbitmq.publisher-confirm-type=correlated #新版本springboot配置写法
#开启发送端抵达队列确认
spring.rabbitmq.publisher-returns=true
#只要抵达队列，以异步发送优先回调我们这个returnConfirm
spring.rabbitmq.template.mandatory=true

添加“com.atguigu.gulimall.order.config.MyRabbitConfig”类，代码如下：

package com.atguigu.gulimall.order.config;
 
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;
 
@Configuration
public class MyRabbitConfig {
 
    @Autowired
    RabbitTemplate rabbitTemplate;
 
    /**
     * 使用JSON序列化机制，进行消息转换
     *
     * @return
     */
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
 
    /**
     * 定制RabbitTemplate
     * 服务器收到消息就回调
     * 1、开启发送端确认
     *      1、spring.rabbitmq.publisher-confirms=true
     *      2、设置确认回调
     * 2、消息抵达队列就回调
     *      1、#开启发送端抵达队列确认
     *         spring.rabbitmq.publisher-returns=true
     *         #只要抵达队列，以异步发送优先回调我们这个returnConfirm
     *         spring.rabbitmq.template.mandatory=true
     *      2、设置确认回调ReturnCallback
     *  3、消费端确认（保证每个消息被正确消费，此时才可以保证broker删除这个消息）
     */
    @PostConstruct   //MyRabbitConfig对象创建完以后，执行这个方法
    public void initRabbitTemplate(){
        //设置确认回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            /**
             * 只要消息抵达Broker就b = true
             * @param correlationData 当前消息的唯一关联数据（这个消息的唯一id）
             * @param b  消息是否成功收到
             * @param s 失败的原因
             */
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                System.out.println("confirm...correlationData["+correlationData+"]==>b["+b+"]s==>["+s+"]");
            }
        });
 
        //设置消息没有抵达队列的回调
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            /**
             * 只要消息没有投递给指定的队列，就触发这个失败回调
             * @param message 投递失败的消息详细信息
             * @param i 回复的状态码
             * @param s 回复的文本内容
             * @param s1 当时这个消息发给哪个交换机
             * @param s2 当时这个消息用哪个路由键
             */
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                System.out.println("Fail Message["+message+"]==>i["+i+"]==>s["+s+"]==>s1["+s1+"]==>s2["+s2+"]");
            }
        });
    }
}

bug：启动发现出现了循环依赖现象

***************************
APPLICATION FAILED TO START
***************************
 
Description:
 
The dependencies of some of the beans in the application context form a cycle:
 
┌─────┐
|  myRabbitConfig (field private org.springframework.amqp.rabbit.core.RabbitTemplate com.atguigu.gulimall.order.config.MyRabbitConfig.rabbitTemplate)
↑     ↓
|  rabbitTemplate defined in class path resource [org/springframework/boot/autoconfigure/amqp/RabbitAutoConfiguration$RabbitTemplateConfiguration.class]
↑     ↓
|  rabbitTemplateConfigurer defined in class path resource [org/springframework/boot/autoconfigure/amqp/RabbitAutoConfiguration$RabbitTemplateConfiguration.class]
└─────┘
 

我们可以看到自定义的mq配置中，注入了一个 MessageConverter

再来看RabbitTemplateConfiguration也有一个 MessageConverter

于是乎形成了

MyRabbitConfig 需要 RabbitTemplate

RabbitTemplate 需要 MessageConverter

MyRabbitConfig 又定义了 MessageConverter

反反复复,就形成了循环依赖

解决方法：

只要把 自定义 MessageConverter 与自定义 RabbitTemplate 不写在同一个类MyRabbitConfig 就行了

添加“com.atguigu.gulimall.order.config.MyMessageConverterConfig”类，代码如下：

@Configuration
public class MyMessageConverterConfig {
/**
 * 使用JSON序列化机制，进行消息转换
 */
@Bean
public MessageConverter messageConverter(){
    return new Jackson2JsonMessageConverter();
}

}
把MyRabbitConfig配置类中的消息转换配置删掉即可。

发送消息打印结果

confirm...correlationData[CorrelationData [id=17a39c15-d309-45a9-b93d-a0e67151a878]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=37d1b2dc-ddb9-40eb-933f-96729cb4cf36]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=4a86e085-6785-4c86-8b0f-22ac68b2d333]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=82026297-a94b-410c-a592-70f7709acad2]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=81ae0e70-156f-4afe-8f16-af16c1b9ff31]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=e97032ed-e460-474f-b518-e3bc472f3a6e]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=f88d5c09-c881-4193-b902-e3974895687e]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=84289b43-9025-4088-98b7-d7fb8f53355b]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=7cb6b25d-67e9-426f-9947-1cd6826f2dc4]]==>ack[true]s==>[null]
confirm...correlationData[CorrelationData [id=8cd46d86-2e73-4747-8192-e4d725173b02]]==>ack[true]s==>[null]

制造错误

打印

10、消息确认机制-消费端确认

Ack消息确认机制

消费者获取到消息，成功处理，可以回复Ack给Broker
basic.ack用于肯定确认；broker将移除此消息
basic.nack用于否定确认；可以指定broker是否丢弃此消息，可以批量
basic.reject用于否定确认；同上，但不能批量
默认自动ack，消息被消费者收到，就会从broker的queue中移除（重启立即删除队列）
queue无消费者，消息依然会被存储，直到消费者消费

消费者收到消息，默认会自动ack。但是如果无法确定此消息是否被处理完成，或者成功处理。我们可以开启手动ack模式
消息处理成功，ack()，接受下一个消息，此消息broker就会移除
消息处理失败，nack()/reject()，重新发送给其他人进行处理，或者容错处理后ack
消息一直没有调用 ack/nack 方法， broker 认为此消息正在被处理，不会投递给别人，此时客户端断开，消息不会被broker 移除，会投递给别人

消费端确认（保证每个消息被正确消费，此时才可以保证broker删除这个消息）

1、默认是自动确认的，只要消息接收到，客户端会自动确认，服务端就会移除这个消息

问题：

我们收到很多消息，自动回复给服务器ack,只有一个消息处理成功，宕机了。发生消息丢失

手动确认模式。只要我们没有明确告诉MQ,货物被签收，没有ACK,消息就一直unacked状态，即使Consumer宕机。消息不会丢失，会重新变为Ready,下一次有新的Consumer连接进来就发给他。

2、 1)、#手动确认收货（ack）

spring.rabbitmq.listener.simple.acknowledge-mode=manual

2)、channel.basicAck(deliveryTag,false);签收；业务成功完成就应该签收

channel.basicNack(deliveryTag,false,true);拒签：业务失败，拒签

添加application.properties

#手动确认收货（ack）
spring.rabbitmq.listener.simple.acknowledge-mode=manual

    @RabbitHandler
    public void receiverMessage(Message message, OrderReturnApplyEntity content, Channel channel) {
        // 消息体
        byte[] body = message.getBody();
        // 消息头属性信息
        MessageProperties properties = message.getMessageProperties();
        System.out.println("接收到消息...内容:" + content);
//        Thread.sleep(3000);
        System.out.println("消息处理完成=》" + content.getReturnName());
        // channel内按顺序自增的
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        System.out.println("deliveryTag:" + deliveryTag);
        // 签收货物，非批量模式
        try {
            if (deliveryTag % 2 == 0) {
                // 收货
                channel.basicAck(deliveryTag, false);
                System.out.println("签收了货物。。。" + deliveryTag);
            } else {
                // 退货requeue=false 丢弃  requeue=true发挥服务器，服务器重新入队。
                channel.basicNack(deliveryTag, false, true);
                System.out.println("没有签收货物..." + deliveryTag);
            }
        } catch (Exception e) {
            //网络中断
        }
 
    }

接收到消息...内容:OrderReturnApplyEntity(id=1, orderId=null, skuId=null, orderSn=null, createTime=Mon May 20 16:31:49 CST 2024, memberUsername=null, returnAmount=null, returnName=哈哈哈0, returnPhone=null, status=null, handleTime=null, skuImg=null, skuName=null, skuBrand=null, skuAttrsVals=null, skuCount=null, skuPrice=null, skuRealPrice=null, reason=null, description述=null, descPics=null, handleNote=null, handleMan=null, receiveMan=null, receiveTime=null, receiveNote=null, receivePhone=null, companyAddress=null)
消息处理完成=》哈哈哈0
deliveryTag:1
没有签收货物...1
接收到消息...内容:OrderReturnApplyEntity(id=1, orderId=null, skuId=null, orderSn=null, createTime=Mon May 20 16:31:49 CST 2024, memberUsername=null, returnAmount=null, returnName=哈哈哈2, returnPhone=null, status=null, handleTime=null, skuImg=null, skuName=null, skuBrand=null, skuAttrsVals=null, skuCount=null, skuPrice=null, skuRealPrice=null, reason=null, description述=null, descPics=null, handleNote=null, handleMan=null, receiveMan=null, receiveTime=null, receiveNote=null, receivePhone=null, companyAddress=null)
消息处理完成=》哈哈哈2
deliveryTag:2
签收了货物。。。2
接收到消息...内容:OrderReturnApplyEntity(id=1, orderId=null, skuId=null, orderSn=null, createTime=Mon May 20 16:31:49 CST 2024, memberUsername=null, returnAmount=null, returnName=哈哈哈4, returnPhone=null, status=null, handleTime=null, skuImg=null, skuName=null, skuBrand=null, skuAttrsVals=null, skuCount=null, skuPrice=null, skuRealPrice=null, reason=null, description述=null, descPics=null, handleNote=null, handleMan=null, receiveMan=null, receiveTime=null, receiveNote=null, receivePhone=null, companyAddress=null)
消息处理完成=》哈哈哈4
deliveryTag:3
没有签收货物...3
接收到消息...内容:OrderReturnApplyEntity(id=1, orderId=null, skuId=null, orderSn=null, createTime=Mon May 20 16:31:49 CST 2024, memberUsername=null, returnAmount=null, returnName=哈哈哈6, returnPhone=null, status=null, handleTime=null, skuImg=null, skuName=null, skuBrand=null, skuAttrsVals=null, skuCount=null, skuPrice=null, skuRealPrice=null, reason=null, description述=null, descPics=null, handleNote=null, handleMan=null, receiveMan=null, receiveTime=null, receiveNote=null, receivePhone=null, companyAddress=null)
消息处理完成=》哈哈哈6
deliveryTag:4
签收了货物。。。4
接收到消息...内容:OrderReturnApplyEntity(id=1, orderId=null, skuId=null, orderSn=null, createTime=Mon May 20 16:31:49 CST 2024, memberUsername=null, returnAmount=null, returnName=哈哈哈8, returnPhone=null, status=null, handleTime=null, skuImg=null, skuName=null, skuBrand=null, skuAttrsVals=null, skuCount=null, skuPrice=null, skuRealPrice=null, reason=null, description述=null, descPics=null, handleNote=null, handleMan=null, receiveMan=null, receiveTime=null, receiveNote=null, receivePhone=null, companyAddress=null)
消息处理完成=》哈哈哈8
deliveryTag:5
没有签收货物...5