微服务项目使用RabbitMQ
很久未用RabbitMQ了,根据网上的Demo,大多数是一个服务包含所有的生产者和消费者和配置,当自己去搭建服务的时候,还需要一些思考各种包的划分.无法无脑CV大法,所以,下文,我根据实际项目抽离出一个比较完整的小Demo演示微服务项目使用RabbitMQ.注意:这个小Demo并没有做消息的可靠性相关操作!
公共服务
一般微服务有一个公共服务,用于存放一些配置类或者常量类等,这里我以service_util举例,这个服务下,应该会有以下四个包,如下:
├─config -- 配置类的包 │ RabbitMQConfig.java │ ├─MQConstant -- 常量的包 │ RabbitMQConstant.java │ ├─service -- 封装的服务的包 │ RabbitService.java │ └─vo -- Vo的包 OrderVo.java RabbitMQConfig
这个类下,我们可以配置基本的RabbitMQ的消息的序列化,在消息的传输过程中,对象和Json的互相转化,详细代码如下:
package com.leixin.mq.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; /** * @Classname RabbitMQConfig * @Description TODO * @Date 2023-08-07 21:25 * @Created by LeiXin */ /* 这个配置类的作用是将一个消息转换器添加到RabbitMQ中， 这样在消息发送和接收时， 就能够自动地将Java对象转换为JSON格式的消息， 以及将JSON格式的消息转换回Java对象。 */ @Configuration public class RabbitMQConfig { // 声明一个@Bean注解，表示这是一个Spring Bean，会由Spring容器进行管理和实例化 @Bean public MessageConverter getMessageConverter() { // 创建并返回一个Jackson2JsonMessageConverter实例作为消息转换器 // 这个转换器用于在消息发送和接收之间进行JSON格式的转换 return new Jackson2JsonMessageConverter(); } } RabbitMQConstant
这个包用于指定消息队列的常量
package com.leixin.mq.MQConstant; /** * @Classname RabbitMQConstant * @Description TODO * @Date 2023-08-07 23:23 * @Created by LeiXin */ public class RabbitMQConstant { /** * 预约下单 */ public static final String EXCHANGE_DIRECT_ORDER = "exchange.direct.order"; public static final String ROUTING_ORDER = "order"; //队列 public static final String QUEUE_ORDER = "queue.order"; /** * 短信 */ public static final String EXCHANGE_DIRECT_SMS = "exchange.direct.msm"; public static final String ROUTING_SMS_ITEM = "msm.item"; //队列 public static final String QUEUE_MSM_SMS = "queue.msm.item"; } RabbitService
这个包用于封装消息的发送方式,之后发送消息只要使用RabbitService.
package com.leixin.mq.service; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; /** * @Classname RabbitService * @Description TODO * @Date 2023-08-07 23:07 * @Created by LeiXin */ @Service public class RabbitService { @Autowired private RabbitTemplate rabbitTemplate; /** * 发送消息 * @param exchange * @param routingKey * @param message * @return */ public boolean sendMessage(String exchange,String routingKey,Object message){ rabbitTemplate.convertAndSend(exchange,routingKey,message); return true; } } OrderVo
这里注意Vo包都要放在公共包中,因为生产者和消费者的服务都使用这个对象进行消息的封装传输,这个OrderVo类,只是简单举个例子.
package com.leixin.mq.vo; /** * @Classname OrderVo * @Description TODO * @Date 2023-08-08 0:04 * @Created by LeiXin */ public class OrderVo { String id; String orderName; Integer count; public Integer getCount() { return count; } public void setCount(Integer count) { this.count = count; } public String getId() { return id; } public void setId(String id) { this.id = id; } public String getOrderName() { return orderName; } public void setOrderName(String orderName) { this.orderName = orderName; } } 依赖
<parent> <groupId>com.leixin</groupId> <artifactId>SpringBoot-RabbitMq</artifactId> <version>1.0-SNAPSHOT</version> </parent> <artifactId>service_util</artifactId> <properties> <maven.compiler.source>17</maven.compiler.source> <maven.compiler.target>17</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <!-- web --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- 提供 Spring Boot 对 AMQP（高级消息队列协议）的支持，包括 RabbitMQ --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <!-- 提供 Java 的 Lombok 库，可以简化代码中的重复和样板代码，提高代码可读性和可维护性 --> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <!-- 提供 Spring Boot 单元测试和集成测试所需的支持 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> </dependencies> 生产者
生产者用于发送消息,简单举例,就直接在Controller层发送消息
└─producer │ XxxApplication.java │ └─controller XxxController.java XxxApplication
SpringBoot的启动类,注意,这个CompontScan主要是为了扫描到公共包的一些配置或者组件
@SpringBootApplication @ComponentScan("com.leixin") public class XxxApplication { public static void main(String[] args) { SpringApplication.run(XxxApplication.class, args); } } XxxController
生产者的服务发送至消息队列
@RestController @RequestMapping("/test") public class XxxController { @Autowired private RabbitService rabbitService; @GetMapping public void testSendMessage(){ OrderVo orderVo = new OrderVo(); orderVo.setCount(100); rabbitService.sendMessage(RabbitMQConstant.EXCHANGE_DIRECT_ORDER ,RabbitMQConstant.ROUTING_ORDER, orderVo); } } 配置类
application.yml
# Spring配置部分 spring: rabbitmq: # RabbitMQ主机地址 host: localhost # RabbitMQ端口 port: 5672 # 连接RabbitMQ的用户名 username: guest # 连接RabbitMQ的密码 password: guest # 虚拟主机，默认为根虚拟主机"/" virtual-host: / # Spring Boot内嵌服务器配置部分 server: # 内嵌服务器监听的端口号 port: 8092 依赖
<dependencies> <dependency> <groupId>com.leixin</groupId> <artifactId>service_util</artifactId> <version>1.0-SNAPSHOT</version> </dependency> </dependencies> 消费者
消费者主要用于处理一些消息,文件如下:
│ YyyApplication.java │ └─listener YyyMQListener.java YyyMQListener
这里就是用于处理消息
package com.leixin.mq.listener; import com.leixin.mq.MQConstant.RabbitMQConstant; import com.leixin.mq.vo.OrderVo; import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; 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; /** * @Classname YyyMQListener * @Description TODO * @Date 2023-08-07 23:54 * @Created by LeiXin */ @Component public class YyyMQListener { // 使用 @RabbitListener 注解标记这个方法作为消息监听器 // bindings 属性用于定义队列、交换机以及路由关系 @RabbitListener(bindings = { @QueueBinding( value = @Queue(name = RabbitMQConstant.QUEUE_ORDER, durable = "true"), // 创建队列 exchange = @Exchange(name = RabbitMQConstant.EXCHANGE_DIRECT_ORDER), // 创建交换机 key = RabbitMQConstant.ROUTING_ORDER ) }) // consume 方法用于处理收到的消息 public void consume(OrderVo order, Message message, Channel channel) { // 从传入的 OrderVo 对象中获取订单数量，并将其减少 Integer count = order.getCount(); count--; // 输出调试信息，展示订单数量的变化 System.out.println(count); // 在这里可以添加你的业务逻辑，处理订单相关的操作 // 例如，更新数据库中的订单状态、发送通知等 // 最后，可能需要手动确认消息的处理，通过 channel.basicAck 方法 } } 配置文件
application.yml
spring: rabbitmq: host: localhost port: 5672 username: guest password: guest virtual-host: / server: port: 8093 依赖
<dependencies> <dependency> <groupId>com.leixin</groupId> <artifactId>service_util</artifactId> <version>1.0-SNAPSHOT</version> </dependency> </dependencies> 
![洛谷 P11345 [KTSC 2023 R2] 基地简化 题解](http://www.itfaba.com/wp-content/themes/kemi/timthumb.php?src=http://www.itfaba.com/wp-content/themes/kemi/img/random/1.jpg&w=218&h=124&zc=1)
