入门
原项目：
第一，拓展性差
第二，性能下降
由于我们采用了同步调用，调用者需要等待服务提供者执行完返回结果后，才能继续向下执行，也就是说每次远程调用，调用者都是阻塞等待状态。最终整个业务的响应时长就是每次远程调用的执行时长之和：
第三，级联失败
由于我们是基于OpenFeign调用交易服务、通知服务。当交易服务、通知服务出现故障时，整个事务都会回滚，交易失败。
基本介绍
类似于darabase，隔离不同的项目
快速入门
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>配置mq地址
spring:
rabbitmq:
host: 192.168.150.101 # 你的虚拟机IP
port: 5672 # 端口
virtual-host: /hmall # 虚拟主机
username: hmall # 用户名
password: 123 # 密码消息发送
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
public class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue() {
// 队列名称
String queueName = "simple.queue";
// 消息
String message = "hello, spring amqp!";
// 发送消息
rabbitTemplate.convertAndSend(queueName, message);
}
}消息接收
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class SpringRabbitListener {
// 利用RabbitListener来声明要监听的队列信息
// 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
// 可以看到方法体中接收的就是消息体的内容
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
System.out.println("spring 消费者接收到消息：【" + msg + "】");
}
}WorkQueues模型
Work queues，任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。
50个消息，一个消费者消费快，一个消费慢，但还是每个25
默认情况下消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。
能者多劳
spring:
rabbitmq:
listener:
simple:
prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息处理快的拿的多，慢的拿的少
Fanout交换机
交换机把消息发送给绑定过的所有队列
与直接向队列发消息不同的是，改为向交换机发消息
@Test
public void testFanoutExchange() {
// 交换机名称
String exchangeName = "hmall.fanout";
// 消息
String message = "hello, everyone!";
rabbitTemplate.convertAndSend(exchangeName, null, message);
}Direct交换机
@Test
public void testFanoutExchange() {
// 交换机名称
String exchangeName = "hmall.fanout";
// 消息
String message = "hello, everyone!";
rabbitTemplate.convertAndSend(exchangeName, blue, message);
}Topic交换机
声明队列和交换机
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 FanoutConfig {
//声明交换机
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("hmall.fanout");
}
//第1个队列
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queue1");
}
//绑定队列和交换机
@Bean
public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}
// 第2个队列
@Bean
public Queue fanoutQueue2(){
return new Queue("fanout.queue2");
}
//绑定队列和交换机
@Bean
public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
}
}
基于注解声明
基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),//消息队列的名称
exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),//交换机的名称
key = {"red", "blue"}//绑定关系
))
public void listenDirectQueue1(String msg){
System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
}
//运行后会自动生成消息队列和交换机
消息转换器
原始模式下，发送一个自定义对象消息格式非常不友好。
改造：
引入依赖
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-xml</artifactId>
<version>2.9.10</version>
</dependency>配置消息转换器，在publisher和consumer两个服务的启动类中添加一个Bean即可：
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}高级
发送者的可靠性
生产者重试机制
为了解决生产与MQ的连接中断，SpringAMQP提供的消息发送时的重试机制。即：当RabbitTemplate与MQ连接超时后，多次重试。
spring:
rabbitmq:
connection-timeout: 1s # 设置MQ的连接超时时间
template:
retry:
enabled: true # 开启超时重试机制
initial-interval: 1000ms # 失败后的初始等待时间
multiplier: 1 # 失败后下次的等待时长倍数，下次等待时长 = initial-interval * multiplier
max-attempts: 3 # 最大重试次数生产者确认机制
当消息投递到MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回ack的确认信息，代表投递成功
临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
持久消息投递到了MQ，并且入队完成持久化，返回ACK ，告知投递成功
其它情况都会返回NACK，告知投递失败
开启生产者确认
spring:
rabbitmq:
publisher-confirm-type: correlated # 开启publisher confirm机制，并设置confirm类型
publisher-returns: true # 开启publisher return机制这里publisher-confirm-type有三种模式可选：
none：关闭confirm机制simple：同步阻塞等待MQ的回执correlated：MQ异步回调返回回执
ConfirmCallback
临时消息投递到了MQ，并且入队成功，返回ACK，告知投递成功
持久消息投递到了MQ，并且入队完成持久化，返回ACK ，告知投递成功
ReturnCallback
当消息投递到MQ，但是路由失败时，通过Publisher Return返回异常信息，同时返回ack的确认信息，代表投递成功
MQ的可靠性
数据持久化
消息到达MQ以后，MQ会把数据保存到内存中，如果MQ宕机，会导致消息丢失
内存有限，如果消费者故障，会导致消息挤压，导致mq堵塞
为了保证数据的可靠性，必须配置数据持久化，包括：
交换机持久化（交换机持久化是指将交换机（包括名称、类型、路由规则、绑定关系等配置信息）持久化到磁盘存储中，确保即使服务重启，交换机的配置也能被恢复，无需重新创建和配置。）
队列持久化（队列中的数据通常存储在内存，队列持久化通过将队列中的消息从内存写入持久化存储）
消息持久化（如果是持久化的消息，消息会自动存入磁盘，如果是非持久化的消息，信息队列满了才会存入磁盘）
交换机持久化
LazyQueue
接收到消息后直接存入磁盘而非内存
消费者要消费消息时才会从磁盘中读取并加载到内存（也就是懒加载）
消费者的可靠性
消费者确认机制
通过下面的配置可以修改SpringAMQP的ACK处理方式：
由于消息回执的处理代码比较统一，因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式，有三种模式：
none：不处理。即消息投递给消费者后立刻ack，消息会立刻从MQ删除。非常不安全，不建议使用manual：手动模式。需要自己在业务代码中调用api，发送ack或reject，存在业务入侵，但更灵活auto：自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强，当业务正常执行时则自动返回ack. 当业务出现异常时，根据异常判断返回不同结果：如果是业务异常，会自动返回
nack；如果是消息处理或校验异常，自动返回
reject;
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: none # 不做处理失败重试机制
当消费者出现异常后，消息会不断requeue（重入队）到队列，再重新发送给消费者。
消费者失败重试机制：在消费者出现异常时利用本地重试，而不是无限制的requeue到mq队列。
spring:
rabbitmq:
listener:
simple:
retry:
enabled: true # 开启消费者失败重试
initial-interval: 1000ms # 初识的失败等待时长为1秒
multiplier: 1 # 失败的等待时长倍数，下次等待时长 = multiplier * last-interval
max-attempts: 3 # 最大重试次数
stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false失败处理策略
在之前的测试中，本地测试达到最大重试次数后，消息会被丢弃。这在某些对于消息可靠性要求较高的业务场景下，显然不太合适了。
因此Spring允许我们自定义重试次数耗尽后的消息处理策略，这个策略是由MessageRecovery接口来定义的，它有3个不同实现：
RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息。默认就是这种方式ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机
业务幂等性
在程序开发中，则是指同一个业务，执行一次或多次对业务状态的影响是一致的。例如：
根据id删除数据
查询数据
但数据的更新往往不是幂等的，如果重复执行可能造成不一样的后果。比如：
取消订单，恢复库存的业务。如果多次恢复就会出现库存重复增加的情况
退款业务。重复退款对商家而言会有经济损失。
因此，我们必须想办法保证消息处理的幂等性。这里给出两种方案：
唯一消息ID
每一条消息都生成一个唯一的id，与消息一起投递给消费者。
消费者接收到消息后处理自己的业务，业务处理成功后将消息ID保存到数据库
如果下次又收到相同消息，去数据库查询判断是否存在，存在则为重复消息放弃处理。
我们该如何给消息添加唯一ID呢？
其实很简单，SpringAMQP的MessageConverter自带了MessageID的功能，我们只要开启这个功能即可。
以Jackson的消息转换器为例：
@Bean
public MessageConverter messageConverter(){
// 1.定义消息转换器
Jackson2JsonMessageConverter jjmc = new Jackson2JsonMessageConverter();
// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
jjmc.setCreateMessageIds(true);
return jjmc;
}缺点：要查数据库，业务侵入性
业务状态判断
类似于乐观锁
兜底方案
既然MQ通知不一定发送到交易服务，那么交易服务就必须自己主动去查询支付状态。这样即便支付服务的MQ通知失败，我们依然能通过主动查询来保证订单状态的一致。
延迟消息
死信交换机
DelayExchange插件（延迟消息插件）
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "delay.queue", durable = "true"),
exchange = @Exchange(name = "delay.direct", delayed = "true"),
key = "delay"
))
public void listenDelayMessage(String msg){
log.info("接收到delay.queue的延迟消息：{}", msg);
}import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Slf4j
@Configuration
public class DelayExchangeConfig {
@Bean
public DirectExchange delayExchange(){
return ExchangeBuilder
.directExchange("delay.direct") // 指定交换机类型和名称
.delayed() // 设置delay的属性为true
.durable(true) // 持久化
.build();
}
@Bean
public Queue delayedQueue(){
return new Queue("delay.queue");
}
@Bean
public Binding delayQueueBinding(){
return BindingBuilder.bind(delayedQueue()).to(delayExchange()).with("delay");
}
}