Bus消息总线 一、概述在微服务架构的系统中，通常会使用轻量级的消息代理来构建一个共用的消息主题，并让系统中所有微服务实例都连接上来。由于该主题中产生的消息会被所有实例监听和消费，所以称它为消息总线.二、基本原理ConfigClient实例都监听MQ中同一个topic（默认Spring Cloud Bus），当一个服务刷新数据的时候，它会把这个消息放到topic中，这样其他监听同一个topic的服务就能收到通知，然后去更新自身的配置。三、流程四、手动定点刷新curl -X POSt "http://localhost:3344/actuator/bus-refresh/config-client:3355"五、通过分组解决消息重复消费的问题图片来源：尚硅谷-周阳 Spring CLoud Alibaba注：通过分组解决消息重复消费的同时，也实现了消息持久化

Gateway 一、概述Spring Cloud GateWay作为Spring Cloud生态系统中的网关，目的是代替zuul。其基于WebFlux框架，底层使用高性能的Reactor模式的通信框架Netty。提供统一的路由方式且基于Filter链的方式提供网关的基本功能，如：安全、监控/指标、限流等。二、特性1、基于Spring Framework 5，Project Reactor，SpringBoot 2.0进行构建2、动态路由：能够匹配任何请求属性3、可以对路由指定Predicates（断言）和Filter（过滤器）4、集成Hystrix的断路器功能5、集成SpringCloud服务发现功能6、请求限流7、路径重写……三、三个概念1、Route（路由）路由是构建网关的基本模块，由ID、目标URI和一系列的断言、过滤器组成，如果断言为true则匹配该路由2、Predicates（断言）匹配Http请求中的所有内容（请求头或请求参数等）3、Filter（过滤）指Spring框架中GatewayFilter的实例，使用过滤时，可以在请求被路由之前或之后对请求做出修改四、工作流程客户端向Spring Cloud GateWay发出请求，然后在Gateway Handler Mapping中找到与请求相匹配的路由，将其发送到Gateway Web Handler；Handler再通过指定的过滤器链将请求发送到实际的执行业务逻辑，然后返回（过滤器可能会在发送代理请求之前（pre）或之后（post）执行业务逻辑）；Filter在“pre”类型的过滤器可以做参数校验、权限校验、流量监控、日志输出、协议转换等；在“post”类型的过滤器可以做内容响应、响应头的修改，日志的输出、流量监控等功能Predicates cookies路由规则HeaderHostMethodQuery图片来源：尚硅谷-周阳 Spring CLoud Alibaba

分布式事务 一、概述指事务的操作位于不同的节点上，需要保证事务的ACID原则，如：订单与库存不在同一个节点上，但需要保证两个事务的一致性（下单—减少库存，取消订单—库存恢复）二、实现分布式事务的方式1、两段提交（2PC）需要数据库厂商的支持，java组件有atomikos等两阶段提交（Two-phase Commit），通过引入协调者（Coordinator）来协调参与者的行为，并最终决定这些参与者是否要真正执行事务准备阶段协调者询问参与者事务是否执行成功，参与者返执行回结果提交阶段如果事务在每个参与者上都执行成功，事务协调者发送通知让参与者提交事务；否则反之，即回滚事务注：参与者在准备阶段执行成功，但未提交事务，只有在提交阶段收到协调者的消息，才会执行提交或回滚存在的问题1、同步阻塞，所有参与者在等待其他参与者相应的时候都处于同步阻塞状态，无法进行其他操作2、单点问题，协调者在2PC中起到非常大的作用，产生故障时影响也非常大3、数据不一致，在阶段二，如果协调者只发送了部分Commit消息，此时如果产生网络异常，则只有接收到Commit消息的参与者提交了事务，使得数据不一致4、任意一个节点失败，就会导致整个事务失败，容错机制不完善2、补偿事务（TCC）针对每个操作，都注册一个与其对应的确认和补偿（撤销）操作，分为三个阶段：Try 对业务系统做检测及资源预留Confirm 对业务系统做确认提交，Try阶段执行成功并开始执行Confirm阶段时，默认Confirm阶段是不会出错的（只要Try成功，Confirm一定成功）Cancel 在业务执行错误，需要回滚时执行业务的取消，释放预留资源优缺点实现流程相较于2PC简单，但数据一致性比2PC要更差一些TCC属于应用层的一种补偿方式，所以在实现业务的时候需要编写很多的补偿代码，在一些业务场景中，业务处理不太友好3、本地消息表（异步确保）本地消息表与业务数据表处于同一个数据库中，这样就能利用本地事务来保证在对这两个表的操作满足事务特性，并且使用消息队列保证最终一致性。在分布式事务操作的一方完成写业务数据的操作之后向本地消息表发送一个消息，本地事务能保证这个消息一定会被写入本地消息表中之后将本地消息表的消息转发到Kafka等消息队列中，如果转发成功则将消息从本地消息表中删除，否则继续重新转发在分布式事务操作的另一方从消息队列中读取一个消息，并执行消息中的操作优缺点1、避免了分布式事务，实现数据一致性2、消息表会耦合到业务系统中，耦合度太高4、MQ事务消息一些第三方的MQ是支持事务消息的，如RocketMQ，它们支持事务的方式类似于二阶段提交，流程大致为：第一阶段：Prepared消息，拿到消息的地址；第二阶段：执行本地事务；第三阶段：通过第一阶段拿到的地址去访问消息，并修改状态也就是说在业务方法内要让消息队列提交两次请求，一次发送消息，一次确认消息。如果确认消息发送失败了，RabbitMQ会定期扫描消息集群中的事务消息，这时候发现了Prepared消息，它会向消息发送者确认，所以生产方需要实现一个check接口，RabbitMQ会根据发送端设置的策略来决定回滚还是继续发送确认消息，这样就保证了消息发送与本地事务同时成功或同时失败优缺点1、实现最终一致性，不需要依赖本地数据库事务2、实现难度大，主流MQ不支持

