Compare commits
13
Commits
add0848464
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2aed782d6d | ||
|
|
b905844154 | ||
|
|
5773645739 | ||
|
|
0eeee6ba1f | ||
|
|
0cab9d6b24 | ||
|
|
dc14fd3bb9 | ||
|
|
bde3725684 | ||
|
|
0f2c656e31 | ||
|
|
cbd8f91a55 | ||
|
|
e477ad36ba | ||
|
|
1d4fa82fe7 | ||
|
|
b1eb8e11c2 | ||
|
|
d072522937 |
+2
-1
@@ -1 +1,2 @@
|
||||
/.idea
|
||||
/.idea
|
||||
target
|
||||
@@ -1,22 +1,80 @@
|
||||
# rabbitmq-template
|
||||
|
||||
🦒 RabbitMQ 消息队列项目模板
|
||||
## 📋 项目概述
|
||||
|
||||
| 序号 | 任务 | 内容 | 是否完成 | 备注 |
|
||||
|------|----------|----------|----------|------|
|
||||
| 1 | Spring Boot 3 集成 RabbitMQ | 添加 `spring-boot-starter-amqp` 依赖,配置 `application.yml` 或 `application.properties` | | |
|
||||
| 2 | 实现简单消息发送与接收 | 编写生产者发送消息,消费者监听队列接收消息,测试流程 | | |
|
||||
| 3 | 理解 RabbitMQ 交换机类型 | 学习 direct、fanout、topic、headers 交换机类型,实现不同类型的消息路由 | | |
|
||||
| 4 | 消息持久化 | 配置队列和消息的持久化,测试 RabbitMQ 重启后消息是否保留 | | |
|
||||
| 5 | 消息确认机制(ACK) | 学习手动确认机制,实现消息确认和失败重试 | | |
|
||||
| 6 | 死信队列(DLQ) | 配置死信交换机与队列,演示消息被拒绝或过期后进入 DLQ | | |
|
||||
| 7 | 消息延迟(Delayed Message) | 使用 RabbitMQ 插件实现延迟消息,编写生产者和消费者测试 | | |
|
||||
| 8 | 消息事务与可靠性 | 使用事务机制或 confirm 模式保证消息发送可靠性 | | |
|
||||
| 9 | 消息限流与消费者并发控制 | 配置消费者并发数,使用 `prefetchCount` 控制消息分发速率 | | |
|
||||
| 10 | 使用 `@RabbitListener` 注解 | 学习监听器基本用法,处理消息的自动转换(如 JSON) | | |
|
||||
| 11 | 消息序列化与反序列化 | 配置消息的序列化方式(如 JSON),自定义消息转换器 | | |
|
||||
| 12 | 消息发布/订阅模式 | 使用 fanout 实现广播,topic 实现通配符匹配 | | |
|
||||
| 13 | 消息可靠性投递与重试机制 | 使用 `RabbitTemplate` 发送消息,配置失败重试策略 | | |
|
||||
| 14 | 消息幂等性处理 | 理解重复消费问题,实现消息幂等性校验(如去重标识) | | |
|
||||
| 15 | 监控与管理 RabbitMQ | 使用 RabbitMQ 管理界面监控,Spring Boot Actuator 集成监控 | | |
|
||||
| 16 | 实战项目练习 | 构建异步任务系统(如订单处理、日志记录),完整流程测试 | | |
|
||||
本项目是一个 Spring Boot 3 集成 RabbitMQ 的模板项目,包含了消息队列开发中的常见场景实现,可作为学习和开发的参考模板。
|
||||
|
||||
## 🚀 快速开始
|
||||
|
||||
### 环境要求
|
||||
- JDK 21
|
||||
- Spring Boot 3.2.2
|
||||
- RabbitMQ Server
|
||||
|
||||
### 安装配置
|
||||
|
||||
1. 克隆项目
|
||||
|
||||
```bash
|
||||
git clone [repository-url]
|
||||
cd rabbitmq-template
|
||||
```
|
||||
|
||||
2. 配置 RabbitMQ
|
||||
|
||||
在 `application.yml` 中配置你的 RabbitMQ 连接信息:
|
||||
|
||||
```yaml
|
||||
spring:
|
||||
rabbitmq:
|
||||
host: localhost
|
||||
port: 5672
|
||||
username: your-username
|
||||
password: your-password
|
||||
virtual-host: /template
|
||||
```
|
||||
|
||||
3. 运行项目
|
||||
|
||||
```bash
|
||||
mvn spring-boot:run
|
||||
```
|
||||
|
||||
## 📁 项目结构
|
||||
|
||||
```
|
||||
rabbitmq-template/
|
||||
├── src/main/java/cn/hezhaohui/rabbitmq/
|
||||
│ ├── Application.java # 主程序入口
|
||||
│ ├── Message.java # 消息实体类
|
||||
│ ├── MessageProducer.java # 消息生产者
|
||||
│ ├── MessageConsumer.java # 消息消费者
|
||||
│ └── RabbitmqConfig.java # RabbitMQ 配置类
|
||||
└── src/test/java/
|
||||
└── RabbitMQTest.java # 测试类
|
||||
```
|
||||
|
||||
## 🎯 核心功能
|
||||
|
||||
### 1. 消息发送与接收
|
||||
- 支持邮件消息队列
|
||||
- 支持短信消息队列
|
||||
- 使用 Topic 交换机实现消息路由
|
||||
|
||||
### 2. 消息配置
|
||||
- 自定义消息转换器(JSON)
|
||||
- 队列持久化配置
|
||||
- 交换机配置
|
||||
|
||||
### 3. 测试用例
|
||||
- 邮件消息发送测试
|
||||
- 短信消息发送测试
|
||||
|
||||
|
||||
## 🛠️ 技术栈
|
||||
|
||||
- Spring Boot 3.2.2
|
||||
- Spring AMQP
|
||||
- RabbitMQ
|
||||
- Lombok
|
||||
- Jackson
|
||||
@@ -0,0 +1,51 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
|
||||
http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
|
||||
<!-- 基本信息配置 -->
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>3.2.2</version>
|
||||
</parent>
|
||||
|
||||
<groupId>cn.hezhaohui</groupId>
|
||||
<artifactId>rabbitmq-template</artifactId>
|
||||
<version>1.0.0</version>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<!-- 属性配置 -->
|
||||
<properties>
|
||||
<java.version>21</java.version>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-amqp</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.fasterxml.jackson.core</groupId>
|
||||
<artifactId>jackson-databind</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
|
||||
</project>
|
||||
@@ -0,0 +1,11 @@
|
||||
package cn.hezhaohui.rabbitmq;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class Application {
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(Application.class, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package cn.hezhaohui.rabbitmq.amqp;
|
||||
|
||||
import cn.hezhaohui.rabbitmq.entity.Message;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
import static cn.hezhaohui.rabbitmq.constant.RabbitmqConstant.QUEUE_EMAIL_NAME;
|
||||
import static cn.hezhaohui.rabbitmq.constant.RabbitmqConstant.QUEUE_SMS_NAME;
|
||||
|
||||
/**
|
||||
* 消息消费者
|
||||
*/
|
||||
@Component
|
||||
@Slf4j
|
||||
public class MessageConsumer {
|
||||
|
||||
@RabbitListener(queues = QUEUE_EMAIL_NAME)
|
||||
public void receiveEmailMessage(Message message) {
|
||||
log.info("receiveEmailMessage: {}", message.getContent());
|
||||
}
|
||||
|
||||
@RabbitListener(queues = QUEUE_SMS_NAME)
|
||||
public void receiveSmsMessage(Message message) {
|
||||
log.info("receiveSmsMessage: {}", message.getContent());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
package cn.hezhaohui.rabbitmq.amqp;
|
||||
|
||||
import cn.hezhaohui.rabbitmq.entity.Message;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.Data;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import static cn.hezhaohui.rabbitmq.constant.RabbitmqConstant.*;
|
||||
|
||||
/**
|
||||
* 消息生产者
|
||||
*/
|
||||
@Component
|
||||
@Data
|
||||
@Slf4j
|
||||
public class MessageProducer {
|
||||
|
||||
@Resource
|
||||
private RabbitTemplate rabbitTemplate;
|
||||
|
||||
// 实现发送消息的方法
|
||||
public void sendEmailMessage(Message message) {
|
||||
rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY_EMAIL, message);
|
||||
log.info("发送邮件消息:{}", message);
|
||||
}
|
||||
|
||||
public void sendSmsMessage(Message message) {
|
||||
rabbitTemplate.convertAndSend(EXCHANGE_NAME, ROUTING_KEY_SMS, message);
|
||||
log.info("发送短信消息:{}", message);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
package cn.hezhaohui.rabbitmq.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 MyMessageConverter {
|
||||
@Bean
|
||||
MessageConverter createMessageConverter() {
|
||||
return new Jackson2JsonMessageConverter();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
package cn.hezhaohui.rabbitmq.constant;
|
||||
|
||||
public class RabbitmqConstant {
|
||||
public static final String EXCHANGE_NAME = "exchange_topics";
|
||||
public static final String QUEUE_EMAIL_NAME = "email.queue";
|
||||
public static final String QUEUE_SMS_NAME = "sms.queue";
|
||||
public static final String ROUTING_KEY_EMAIL = "email.#";
|
||||
public static final String ROUTING_KEY_SMS = "sms.#";
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package cn.hezhaohui.rabbitmq.entity;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
/**
|
||||
* 消息实体类
|
||||
*/
|
||||
@Data
|
||||
@JsonIgnoreProperties(ignoreUnknown = true)
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
public class Message {
|
||||
private String content;
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
server:
|
||||
port: 8081
|
||||
spring:
|
||||
application:
|
||||
name: rabbitmq-template
|
||||
rabbitmq:
|
||||
host: localhost
|
||||
port: 5672
|
||||
username: wonder
|
||||
password: dn293830
|
||||
virtual-host: /template
|
||||
@@ -0,0 +1,32 @@
|
||||
package cn.hezhaohui.rabbitmq.amqp;
|
||||
|
||||
import cn.hezhaohui.rabbitmq.entity.Message;
|
||||
import jakarta.annotation.Resource;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
|
||||
@SpringBootTest
|
||||
class MessageProducerTest {
|
||||
|
||||
@Resource
|
||||
private MessageProducer messageProducer;
|
||||
|
||||
|
||||
/**
|
||||
* 测试方法:用于发送电子邮件消息
|
||||
* 该方法使用messageProducer对象调用sendEmailMessage方法
|
||||
* 传入一个Message对象,其中包含电子邮件地址"hello@example.com"
|
||||
*/
|
||||
@Test
|
||||
void sendEmailMessage() throws InterruptedException { // 测试用例:验证电子邮件消息发送功能
|
||||
messageProducer.sendEmailMessage(new Message("hello@example.com")); // 创建并发送电子邮件消息
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
|
||||
@Test
|
||||
void sendSmsMessage() throws InterruptedException {
|
||||
messageProducer.sendSmsMessage(new Message("1234567890"));
|
||||
Thread.sleep(1000);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user