云端网络网站,北京文化馆设计公司怎么看,使用joomla的网站,廊坊百度seo公司前言
在现代分布式系统中#xff0c;消息队列是实现服务解耦和异步处理的关键组件。Spring框架提供了强大的支持#xff0c;使得与消息队列#xff08;如RabbitMQ、Kafka等#xff09;的集成变得更加便捷和灵活。本文将深入探讨如何利用Spring的注解驱动方式来配置和管理队…前言
在现代分布式系统中消息队列是实现服务解耦和异步处理的关键组件。Spring框架提供了强大的支持使得与消息队列如RabbitMQ、Kafka等的集成变得更加便捷和灵活。本文将深入探讨如何利用Spring的注解驱动方式来配置和管理队列、交换机、消息转换器等组件从而实现一个高效且可扩展的消息处理架构。
在本博客中我们将重点介绍
如何使用Spring的注解方式配置RabbitMQ的队列和交换机。 如何配置消息转换器如Jackson2JsonMessageConverter来处理不同格式的消息。 如何根据业务需求对现有代码进行改造将消息队列引入到系统中从而实现消息的异步处理与解耦。 通过这篇文章您将了解如何使用Spring框架的注解配置简化消息队列的管理同时提升系统的可扩展性和维护性。 基于注解的声明队列交换机
利用SpringAMQP声明DirectExchange并与队列绑定 需求如下:
在consumer服务中声明队列direct.queue1和direct.queue2在consumer服务中声明交换机hmall.direct将两个队列与其绑定在consumer服务中编写两个消费者方法分别监听direct.queue1和direct.queue2 基于Bean声明队列和交换机代码如下
package com.itheima.consumer.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class DirectConfiguration {Beanpublic DirectExchange directExchange(){return new DirectExchange(hmall.direct)}Beanpublic Queue directQueue1(){return new Queue(direct.queuue1);}Beanpublic Binding directQueue1bindingRed( Queue directQueue1, DirectExchange directExchange ){return BindingBuilder.bind(directQueue1).to(directExchange).with(red);}Beanpublic Binding directQueue1bindingBlue( Queue directQueue1, DirectExchange directExchange ){return BindingBuilder.bind(directQueue1).to(directExchange).with(blue);}Beanpublic Queue directQueue2(){return new Queue(direct.queuue2);}Beanpublic Binding directQueue2bindingRed( Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with(red);}Beanpublic Binding directQueue2bindingYellow( Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with(yellow);}
}SpringAMOP还提供了基于RabbitListener注解来声明队列和交换机的方式
RabbitListener(bindings QueueBinding(value Queue(name direct.queue1),exchange Exchange(name itcast.direct,type ExchangeTypes.DIRECT),key {red,blue}
))
public void listenDirectQueuel(string msg){System.out.println(消费者1接收到Direct消息:【msg】);
}接收者代码如下 RabbitListener(bindings QueueBinding(value Queue(name direct.queue1,durable true),exchange Exchange(name hmall.direct,type ExchangeTypes.DIRECT),key {red,blue}))public void listenDirectQueue1(String message)throws Exception {log.info(消费者1监听到direct.queue2的消息[message]);}RabbitListener(bindings QueueBinding(value Queue(name direct.queue2,durable true),exchange Exchange(name hmall.direct,type ExchangeTypes.DIRECT),key {red,yellow}))消息转换器
消息转换器 需求:测试利用SpringAMQP发送对象类型的消息
声明一个队列名为object.queue编写单元测试向队列中直接发送一条消息消息类型为Map在控制台查看消息总结你能发现的问题
// 准备消息
MapString,0bjectmsg new HashMap();
msg.put(name,Jack);
msg.put(age21);创建队列object.queue 测试代码如下 Testpublic void TestSendObject(){MapString, Object msg new HashMap();msg.put(name, Jack);msg.put(age, 18);//3.发送消息 参数分别是交换机名称、RoutingKey暂时为空、消息rabbitTemplate.convertAndSend(object.queue,msg);}在控制台上找到object.queue中得到消息 Spring的对消息对象的处理是由org.springframework.amgp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter基于IDK的ObjectOutputStream完成序列化。存在下列问题:
JDK的序列化有安全风险JDK序列化的消息太大JDK序列化的消息可读性差
建议采用JSON序列化代替默认的JDK序列化要做两件事情 在publisher和consumer中都要引入jackson依赖:
dependencygroupIdcom.fasterxml.jackson.core/groupIdartifactIdjackson-databind/artifactId
/dependency在publisher和consumer中都要配置Messageconverter:
Bean
public MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();
}消费者代码 RabbitListener(queues object.queue)public void listenObjectQueue(MapString,Object msg)throws Exception {log.info(消费者监听到pbject.queue的消息[msg]);}运行结果如下
业务改造
需求:改造余额支付功能不再同步调用交易服务的0penFeign接口而是采用异步MO通知交易服务更新订单状态。 在trade-service微服务消费者配置和pay-service微服务发送者都配置MQ依赖 !--消息发送--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency在trade-service微服务和pay-service微服务添加上RabbitMQ配置信息
spring:rabbitmq:host: 192.168.244.136port: 5672virtual-host: /hmallusername: hmallpassword: 1234因为消费者和发送者都需要消息转换器故直接将代码写到hm-common服务中在config包中创建MqConfig类
package com.hmall.common.config;import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class MqConfig {Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}
}同时trade-service微服务和pay-service微服务是无法自动扫描到该类采用SpringBoot自动装配的原理在resource文件夹下的META-INF文件夹下的spring.factories文件中添加类路径
在接收者trade-service微服务中创建PayStatusListener
package com.hmall.trade.listener;import com.hmall.trade.service.IOrderService;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
RequiredArgsConstructor
public class PayStatusListener {private final IOrderService orderService;RabbitListener(bindings QueueBinding(value Queue(trade.pay.success.queue),exchange Exchange(value pay.direct),key pay.success))public void ListenPaySuccess(Long orderId) {orderService.markOrderPaySuccess(orderId);}}修改pay-service服务下的com.hmall.pay.service.impl.PayOrderServiceImpl类中的tryPayOrderByBalance方法
Service
RequiredArgsConstructor
Slf4j
public class PayOrderServiceImpl extends ServiceImplPayOrderMapper, PayOrder implements IPayOrderService {private final RabbitTemplate rabbitTemplate;...OverrideTransactionalpublic void tryPayOrderByBalance(PayOrderDTO payOrderDTO) {// 1.查询支付单PayOrder po getById(payOrderDTO.getId());// 2.判断状态if(!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())){// 订单不是未支付状态异常throw new BizIllegalException(交易已支付或关闭);}// 3.尝试扣减余额userClient.deductMoney(payOrderDTO.getPw(), po.getAmount());// 4.修改支付单状态boolean success markPayOrderSuccess(payOrderDTO.getId(), LocalDateTime.now());if (!success) {throw new BizIllegalException(交易已支付或关闭);}// 5.修改订单状态// tradeClient.markOrderPaySuccess(po.getBizOrderNo());try {rabbitTemplate.convertAndSend(pay.direct, pay.success, po.getBizOrderNo());} catch (Exception e) {log.error(支付成功的消息发送失败支付单id{} 交易单id{}, po.getId(), po.getBizOrderNo(), e);}}
}总结
本文介绍了基于Spring框架的注解方式来配置消息队列、交换机以及消息转换器的实现方法。通过注解配置开发者可以更轻松地创建和管理RabbitMQ等消息队列的组件而无需过多的 XML 配置或繁琐的手动配置。具体来说我们探讨了如何
使用 RabbitListener 和 EnableRabbit 注解配置消息监听器和消息队列。 配置消息转换器特别是如何通过 Jackson2JsonMessageConverter 将消息转换为JSON格式从而实现数据的序列化与反序列化。 结合业务需求讲解如何对现有系统进行改造集成消息队列实现异步处理和服务解耦。 通过这些配置和改造系统的消息处理能力得到了增强性能和可扩展性也得到了显著提升。消息队列的使用不仅能够减少服务之间的紧耦合还能够通过异步方式提高系统的响应速度和吞吐量。
希望本博客能够帮助您理解Spring在消息队列方面的强大功能并为您的业务应用提供参考。随着系统复杂度的增加合理的使用消息队列将成为构建高可用、高性能系统的关键之一。