RabbitMQ集群 以单机多节点搭建为例1、启动第一个节点 rabbit-1RABBITMQ_NODE_PORT=5672 RABBITMQ_NODENAME=rabbit-1 rabbitmq-server start &2、启动第二个节点 rabbit-2注：web管理插件的端口号需要另外指定，否则会提示默认的 5672 端口占用RABBITMQ_NODE_PORT=5673 RABBIT_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME=rabbit-2 rabbitmq-server start &3、查看服务ps aux|grep rabbitmq4、设置rabbit-1为主节点# 停止应用 rabbitmqctl -n rabbit-1 stop_app # 重置，清除节点上的历史信息 rabbitmqctl -n rabbit-1 reset # 启动 rabbitmqctl -n rabbit-1 start_app5、设置从节点# 停止应用 rabbitmqctl -n rabbit-2 stop_app # 重置，清除节点上的历史信息 rabbitmqctl -n rabbit-2 reset # 将rabbit-2加入到rabbit-1主节点 # 'Server-node' 主节点名，实际操作时需更换为主机名 rabbitmqctl -n rabbit-2 join_cluster rabbit-1@'Server-node' # 启动 rabbitmqctl -n rabbit-2 start_app6、验证集群状态rabbitmqctl cluster_status -n rabbit-17、web管理给node节点设置用户名和密码rabbitmqctl -n rabbit-1 add_user admin 123456 rabbitmqctl -n rabbit-1 set_user_tags admin administrator rabbitmqctl -n rabbit-1 set_permissions -p / admin ".*" ".*" ".*" # 其他节点以此类推补充如果采用多机部署集群的方式，需读取其中一个节点的cookie，并复制到其他节点（节点之间通过cookie确认相互间是否能进行通信）cookie存放路径：/var/lib/rabbitmq/.erlang.cookie

