SpringCloud系列教程：微服务的未来（二十五）-基于注解的声明队列交换机、消息转换器、业务改造

news/2025/2/23 16:30:26

前言

在现代分布式系统中，消息队列是实现服务解耦和异步处理的关键组件。Spring框架提供了强大的支持，使得与消息队列（如RabbitMQ、Kafka等）的集成变得更加便捷和灵活。本文将深入探讨如何利用Spring的注解驱动方式来配置和管理队列、交换机、消息转换器等组件，从而实现一个高效且可扩展的消息处理架构。

在本博客中，我们将重点介绍：

如何使用Spring的注解方式配置RabbitMQ的队列和交换机。
如何配置消息转换器（如Jackson2JsonMessageConverter）来处理不同格式的消息。
如何根据业务需求对现有代码进行改造，将消息队列引入到系统中，从而实现消息的异步处理与解耦。
通过这篇文章，您将了解如何使用Spring框架的注解配置简化消息队列的管理，同时提升系统的可扩展性和维护性。

基于注解的声明队列交换机

利用SpringAMQP声明DirectExchange并与队列绑定
需求如下:

  1. 在consumer服务中，声明队列direct.queue1和direct.queue2
  2. 在consumer服务中，声明交换机hmall.direct，将两个队列与其绑定
  3. 在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 {
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("hmall.direct")
    }
    @Bean
    public Queue directQueue1(){
        return new Queue("direct.queuue1");
    }
    @Bean
    public Binding directQueue1bindingRed( Queue directQueue1, DirectExchange directExchange ){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("red");

    }
    @Bean
    public Binding directQueue1bindingBlue( Queue directQueue1, DirectExchange directExchange ){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");

    }
    @Bean
    public Queue directQueue2(){
        return new Queue("direct.queuue2");
    }
    @Bean
    public Binding directQueue2bindingRed( Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("red");

    }
    @Bean
    public 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
  • 在控制台查看消息，总结你能发现的问题
// 准备消息
Map<String,0bject>msg = new HashMap<>();
msg.put("name","Jack");
msg.put("age"，21);

创建队列object.queue
在这里插入图片描述
测试代码如下：

	@Test
    public void TestSendObject(){
        Map<String, 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依赖:

<dependency>
	<groupId>com.fasterxml.jackson.core</groupId>
	<artifactId>jackson-databind</artifactId>
</dependency>

在publisher和consumer中都要配置Messageconverter:

@Bean
public MessageConverter messageConverter(){
	return new Jackson2JsonMessageConverter();
}

在这里插入图片描述
在这里插入图片描述
消费者代码：

	@RabbitListener(queues = "object.queue")
    public void listenObjectQueue(Map<String,Object> msg)throws Exception {
        log.info("消费者监听到pbject.queue的消息，["+msg+"]");
    }

在这里插入图片描述
运行结果如下：
在这里插入图片描述

业务改造

需求:改造余额支付功能，不再同步调用交易服务的0penFeign接口，而是采用异步MO通知交易服务更新订单状态。
在这里插入图片描述
在trade-service微服务消费者配置和pay-service微服务发送者都配置MQ依赖

	<!--消息发送-->
  <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>

在trade-service微服务和pay-service微服务添加上RabbitMQ配置信息

spring:
  rabbitmq:
    host: 192.168.244.136
    port: 5672
    virtual-host: /hmall
    username: hmall
    password: 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 {
    @Bean
    public 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 ServiceImpl<PayOrderMapper, PayOrder> implements IPayOrderService {
	private final RabbitTemplate rabbitTemplate;
	...
	@Override
	@Transactional
	public 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在消息队列方面的强大功能，并为您的业务应用提供参考。随着系统复杂度的增加，合理的使用消息队列将成为构建高可用、高性能系统的关键之一。


http://www.niftyadmin.cn/n/5863582.html

相关文章

WordPress Elementor提示错误无法保存500的解决指南

WordPress Elementor提示错误无法保存500的解决指南

500内部服务器错误是一种常见的服务器错误&#xff0c;通常由网站的服务器环境引起。这种错误可能导致网站无法正常访问&#xff0c;影响用户体验。本文将探讨500错误的常见原因&#xff0c;并提供解决方案&#xff0c;特别针对使用Elementor构建的WordPress网站。 500错误的常…
阅读更多...
【WebGL】fbo双pass案例

【WebGL】fbo双pass案例

双pass渲染案例&#xff08;离线渲染一个三角面&#xff0c;然后渲染到一个占满屏幕的矩阵上&#xff09; 离线渲染如何需要开启深度测试的话&#xff0c;需要额外操作&#xff0c;这里不展开 <!DOCTYPE html> <html lang"en"><head><meta ch…
阅读更多...
区块链讲解

区块链讲解

1.基本概念&#xff1a;区块链是一种按照时间顺序将数据区块以顺序相连的方式组合成的链式数据结构&#xff0c;并以密码学方式保证的不可篡改和不可伪造的分布式账本。简单来说&#xff0c;它就像是一个由众多节点共同维护的账本&#xff0c;每个节点都有这个账本的完整副本&a…
阅读更多...
PTA：有序顺序表的插入

PTA：有序顺序表的插入

请设计一个算法&#xff0c;在有序顺序表L中插入元素x&#xff0c;使得表依然有序&#xff0c;并输出新增元素后的表数据。 例如&#xff1a; L的元素 1 3 5 7 插入新元素 4 输出 1 3 4 5 7 其中&#xff0c;L的长度不超过1000&#xff0c;当中的元素为非递减排序。 输入格式…
阅读更多...
Trae+Qt+MSVC环境配置

Trae+Qt+MSVC环境配置

Trae Trae是字节跳动基于VSCode推出的AI集成开发环境&#xff08;IDE&#xff09;&#xff0c;是一款专为中文开发者深度定制的智能编程工具。其目标是通过AI技术实现从“Copilot”到“Autopilot”的编程模式演进。 类似这样的IDE比如Windsurf、Cursor&#xff0c;都是基于VS…
阅读更多...
pycharm中配置PyQt6详细教程

pycharm中配置PyQt6详细教程

PyQt6 是 Qt 框架的 Python 绑定库，基于 Qt 6 开发，专为创建跨平台图形用户界面（GUI）应用程序设计。 本章教程，主要记录在pycharm中配置使用PyQt6的流程。 一、安装基础环境 在此之前，你需要提前安装好Python解释器，推荐使用anaconda创建虚拟环境。 conda create -n pyt…
阅读更多...
从零开始开发纯血鸿蒙应用之网页浏览

从零开始开发纯血鸿蒙应用之网页浏览

从零开始开发纯血鸿蒙应用 〇、前言一、优化菜单交互1、BuilderFunction.ets2、改造 PageTitleBar 二、网址打开1、方式选择1、使用浏览器打开2、内部打开2.1、声明权限2.2、封装 WebViewPage2.2.1、组件字段2.2.2、aboutToAppear2.2.3、onBackPress2.2.4、标题栏2.2.4、网页内…
阅读更多...
IWPA_CEC2005

IWPA_CEC2005

3种策略改进WPA狼群算法&#xff0c;独家原创&#xff0c;效果非常好&#xff0c;可以直接拿来写毕设或者论文 算法设计、毕业设计、期刊专利&#xff01;感兴趣可以联系我。 &#x1f3c6;代码获取方式1&#xff1a; 私信博主 &#x1f3c6;代码获取方式2 利用同等价值的matl…
阅读更多...
最新文章

Copyright @ 2022~2023