内存和磁盘监控 官方文档：https://www.rabbitmq.com/configure.html1、内存当出现警告的时候，可以通过修改或调整配置一般建议大小为物理机内存的40%~70%之间命令行rabbitmqctl set_vm_memory_high_watermark <fraction> # 二者任选其一 rabbitmqctl set_vm_memory_high_watermark absolute 50MB配置文件 rabbitmq.conf# /etc/rabbitmq/rabbitmq.conf # 默认 vm_memory_high_watermark.relative = 0.4 # 取值在0.4~0.7之间，不建议超过0.7 # 设置固定值 vm_memory_high_watermark.absolute = 1GB 测试手动修改内存为固定的50MBrabbitmq的运行内存超过了50MB，监控面板内存报红消息队列随之被挂起，不再提供服务测试之后手动恢复rabbitmqctl set_vm_memory_high_watermark 0.42、内存换页在某个Broker节点及内存阻塞生产者之前，它会将队列中的消息换页到磁盘中，以释放内存空间，持久化和非持久化的消息都会写入磁盘中（持久化消息本身就有一个副本在磁盘中，所以在转移过程中持久化的消息会率先从内存中转移）默认情况下，内存达到阈值的50%就会进行换页处理，即：默认阈值为0.4的情况下，当内存超过0.4*0.5=0.2时就会进行换页处理，因为当出现内存警告，服务被挂起时才来进行换页已经没有意义。vm_memory_high_watermark.relative = 0.4 vm_memory_high_watermark_paging_ratio = 0.7 # 设置小于1的值(一般不超过0.7)3、磁盘当磁盘剩余空间小于阈值时，RabbitMQ同样会产生预警，将服务挂起，同理内存预警命令行rabbitmqctl set_disk_free_limit <disk_limit> rabbitmqctl set_disk_free_limit memory_limit <faction> # disk_limit 固定值，单位KB MB GB # faction 相对阈值，建议范围在1.0~2.0之间，此处须于内存的阈值区分开来配置文件disk_free_limit.relative = 3.0 disk_free_limit.absolute = 50mb

死信队列 修改已创建队列的参数，重启服务报错，原因：修改后的参数随服务的重启并不会进行覆盖1、概述Dead-Letter-Exchange（DLX）死信队列，也称作死信交换机，死信邮箱。当一个消息在队列中变成死信（dead message）之后，它会被重新发送到一个交换机中，这个交换机就是DLX，绑定DLX的队列就是死信队列。消息变为死信的原因有：消息被拒绝消息过期队列达到最大长度DLX也是一个正常的交换机，和一般的交换机没有区别，它能在任何的队列上被指定，也就是设置某一个队列的属性。当这个队列中存在死信时，RabbitMQ就会可将这个消息重新发送到设置的DLX上，进而被路由到另一个队列，即死信队列。使用：在定义队列的时候设置队列参数 x-dead-letter-exchange指定交换机即可 图片来源：学相伴 - 飞哥 - RabbitMQ测试：创建队列@Configuration public class DeadRabbitMQConfig { //1、声明注册direct模式的交换机 @Bean public DirectExchange DeadDirectExchange() { return new DirectExchange("dead_direct_exchange", true, false); } //2、声明队列过期时间 @Bean public Queue deadQueue() { return new Queue("dead.direct.queue", true); } @Bean public Binding deadBind() { return BindingBuilder.bind(deadQueue()).to(DeadDirectExchange()).with("dead"); } }绑定 TTL队列 和 死信队列@Configuration public class TTLRabbitMQConfig { //1、声明注册direct模式的交换机 @Bean public DirectExchange ttlDirectExchange() { return new DirectExchange("ttl_direct_exchange", true, false); } //2、声明队列过期时间 @Bean public Queue directTTLQueue() { Map<String, Object> args = new HashMap<>(); //单位 ms args.put("x-message-ttl", 5000); //绑定死信交换机 args.put("x-dead-letter-exchange", "dead_direct_exchange"); //fanout模式不需要指定路由key args.put("x-dead-letter-routing-key", "dead"); return new Queue("ttl.direct.queue", true, false, false, args); } @Bean public Queue directTTLMessageQueue() { return new Queue("ttl.message.direct.queue", true); } //过期队列 @Bean public Binding ttlQueue(){ return BindingBuilder.bind(directTTLQueue()).to(ttlDirectExchange()).with("ttl"); } //消息含有过期时间，队列没有设置过期时间 @Bean public Binding ttlMessageQueue(){ return BindingBuilder.bind(directTTLMessageQueue()).to(ttlDirectExchange()).with("ttlMessage"); } }生产者@Service public class OrderService { @Autowired private RabbitTemplate rabbitTemplate; // ttl 单独设置过期时间 public void makeOrderTtlMessage(String userId, String productId, int num) { //1、根据商品id查询库存 //2、保存订单 String orderId = UUID.randomUUID().toString().replaceAll("-", ""); System.out.println("创建订单成功：" + orderId); //3、通过 MQ 来完成消息的分发 String exchangeName = "ttl_direct_exchange"; String routingKry = "ttlMessage"; //设置消息过期时间 MessagePostProcessor message = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("5000"); message.getMessageProperties().setContentEncoding("UTF-8"); return message; } }; /** * 过期队列：消息存入死信队列 * 消息设置过期时间：过期后，消息直接移除 */ //交换机 路由key/queue队列名 消息内容 rabbitTemplate.convertAndSend(exchangeName, routingKry, orderId, message); } }消费者 暂不设置，目的是让消息过期 路由 到死信队列中消息未过期消息过期后

过期时间 TTL 1、概述过期时间TTL指可以对消息设置预期的时间，在这个时间内都可以被消费者接收获取；过后将被自动删除，RabbitMQ设置TTL的两种方式：通过队列属性设置，队列中所有的消息都有相同的过期时间对消息进行单独的设置，每条消息的TTL不同如果同时使用以上两种方式，则以较小的那个时间为准。消息在队列的生存时间一旦超过设置的TTL值，将被投递到死信队列，成为dead message，消费者将无法再收到该消息。过期队列：消息存入死信队列 消息设置过期时间：过期后，消息直接移除创建队列@Configuration public class TTLRabbitMQConfig { //1、声明注册direct模式的交换机 @Bean public DirectExchange ttlDirectExchange() { return new DirectExchange("ttl_direct_exchange", true, false); } //2、声明队列过期时间 @Bean public Queue directTTLQueue() { Map<String, Object> args = new HashMap<>(); //单位 ms args.put("x-message-ttl", 5000); return new Queue("ttl.direct.queue", true, false, false, args); } @Bean public Queue directTTLMessageQueue() { return new Queue("ttl.message.direct.queue", true); } //过期队列 @Bean public Binding ttlQueue(){ return BindingBuilder.bind(directTTLQueue()).to(ttlDirectExchange()).with("ttl"); } //消息含有过期时间，队列没有设置过期时间 @Bean public Binding ttlMessageQueue(){ return BindingBuilder.bind(directTTLMessageQueue()).to(ttlDirectExchange()).with("ttlMessage"); } }生产者@Service public class OrderService { @Autowired private RabbitTemplate rabbitTemplate; // ttl public void makeOrderTtl(String userId, String productId, int num) { //1、根据商品id查询库存 //2、保存订单 String orderId = UUID.randomUUID().toString().replaceAll("-", ""); System.out.println("创建订单成功：" + orderId); //3、通过 MQ 来完成消息的分发 String exchangeName = "ttl_direct_exchange"; String routingKry = "ttl"; //交换机 路由key/queue队列名 消息内容 rabbitTemplate.convertAndSend(exchangeName, routingKry, orderId); } // ttl 单独设置过期时间 public void makeOrderTtlMessage(String userId, String productId, int num) { //1、根据商品id查询库存 //2、保存订单 String orderId = UUID.randomUUID().toString().replaceAll("-", ""); System.out.println("创建订单成功：" + orderId); //3、通过 MQ 来完成消息的分发 String exchangeName = "ttl_direct_exchange"; String routingKry = "ttlMessage"; //设置消息过期时间 MessagePostProcessor message = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { message.getMessageProperties().setExpiration("5000"); message.getMessageProperties().setContentEncoding("UTF-8"); return message; } }; //交换机 路由key/queue队列名 消息内容 rabbitTemplate.convertAndSend(exchangeName, routingKry, orderId, message); } }消费者//消费者代码同理fanout模式的消费者

SpringBoot整合RabbitMQ 1、Fanout模式创建队列@Configuration public class FanoutRabbitMQConfig { //1、声明注册fanout模式的交换机 @Bean public FanoutExchange fanoutExchange() { return new FanoutExchange("fanout_order_exchange", true, false); } //2、声明队列 msg.fanout.queue,email.fanout.queue,sms.fanout.queue @Bean public Queue fanoutMsgQueue() { return new Queue("msg.fanout.queue", true); } @Bean public Queue fanoutEmailQueue() { return new Queue("email.fanout.queue", true); } @Bean public Queue fanoutSmsQueue() { return new Queue("sms.fanout.queue", true); } //3、完成绑定 @Bean public Binding msgBinding() { return BindingBuilder.bind(fanoutMsgQueue()).to(fanoutExchange()); } @Bean public Binding emailBinding() { return BindingBuilder.bind(fanoutEmailQueue()).to(fanoutExchange()); } @Bean public Binding smsBinding() { return BindingBuilder.bind(fanoutSmsQueue()).to(fanoutExchange()); } } 生产者@Service public class OrderService { @Autowired private RabbitTemplate rabbitTemplate; //fanout模式 模拟用户下单 public void makeOrderFanout(String userId, String productId, int num) { //1、根据商品id查询库存 //2、保存订单 String orderId = UUID.randomUUID().toString().replaceAll("-", ""); System.out.println("创建订单成功：" + orderId); //3、通过 MQ 来完成消息的分发 String exchangeName = "fanout_order_exchange"; String routingKey = ""; //交换机 路由key/queue队列名 消息内容 rabbitTemplate.convertAndSend(exchangeName, routingKey, orderId); } }消费者@Service @RabbitListener(queues = {"email.fanout.queue"}) public class FanoutEmailConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("email.fanout 接收到订单信息：" + message); } }@Service @RabbitListener(queues = {"msg.fanout.queue"}) public class FanoutMsgConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("msg.fanout 接收到订单信息：" + message); } } @Service @RabbitListener(queues = {"sms.fanout.queue"}) public class FanoutSmsConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("sms.fanout 接收到订单信息:" + message); } } 2、Direct模式创建队列@Configuration public class DirectRabbitMQConfig { //1、声明注册direct模式的交换机 @Bean public DirectExchange directExchange() { return new DirectExchange("direct_order_exchange", true, false); } //2、声明队列 msg.direct.queue,email.direct.queue,sms.direct.queue @Bean public Queue directMsgQueue() { return new Queue("msg.direct.queue", true); } @Bean public Queue directEmailQueue() { return new Queue("email.direct.queue", true); } @Bean public Queue directSmsQueue() { return new Queue("sms.direct.queue", true); } //3、完成绑定 @Bean public Binding msgBinding() { return BindingBuilder.bind(directMsgQueue()).to(directExchange()).with("smg"); } @Bean public Binding emailBinding() { return BindingBuilder.bind(directEmailQueue()).to(directExchange()).with("email"); } @Bean public Binding smsBinding() { return BindingBuilder.bind(directSmsQueue()).to(directExchange()).with("sms"); } }生产者@Service public class OrderService { @Autowired private RabbitTemplate rabbitTemplate; //direct模式 public void makeOrderDirect(String userId, String productId, int num) { //1、根据商品id查询库存 //2、保存订单 String orderId = UUID.randomUUID().toString().replaceAll("-", ""); System.out.println("创建订单成功：" + orderId); //3、通过 MQ 来完成消息的分发 String exchangeName = "direct_order_exchange"; //交换机 路由key/queue队列名 消息内容 rabbitTemplate.convertAndSend(exchangeName, "sms", orderId); rabbitTemplate.convertAndSend(exchangeName, "email", orderId); } }消费者@Service @RabbitListener(queues = {"email.direct.queue"}) public class DirectEmailConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("email.direct接收到订单信息:" + message); } } @Service @RabbitListener(queues = {"msg.direct.queue"}) public class DirectMsgConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("msg.direct 接收到订单信息:" + message); } } @Service @RabbitListener(queues = {"sms.direct.queue"}) public class DirectSmsConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("sms.direct 接收到订单信息:" + message); } } 3、Topic模式生产者@Service public class OrderService { @Autowired private RabbitTemplate rabbitTemplate; //topic模式 public void makeOrderTopic(String userId, String productId, int num) { //1、根据商品id查询库存 //2、保存订单 String orderId = UUID.randomUUID().toString().replaceAll("-", ""); System.out.println("创建订单成功：" + orderId); //3、通过 MQ 来完成消息的分发 String exchangeName = "topic_order_exchange"; String routingKry = "com.msg"; //交换机 路由key/queue队列名 消息内容 // #.msg.# msg // *.email.# email // com.# sms rabbitTemplate.convertAndSend(exchangeName, routingKry, orderId); } }创建队列（通过注解）、消费者@Service @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "email.topic.queue", durable = "true", autoDelete = "false"), exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC), key = "*.email.#" )) public class TopicEmailConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("email.topic 接收到订单信息：" + message); } }@Service @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "msg.topic.queue", durable = "true", autoDelete = "false"), exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC), key = "#.msg.#" )) public class TopicMsgConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("msg.topic 接收到订单信息：" + message); } } @Service @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "sms.topic.queue", durable = "true", autoDelete = "false"), exchange = @Exchange(value = "topic_order_exchange", type = ExchangeTypes.TOPIC), key = "com.#" )) public class TopicSmsConsumer { @RabbitHandler public void receiveMessage(String message){ System.out.println("sms.topic 接收到订单信息:" + message); } }