Java中的RabbitMQ使用場(chǎng)景和實(shí)踐完全指南
更新時(shí)間：2025年08月29日 14:58:30 作者：重生成為編程大王
本文涵蓋了RabbitMQ在Java中的主要使用場(chǎng)景和實(shí)踐方法,在實(shí)際應(yīng)用中,還需要根據(jù)具體的業(yè)務(wù)需求和系統(tǒng)架構(gòu)進(jìn)行適當(dāng)?shù)恼{(diào)整和優(yōu)化,本文結(jié)合代碼給大家介紹的非常詳細(xì),感興趣的朋友跟隨小編一起看看吧
Java中的RabbitMQ完全指南
1. 引言
什么是RabbitMQ
RabbitMQ是一個(gè)開(kāi)源的消息代理和隊(duì)列服務(wù)器，實(shí)現(xiàn)了高級(jí)消息隊(duì)列協(xié)議（AMQP）。它充當(dāng)應(yīng)用程序之間的消息中間件，允許分布式系統(tǒng)中的不同組件進(jìn)行異步通信。RabbitMQ使用Erlang語(yǔ)言開(kāi)發(fā)，以其高性能、可靠性和可擴(kuò)展性而聞名。
消息隊(duì)列的核心概念
消息隊(duì)列是一種異步通信機(jī)制，它允許應(yīng)用程序通過(guò)發(fā)送和接收消息來(lái)進(jìn)行通信，而不需要直接連接。這種模式帶來(lái)了以下優(yōu)勢(shì)：
- 解耦：生產(chǎn)者和消費(fèi)者不需要同時(shí)在線
- 可擴(kuò)展性：可以獨(dú)立擴(kuò)展生產(chǎn)者和消費(fèi)者
- 可靠性：消息可以持久化存儲(chǔ)，確保不丟失
- 靈活性：支持多種消息傳遞模式
RabbitMQ的優(yōu)勢(shì)和應(yīng)用場(chǎng)景
RabbitMQ在企業(yè)級(jí)應(yīng)用中具有以下優(yōu)勢(shì)：
- 多協(xié)議支持：支持AMQP、STOMP、MQTT等多種協(xié)議
- 靈活的路由：支持多種Exchange類型和復(fù)雜的路由規(guī)則
- 集群支持：可以構(gòu)建高可用的集群架構(gòu)
- 管理界面：提供Web管理控制臺(tái)
- 豐富的客戶端庫(kù)：支持多種編程語(yǔ)言
常見(jiàn)應(yīng)用場(chǎng)景包括：微服務(wù)解耦、異步任務(wù)處理、系統(tǒng)集成、削峰填谷等。
2. RabbitMQ基礎(chǔ)概念
Exchange（交換器）類型詳解
Exchange是RabbitMQ的核心組件，負(fù)責(zé)接收生產(chǎn)者發(fā)送的消息并將其路由到相應(yīng)的隊(duì)列。主要有四種類型：
Direct Exchange（直連交換器）
- 根據(jù)routing key精確匹配路由消息
- 適用于單播消息傳遞
- 默認(rèn)的交換器類型
Fanout Exchange（扇形交換器）
- 將消息廣播到所有綁定的隊(duì)列
- 忽略routing key
- 適用于廣播場(chǎng)景
Topic Exchange（主題交換器）
- 基于通配符模式匹配routing key
- 支持"*“（單個(gè)單詞）和”#"（零個(gè)或多個(gè)單詞）
- 靈活的路由規(guī)則
Headers Exchange（頭交換器）
- 基于消息頭屬性進(jìn)行路由
- 較少使用，性能相對(duì)較低
Queue（隊(duì)列）和消息持久化
Queue是存儲(chǔ)消息的容器，具有以下特性：
- FIFO原則：先進(jìn)先出的消息處理順序
- 持久化：可以配置隊(duì)列和消息的持久化
- 排他性：可以設(shè)置隊(duì)列只能被一個(gè)連接使用
- 自動(dòng)刪除：當(dāng)沒(méi)有消費(fèi)者時(shí)自動(dòng)刪除隊(duì)列
Routing Key和Binding
- Routing Key：生產(chǎn)者發(fā)送消息時(shí)指定的路由鍵
- Binding：Exchange和Queue之間的綁定關(guān)系
- Binding Key：綁定時(shí)指定的鍵，用于匹配routing key
Virtual Host（虛擬主機(jī)）
Virtual Host提供了邏輯隔離，類似于網(wǎng)絡(luò)中的虛擬主機(jī)概念：
- 不同vhost中的Exchange、Queue等資源完全隔離
- 每個(gè)vhost有獨(dú)立的權(quán)限控制
- 默認(rèn)vhost為"/"
3. Java環(huán)境準(zhǔn)備
RabbitMQ服務(wù)器安裝配置
Docker方式安裝（推薦）：
# 拉取RabbitMQ鏡像（包含管理插件） docker pull rabbitmq:3-management # 運(yùn)行RabbitMQ容器 docker run -d --name rabbitmq \ -p 5672:5672 \ -p 15672:15672 \ -e RABBITMQ_DEFAULT_USER=admin \ -e RABBITMQ_DEFAULT_PASS=admin123 \ rabbitmq:3-management
安裝完成后，可以通過(guò) http://localhost:15672 訪問(wèn)管理界面。
Maven/Gradle依賴配置
Maven配置：
<dependencies>
<!-- RabbitMQ Java客戶端 -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.18.0</version>
</dependency>
<!-- Spring Boot RabbitMQ Starter -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 日志依賴 -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>2.0.7</version>
</dependency>
</dependencies>Gradle配置：
dependencies {
implementation 'com.rabbitmq:amqp-client:5.18.0'
implementation 'org.springframework.boot:spring-boot-starter-amqp'
implementation 'org.slf4j:slf4j-simple:2.0.7'
}連接工廠和連接管理
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitMQConnection {
public static Connection getConnection() throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("admin123");
factory.setVirtualHost("/");
// 連接超時(shí)設(shè)置
factory.setConnectionTimeout(30000);
factory.setRequestedHeartbeat(60);
return factory.newConnection();
}
}4. 基礎(chǔ)消息模式實(shí)現(xiàn)
Simple Queue（簡(jiǎn)單隊(duì)列）
最基本的消息模式，一個(gè)生產(chǎn)者發(fā)送消息到隊(duì)列，一個(gè)消費(fèi)者從隊(duì)列接收消息。
生產(chǎn)者代碼：
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class SimpleProducer {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
// 聲明隊(duì)列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 發(fā)送消息
String message = "Hello, RabbitMQ!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
System.out.println("Sent: " + message);
}
}
}消費(fèi)者代碼：
import com.rabbitmq.client.*;
public class SimpleConsumer {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
// 聲明隊(duì)列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 創(chuàng)建消費(fèi)者
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Received: " + message);
};
// 開(kāi)始消費(fèi)
channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
System.out.println("Waiting for messages...");
}
}Work Queues（工作隊(duì)列）
工作隊(duì)列模式用于在多個(gè)消費(fèi)者之間分發(fā)耗時(shí)任務(wù)，實(shí)現(xiàn)負(fù)載均衡。
生產(chǎn)者代碼：
public class WorkQueueProducer {
private final static String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 發(fā)送多個(gè)任務(wù)
for (int i = 1; i <= 10; i++) {
String message = "Task " + i;
channel.basicPublish("", QUEUE_NAME,
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes("UTF-8"));
System.out.println("Sent: " + message);
}
}
}
}消費(fèi)者代碼：
public class WorkQueueConsumer {
private final static String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 設(shè)置每次只處理一個(gè)消息（公平分發(fā)）
channel.basicQos(1);
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Processing: " + message);
try {
// 模擬耗時(shí)處理
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// 手動(dòng)確認(rèn)消息
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
System.out.println("Completed: " + message);
};
// 關(guān)閉自動(dòng)確認(rèn)
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
System.out.println("Worker waiting for tasks...");
}
}Publish/Subscribe（發(fā)布訂閱）
發(fā)布訂閱模式使用fanout類型的Exchange將消息廣播到所有綁定的隊(duì)列。
發(fā)布者代碼：
public class PublishSubscribeProducer {
private final static String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
// 聲明fanout類型的交換器
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
String message = "Broadcast message to all subscribers!";
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
System.out.println("Published: " + message);
}
}
}訂閱者代碼：
public class PublishSubscribeConsumer {
private final static String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
// 聲明交換器
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
// 創(chuàng)建臨時(shí)隊(duì)列
String queueName = channel.queueDeclare().getQueue();
// 將隊(duì)列綁定到交換器
channel.queueBind(queueName, EXCHANGE_NAME, "");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Received: " + message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
System.out.println("Subscriber waiting for messages...");
}
}Routing（路由模式）
路由模式使用direct類型的Exchange根據(jù)routing key將消息路由到特定隊(duì)列。
生產(chǎn)者代碼：
public class RoutingProducer {
private final static String EXCHANGE_NAME = "direct_exchange";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
// 發(fā)送不同級(jí)別的日志消息
String[] levels = {"info", "warning", "error"};
for (String level : levels) {
String message = "This is a " + level + " message";
channel.basicPublish(EXCHANGE_NAME, level, null, message.getBytes("UTF-8"));
System.out.println("Sent [" + level + "]: " + message);
}
}
}
}消費(fèi)者代碼：
public class RoutingConsumer {
private final static String EXCHANGE_NAME = "direct_exchange";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
String queueName = channel.queueDeclare().getQueue();
// 只接收error級(jí)別的消息
channel.queueBind(queueName, EXCHANGE_NAME, "error");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println("Received [" + routingKey + "]: " + message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
System.out.println("Consumer waiting for error messages...");
}
}Topics（主題模式）
主題模式使用topic類型的Exchange支持通配符路由。
生產(chǎn)者代碼：
public class TopicProducer {
private final static String EXCHANGE_NAME = "topic_exchange";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 發(fā)送不同主題的消息
String[] routingKeys = {
"user.info.create",
"user.warning.update",
"order.error.payment",
"system.info.startup"
};
for (String routingKey : routingKeys) {
String message = "Message for " + routingKey;
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
System.out.println("Sent [" + routingKey + "]: " + message);
}
}
}
}消費(fèi)者代碼：
public class TopicConsumer {
private final static String EXCHANGE_NAME = "topic_exchange";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
String queueName = channel.queueDeclare().getQueue();
// 綁定多個(gè)模式
channel.queueBind(queueName, EXCHANGE_NAME, "user.*.*"); // 所有用戶相關(guān)消息
channel.queueBind(queueName, EXCHANGE_NAME, "*.error.*"); // 所有錯(cuò)誤消息
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println("Received [" + routingKey + "]: " + message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
System.out.println("Consumer waiting for messages matching patterns...");
}
}RPC模式
RPC模式實(shí)現(xiàn)遠(yuǎn)程過(guò)程調(diào)用，客戶端發(fā)送請(qǐng)求并等待服務(wù)端響應(yīng)。
RPC服務(wù)端：
public class RPCServer {
private final static String QUEUE_NAME = "rpc_queue";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicQos(1);
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
AMQP.BasicProperties replyProps = new AMQP.BasicProperties
.Builder()
.correlationId(delivery.getProperties().getCorrelationId())
.build();
String response = "";
try {
String message = new String(delivery.getBody(), "UTF-8");
int n = Integer.parseInt(message);
System.out.println("Computing fibonacci(" + n + ")");
response = String.valueOf(fibonacci(n));
} catch (RuntimeException e) {
System.out.println("Error: " + e.toString());
response = "Error: " + e.toString();
} finally {
channel.basicPublish("", delivery.getProperties().getReplyTo(),
replyProps, response.getBytes("UTF-8"));
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
};
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
System.out.println("RPC Server waiting for requests...");
}
private static int fibonacci(int n) {
if (n <= 1) return n;
return fibonacci(n - 1) + fibonacci(n - 2);
}
}RPC客戶端：
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class RPCClient implements AutoCloseable {
private Connection connection;
private Channel channel;
private String requestQueueName = "rpc_queue";
public RPCClient() throws Exception {
connection = RabbitMQConnection.getConnection();
channel = connection.createChannel();
}
public String call(String message) throws Exception {
final String corrId = UUID.randomUUID().toString();
String replyQueueName = channel.queueDeclare().getQueue();
AMQP.BasicProperties props = new AMQP.BasicProperties
.Builder()
.correlationId(corrId)
.replyTo(replyQueueName)
.build();
channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
if (delivery.getProperties().getCorrelationId().equals(corrId)) {
response.offer(new String(delivery.getBody(), "UTF-8"));
}
}, consumerTag -> {});
String result = response.take();
channel.basicCancel(ctag);
return result;
}
@Override
public void close() throws Exception {
connection.close();
}
public static void main(String[] args) throws Exception {
try (RPCClient client = new RPCClient()) {
System.out.println("Requesting fibonacci(10)");
String response = client.call("10");
System.out.println("Got: " + response);
}
}
}5. Spring Boot集成RabbitMQ
Spring AMQP配置
application.yml配置：
spring:
rabbitmq:
host: localhost
port: 5672
username: admin
password: admin123
virtual-host: /
connection-timeout: 30000
publisher-confirm-type: correlated
publisher-returns: true
listener:
simple:
acknowledge-mode: manual
concurrency: 2
max-concurrency: 10配置類：
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
public static final String DIRECT_EXCHANGE = "spring.direct.exchange";
public static final String TOPIC_EXCHANGE = "spring.topic.exchange";
public static final String DIRECT_QUEUE = "spring.direct.queue";
public static final String TOPIC_QUEUE = "spring.topic.queue";
// 聲明Direct Exchange
@Bean
public DirectExchange directExchange() {
return new DirectExchange(DIRECT_EXCHANGE, true, false);
}
// 聲明Topic Exchange
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(TOPIC_EXCHANGE, true, false);
}
// 聲明隊(duì)列
@Bean
public Queue directQueue() {
return QueueBuilder.durable(DIRECT_QUEUE).build();
}
@Bean
public Queue topicQueue() {
return QueueBuilder.durable(TOPIC_QUEUE).build();
}
// 綁定關(guān)系
@Bean
public Binding directBinding() {
return BindingBuilder.bind(directQueue())
.to(directExchange())
.with("direct.routing.key");
}
@Bean
public Binding topicBinding() {
return BindingBuilder.bind(topicQueue())
.to(topicExchange())
.with("topic.*.key");
}
}RabbitTemplate使用
消息生產(chǎn)者服務(wù)：
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class MessageProducerService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendDirectMessage(String message) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.DIRECT_EXCHANGE,
"direct.routing.key",
message
);
System.out.println("Sent direct message: " + message);
}
public void sendTopicMessage(String routingKey, String message) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.TOPIC_EXCHANGE,
routingKey,
message
);
System.out.println("Sent topic message with key " + routingKey + ": " + message);
}
// 發(fā)送對(duì)象消息
public void sendObjectMessage(Object obj) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.DIRECT_EXCHANGE,
"direct.routing.key",
obj
);
}
}@RabbitListener注解詳解
消息消費(fèi)者服務(wù)：
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
import com.rabbitmq.client.Channel;
@Service
public class MessageConsumerService {
// 基礎(chǔ)消費(fèi)者
@RabbitListener(queues = RabbitMQConfig.DIRECT_QUEUE)
public void handleDirectMessage(String message) {
System.out.println("Received direct message: " + message);
}
// 手動(dòng)確認(rèn)消息
@RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE)
public void handleTopicMessage(
@Payload String message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws Exception {
try {
System.out.println("Processing topic message: " + message);
// 模擬業(yè)務(wù)處理
Thread.sleep(1000);
// 手動(dòng)確認(rèn)
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
System.err.println("Error processing message: " + e.getMessage());
// 拒絕消息并重新入隊(duì)
channel.basicNack(deliveryTag, false, true);
}
}
// 接收完整消息對(duì)象
@RabbitListener(queues = RabbitMQConfig.DIRECT_QUEUE)
public void handleCompleteMessage(Message message, Channel channel) throws Exception {
String body = new String(message.getBody());
String routingKey = message.getMessageProperties().getReceivedRoutingKey();
System.out.println("Received message: " + body + " with routing key: " + routingKey);
// 手動(dòng)確認(rèn)
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}消息轉(zhuǎn)換器配置
JSON消息轉(zhuǎn)換器：
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MessageConverterConfig {
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(jsonMessageConverter());
// 配置確認(rèn)回調(diào)
template.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
System.out.println("Message sent successfully");
} else {
System.out.println("Message send failed: " + cause);
}
});
// 配置返回回調(diào)
template.setReturnsCallback(returned -> {
System.out.println("Message returned: " + returned.getMessage());
});
return template;
}
}使用JSON轉(zhuǎn)換器發(fā)送對(duì)象：
// 定義消息對(duì)象
public class UserMessage {
private Long id;
private String name;
private String email;
// 構(gòu)造函數(shù)、getter、setter
public UserMessage() {}
public UserMessage(Long id, String name, String email) {
this.id = id;
this.name = name;
this.email = email;
}
// getter和setter方法...
}
// 發(fā)送和接收對(duì)象消息
@Service
public class UserMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendUserMessage(UserMessage user) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.DIRECT_EXCHANGE,
"user.routing.key",
user
);
}
@RabbitListener(queues = "user.queue")
public void handleUserMessage(UserMessage user) {
System.out.println("Received user: " + user.getName() + " (" + user.getEmail() + ")");
}
}6. 高級(jí)特性
消息確認(rèn)機(jī)制（ACK）
RabbitMQ提供了多種消息確認(rèn)機(jī)制來(lái)保證消息的可靠傳遞：
自動(dòng)確認(rèn)（Auto ACK）：
// 消息被消費(fèi)者接收后立即確認(rèn)
@RabbitListener(queues = "auto.ack.queue", ackMode = "AUTO")
public void handleAutoAck(String message) {
System.out.println("Auto ACK: " + message);
}手動(dòng)確認(rèn)（Manual ACK）：
@RabbitListener(queues = "manual.ack.queue", ackMode = "MANUAL")
public void handleManualAck(
String message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws Exception {
try {
// 處理業(yè)務(wù)邏輯
processMessage(message);
// 確認(rèn)消息
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
// 拒絕消息，重新入隊(duì)
channel.basicNack(deliveryTag, false, true);
}
}發(fā)布確認(rèn)：
@Configuration
public class PublisherConfirmConfig {
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
// 啟用發(fā)布確認(rèn)
template.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
System.out.println("Message published successfully");
} else {
System.out.println("Message publish failed: " + cause);
}
});
// 啟用消息返回
template.setReturnsCallback(returned -> {
System.out.println("Message returned: " + returned.getMessage().toString());
System.out.println("Reply code: " + returned.getReplyCode());
System.out.println("Reply text: " + returned.getReplyText());
});
return template;
}
}死信隊(duì)列（DLX）處理
死信隊(duì)列用于處理無(wú)法正常消費(fèi)的消息，常見(jiàn)的死信場(chǎng)景包括：
- 消息被拒絕且不重新入隊(duì)
- 消息TTL過(guò)期
- 隊(duì)列達(dá)到最大長(zhǎng)度
死信隊(duì)列配置：
@Configuration
public class DeadLetterConfig {
public static final String BUSINESS_EXCHANGE = "business.exchange";
public static final String BUSINESS_QUEUE = "business.queue";
public static final String DEAD_LETTER_EXCHANGE = "dlx.exchange";
public static final String DEAD_LETTER_QUEUE = "dlx.queue";
// 業(yè)務(wù)交換器
@Bean
public DirectExchange businessExchange() {
return new DirectExchange(BUSINESS_EXCHANGE);
}
// 死信交換器
@Bean
public DirectExchange deadLetterExchange() {
return new DirectExchange(DEAD_LETTER_EXCHANGE);
}
// 業(yè)務(wù)隊(duì)列（配置死信交換器）
@Bean
public Queue businessQueue() {
return QueueBuilder
.durable(BUSINESS_QUEUE)
.withArgument("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE)
.withArgument("x-dead-letter-routing-key", "dead.letter.routing.key")
.withArgument("x-message-ttl", 10000) // 消息TTL 10秒
.build();
}
// 死信隊(duì)列
@Bean
public Queue deadLetterQueue() {
return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
}
// 綁定關(guān)系
@Bean
public Binding businessBinding() {
return BindingBuilder.bind(businessQueue())
.to(businessExchange())
.with("business.routing.key");
}
@Bean
public Binding deadLetterBinding() {
return BindingBuilder.bind(deadLetterQueue())
.to(deadLetterExchange())
.with("dead.letter.routing.key");
}
}死信處理服務(wù)：
@Service
public class DeadLetterService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 發(fā)送業(yè)務(wù)消息
public void sendBusinessMessage(String message) {
rabbitTemplate.convertAndSend(
DeadLetterConfig.BUSINESS_EXCHANGE,
"business.routing.key",
message
);
}
// 業(yè)務(wù)消息處理
@RabbitListener(queues = DeadLetterConfig.BUSINESS_QUEUE)
public void handleBusinessMessage(
String message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws Exception {
try {
System.out.println("Processing business message: " + message);
// 模擬處理失敗
if (message.contains("error")) {
throw new RuntimeException("Business processing failed");
}
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
System.err.println("Business processing failed: " + e.getMessage());
// 拒絕消息，不重新入隊(duì)，進(jìn)入死信隊(duì)列
channel.basicNack(deliveryTag, false, false);
}
}
// 死信消息處理
@RabbitListener(queues = DeadLetterConfig.DEAD_LETTER_QUEUE)
public void handleDeadLetterMessage(String message) {
System.out.println("Handling dead letter message: " + message);
// 記錄日志、發(fā)送告警、人工處理等
logDeadLetterMessage(message);
}
private void logDeadLetterMessage(String message) {
// 實(shí)現(xiàn)日志記錄邏輯
System.out.println("Dead letter logged: " + message);
}
}消息TTL和隊(duì)列過(guò)期
消息TTL配置：
@Service
public class TTLMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 發(fā)送帶TTL的消息
public void sendTTLMessage(String message, int ttlSeconds) {
MessageProperties properties = new MessageProperties();
properties.setExpiration(String.valueOf(ttlSeconds * 1000)); // 毫秒
Message msg = new Message(message.getBytes(), properties);
rabbitTemplate.send("ttl.exchange", "ttl.routing.key", msg);
}
// 使用MessagePostProcessor設(shè)置TTL
public void sendTTLMessageWithProcessor(String message, int ttlSeconds) {
rabbitTemplate.convertAndSend(
"ttl.exchange",
"ttl.routing.key",
message,
msg -> {
msg.getMessageProperties().setExpiration(String.valueOf(ttlSeconds * 1000));
return msg;
}
);
}
}隊(duì)列TTL配置：
@Bean
public Queue ttlQueue() {
return QueueBuilder
.durable("ttl.queue")
.withArgument("x-message-ttl", 60000) // 隊(duì)列中消息的默認(rèn)TTL
.withArgument("x-expires", 300000) // 隊(duì)列沒(méi)有消費(fèi)者時(shí)的過(guò)期時(shí)間
.build();
}優(yōu)先級(jí)隊(duì)列
優(yōu)先級(jí)隊(duì)列配置：
@Bean
public Queue priorityQueue() {
return QueueBuilder
.durable("priority.queue")
.withArgument("x-max-priority", 10) // 最大優(yōu)先級(jí)為10
.build();
}
發(fā)送優(yōu)先級(jí)消息：
@Service
public class PriorityMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendPriorityMessage(String message, int priority) {
rabbitTemplate.convertAndSend(
"priority.exchange",
"priority.routing.key",
message,
msg -> {
msg.getMessageProperties().setPriority(priority);
return msg;
}
);
}
@RabbitListener(queues = "priority.queue")
public void handlePriorityMessage(String message,
@Header("priority") Integer priority) {
System.out.println("Received priority " + priority + " message: " + message);
}
}7. 性能優(yōu)化和最佳實(shí)踐
連接池管理
連接池配置：
@Configuration
public class RabbitConnectionConfig {
@Bean
public CachingConnectionFactory connectionFactory() {
CachingConnectionFactory factory = new CachingConnectionFactory("localhost");
factory.setUsername("admin");
factory.setPassword("admin123");
// 連接池配置
factory.setChannelCacheSize(50); // 緩存的Channel數(shù)量
factory.setConnectionCacheSize(10); // 緩存的Connection數(shù)量
factory.setChannelCheckoutTimeout(30000); // Channel獲取超時(shí)時(shí)間
// 連接超時(shí)配置
factory.setConnectionTimeout(30000);
factory.setRequestedHeartBeat(60);
// 發(fā)布確認(rèn)
factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
factory.setPublisherReturns(true);
return factory;
}
}批量處理消息
批量發(fā)送消息：
@Service
public class BatchMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendBatchMessages(List<String> messages) {
// 使用事務(wù)批量發(fā)送
rabbitTemplate.execute(channel -> {
channel.txSelect(); // 開(kāi)啟事務(wù)
try {
for (String message : messages) {
channel.basicPublish(
"batch.exchange",
"batch.routing.key",
null,
message.getBytes("UTF-8")
);
}
channel.txCommit(); // 提交事務(wù)
} catch (Exception e) {
channel.txRollback(); // 回滾事務(wù)
throw new RuntimeException("Batch send failed", e);
}
return null;
});
}
// 使用發(fā)布確認(rèn)批量發(fā)送
public void sendBatchWithConfirm(List<String> messages) {
rabbitTemplate.execute(channel -> {
channel.confirmSelect(); // 開(kāi)啟發(fā)布確認(rèn)模式
for (String message : messages) {
channel.basicPublish(
"batch.exchange",
"batch.routing.key",
null,
message.getBytes("UTF-8")
);
}
// 等待所有消息確認(rèn)
boolean allConfirmed = channel.waitForConfirms(5000);
if (!allConfirmed) {
throw new RuntimeException("Not all messages were confirmed");
}
return null;
});
}
}批量消費(fèi)消息：
@Component
public class BatchConsumerService {
private final List<String> messageBuffer = new ArrayList<>();
private final int BATCH_SIZE = 100;
@RabbitListener(queues = "batch.queue", ackMode = "MANUAL")
public void handleBatchMessage(
String message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws Exception {
synchronized (messageBuffer) {
messageBuffer.add(message);
if (messageBuffer.size() >= BATCH_SIZE) {
processBatch(new ArrayList<>(messageBuffer));
messageBuffer.clear();
// 批量確認(rèn)消息
channel.basicAck(deliveryTag, true);
}
}
}
private void processBatch(List<String> messages) {
System.out.println("Processing batch of " + messages.size() + " messages");
// 批量處理邏輯
for (String message : messages) {
// 處理單個(gè)消息
System.out.println("Processing: " + message);
}
}
// 定時(shí)處理剩余消息
@Scheduled(fixedDelay = 5000)
public void processRemainingMessages() {
synchronized (messageBuffer) {
if (!messageBuffer.isEmpty()) {
processBatch(new ArrayList<>(messageBuffer));
messageBuffer.clear();
}
}
}
}消費(fèi)者并發(fā)控制
并發(fā)配置：
spring:
rabbitmq:
listener:
simple:
concurrency: 5 # 初始消費(fèi)者數(shù)量
max-concurrency: 20 # 最大消費(fèi)者數(shù)量
prefetch: 10 # 預(yù)取消息數(shù)量
acknowledge-mode: manual
retry:
enabled: true
initial-interval: 1000
max-attempts: 3
multiplier: 2動(dòng)態(tài)并發(fā)控制：
@Component
public class DynamicConcurrencyService {
@Autowired
private SimpleRabbitListenerContainerFactory factory;
@EventListener
public void handleHighLoad(HighLoadEvent event) {
// 動(dòng)態(tài)調(diào)整并發(fā)數(shù)
factory.setConcurrentConsumers(10);
factory.setMaxConcurrentConsumers(50);
}
@EventListener
public void handleLowLoad(LowLoadEvent event) {
factory.setConcurrentConsumers(2);
factory.setMaxConcurrentConsumers(10);
}
}監(jiān)控和日志記錄
監(jiān)控配置：
@Component
public class RabbitMQMetrics {
private final MeterRegistry meterRegistry;
private final Counter messagesSent;
private final Counter messagesReceived;
private final Timer messageProcessingTime;
public RabbitMQMetrics(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
this.messagesSent = Counter.builder("rabbitmq.messages.sent")
.description("Number of messages sent")
.register(meterRegistry);
this.messagesReceived = Counter.builder("rabbitmq.messages.received")
.description("Number of messages received")
.register(meterRegistry);
this.messageProcessingTime = Timer.builder("rabbitmq.message.processing.time")
.description("Message processing time")
.register(meterRegistry);
}
public void incrementMessagesSent() {
messagesSent.increment();
}
public void incrementMessagesReceived() {
messagesReceived.increment();
}
public Timer.Sample startProcessingTimer() {
return Timer.start(meterRegistry);
}
}日志配置：
@Component
@Slf4j
public class MessageLoggingService {
@Autowired
private RabbitMQMetrics metrics;
@RabbitListener(queues = "monitored.queue")
public void handleMonitoredMessage(String message) {
Timer.Sample sample = metrics.startProcessingTimer();
try {
log.info("Processing message: {}", message);
// 業(yè)務(wù)處理邏輯
processMessage(message);
metrics.incrementMessagesReceived();
log.info("Message processed successfully: {}", message);
} catch (Exception e) {
log.error("Error processing message: {}", message, e);
throw e;
} finally {
sample.stop(metrics.getMessageProcessingTime());
}
}
private void processMessage(String message) {
// 實(shí)際業(yè)務(wù)處理
}
}8. 實(shí)戰(zhàn)案例
訂單處理系統(tǒng)
在電商系統(tǒng)中，訂單處理涉及多個(gè)步驟：庫(kù)存檢查、支付處理、物流安排等。使用RabbitMQ可以實(shí)現(xiàn)異步處理和系統(tǒng)解耦。
訂單消息定義：
public class OrderMessage {
private Long orderId;
private Long userId;
private List<OrderItem> items;
private BigDecimal totalAmount;
private String status;
private Date createTime;
// 構(gòu)造函數(shù)、getter、setter
public OrderMessage() {}
public OrderMessage(Long orderId, Long userId, List<OrderItem> items, BigDecimal totalAmount) {
this.orderId = orderId;
this.userId = userId;
this.items = items;
this.totalAmount = totalAmount;
this.status = "CREATED";
this.createTime = new Date();
}
// getter和setter方法...
}
public class OrderItem {
private Long productId;
private String productName;
private Integer quantity;
private BigDecimal price;
// 構(gòu)造函數(shù)、getter、setter...
}訂單處理配置：
@Configuration
public class OrderProcessingConfig {
public static final String ORDER_EXCHANGE = "order.exchange";
public static final String ORDER_CREATED_QUEUE = "order.created.queue";
public static final String INVENTORY_CHECK_QUEUE = "inventory.check.queue";
public static final String PAYMENT_PROCESS_QUEUE = "payment.process.queue";
public static final String SHIPPING_ARRANGE_QUEUE = "shipping.arrange.queue";
@Bean
public TopicExchange orderExchange() {
return new TopicExchange(ORDER_EXCHANGE);
}
@Bean
public Queue orderCreatedQueue() {
return QueueBuilder.durable(ORDER_CREATED_QUEUE).build();
}
@Bean
public Queue inventoryCheckQueue() {
return QueueBuilder.durable(INVENTORY_CHECK_QUEUE).build();
}
@Bean
public Queue paymentProcessQueue() {
return QueueBuilder.durable(PAYMENT_PROCESS_QUEUE).build();
}
@Bean
public Queue shippingArrangeQueue() {
return QueueBuilder.durable(SHIPPING_ARRANGE_QUEUE).build();
}
// 綁定關(guān)系
@Bean
public Binding orderCreatedBinding() {
return BindingBuilder.bind(orderCreatedQueue())
.to(orderExchange()).with("order.created");
}
@Bean
public Binding inventoryCheckBinding() {
return BindingBuilder.bind(inventoryCheckQueue())
.to(orderExchange()).with("order.inventory.check");
}
@Bean
public Binding paymentProcessBinding() {
return BindingBuilder.bind(paymentProcessQueue())
.to(orderExchange()).with("order.payment.process");
}
@Bean
public Binding shippingArrangeBinding() {
return BindingBuilder.bind(shippingArrangeQueue())
.to(orderExchange()).with("order.shipping.arrange");
}
}訂單處理服務(wù)：
@Service
@Slf4j
public class OrderProcessingService {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private InventoryService inventoryService;
@Autowired
private PaymentService paymentService;
@Autowired
private ShippingService shippingService;
// 創(chuàng)建訂單
public void createOrder(OrderMessage order) {
log.info("Creating order: {}", order.getOrderId());
// 發(fā)送訂單創(chuàng)建消息
rabbitTemplate.convertAndSend(
OrderProcessingConfig.ORDER_EXCHANGE,
"order.created",
order
);
}
// 處理訂單創(chuàng)建
@RabbitListener(queues = OrderProcessingConfig.ORDER_CREATED_QUEUE)
public void handleOrderCreated(OrderMessage order) {
log.info("Processing created order: {}", order.getOrderId());
// 觸發(fā)庫(kù)存檢查
rabbitTemplate.convertAndSend(
OrderProcessingConfig.ORDER_EXCHANGE,
"order.inventory.check",
order
);
}
// 處理庫(kù)存檢查
@RabbitListener(queues = OrderProcessingConfig.INVENTORY_CHECK_QUEUE)
public void handleInventoryCheck(OrderMessage order) {
log.info("Checking inventory for order: {}", order.getOrderId());
try {
boolean inventoryAvailable = inventoryService.checkInventory(order.getItems());
if (inventoryAvailable) {
order.setStatus("INVENTORY_CONFIRMED");
// 觸發(fā)支付處理
rabbitTemplate.convertAndSend(
OrderProcessingConfig.ORDER_EXCHANGE,
"order.payment.process",
order
);
} else {
order.setStatus("INVENTORY_INSUFFICIENT");
log.warn("Insufficient inventory for order: {}", order.getOrderId());
// 發(fā)送庫(kù)存不足通知
}
} catch (Exception e) {
log.error("Error checking inventory for order: {}", order.getOrderId(), e);
order.setStatus("INVENTORY_CHECK_FAILED");
}
}
// 處理支付
@RabbitListener(queues = OrderProcessingConfig.PAYMENT_PROCESS_QUEUE)
public void handlePaymentProcess(OrderMessage order) {
log.info("Processing payment for order: {}", order.getOrderId());
try {
boolean paymentSuccess = paymentService.processPayment(
order.getUserId(),
order.getTotalAmount()
);
if (paymentSuccess) {
order.setStatus("PAID");
// 觸發(fā)物流安排
rabbitTemplate.convertAndSend(
OrderProcessingConfig.ORDER_EXCHANGE,
"order.shipping.arrange",
order
);
} else {
order.setStatus("PAYMENT_FAILED");
log.warn("Payment failed for order: {}", order.getOrderId());
}
} catch (Exception e) {
log.error("Error processing payment for order: {}", order.getOrderId(), e);
order.setStatus("PAYMENT_ERROR");
}
}
// 處理物流安排
@RabbitListener(queues = OrderProcessingConfig.SHIPPING_ARRANGE_QUEUE)
public void handleShippingArrange(OrderMessage order) {
log.info("Arranging shipping for order: {}", order.getOrderId());
try {
String trackingNumber = shippingService.arrangeShipping(order);
order.setStatus("SHIPPED");
log.info("Order {} shipped with tracking number: {}",
order.getOrderId(), trackingNumber);
} catch (Exception e) {
log.error("Error arranging shipping for order: {}", order.getOrderId(), e);
order.setStatus("SHIPPING_FAILED");
}
}
}異步任務(wù)處理
對(duì)于耗時(shí)的任務(wù)，如圖片處理、報(bào)表生成、郵件發(fā)送等，可以使用RabbitMQ實(shí)現(xiàn)異步處理。
任務(wù)處理配置：
@Configuration
public class TaskProcessingConfig {
public static final String TASK_EXCHANGE = "task.exchange";
public static final String IMAGE_PROCESS_QUEUE = "task.image.process.queue";
public static final String EMAIL_SEND_QUEUE = "task.email.send.queue";
public static final String REPORT_GENERATE_QUEUE = "task.report.generate.queue";
@Bean
public DirectExchange taskExchange() {
return new DirectExchange(TASK_EXCHANGE);
}
@Bean
public Queue imageProcessQueue() {
return QueueBuilder
.durable(IMAGE_PROCESS_QUEUE)
.withArgument("x-max-priority", 10) // 支持優(yōu)先級(jí)
.build();
}
@Bean
public Queue emailSendQueue() {
return QueueBuilder.durable(EMAIL_SEND_QUEUE).build();
}
@Bean
public Queue reportGenerateQueue() {
return QueueBuilder
.durable(REPORT_GENERATE_QUEUE)
.withArgument("x-message-ttl", 300000) // 5分鐘TTL
.build();
}
// 綁定關(guān)系
@Bean
public Binding imageProcessBinding() {
return BindingBuilder.bind(imageProcessQueue())
.to(taskExchange()).with("task.image.process");
}
@Bean
public Binding emailSendBinding() {
return BindingBuilder.bind(emailSendQueue())
.to(taskExchange()).with("task.email.send");
}
@Bean
public Binding reportGenerateBinding() {
return BindingBuilder.bind(reportGenerateQueue())
.to(taskExchange()).with("task.report.generate");
}
}任務(wù)消息定義：
public class TaskMessage {
private String taskId;
private String taskType;
private Map<String, Object> parameters;
private String status;
private Date createTime;
private Date processTime;
// 構(gòu)造函數(shù)、getter、setter...
}
public class ImageProcessTask extends TaskMessage {
private String imageUrl;
private String targetFormat;
private Map<String, Object> processingOptions;
// 構(gòu)造函數(shù)、getter、setter...
}
public class EmailTask extends TaskMessage {
private String to;
private String subject;
private String content;
private List<String> attachments;
// 構(gòu)造函數(shù)、getter、setter...
}任務(wù)處理服務(wù)：
@Service
@Slf4j
public class TaskProcessingService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 提交圖片處理任務(wù)
public void submitImageProcessTask(ImageProcessTask task, int priority) {
rabbitTemplate.convertAndSend(
TaskProcessingConfig.TASK_EXCHANGE,
"task.image.process",
task,
message -> {
message.getMessageProperties().setPriority(priority);
return message;
}
);
log.info("Submitted image process task: {}", task.getTaskId());
}
// 處理圖片處理任務(wù)
@RabbitListener(queues = TaskProcessingConfig.IMAGE_PROCESS_QUEUE,
concurrency = "2-10") // 動(dòng)態(tài)并發(fā)
public void handleImageProcessTask(ImageProcessTask task) {
log.info("Processing image task: {}", task.getTaskId());
try {
task.setStatus("PROCESSING");
task.setProcessTime(new Date());
// 圖片處理邏輯
processImage(task);
task.setStatus("COMPLETED");
log.info("Image task completed: {}", task.getTaskId());
} catch (Exception e) {
task.setStatus("FAILED");
log.error("Image task failed: {}", task.getTaskId(), e);
}
}
// 處理郵件發(fā)送任務(wù)
@RabbitListener(queues = TaskProcessingConfig.EMAIL_SEND_QUEUE)
public void handleEmailSendTask(EmailTask task) {
log.info("Sending email task: {}", task.getTaskId());
try {
sendEmail(task);
task.setStatus("SENT");
log.info("Email sent successfully: {}", task.getTaskId());
} catch (Exception e) {
task.setStatus("FAILED");
log.error("Email send failed: {}", task.getTaskId(), e);
// 重試邏輯或進(jìn)入死信隊(duì)列
throw new AmqpRejectAndDontRequeueException("Email send failed", e);
}
}
private void processImage(ImageProcessTask task) {
// 實(shí)現(xiàn)圖片處理邏輯
log.info("Processing image: {} -> {}", task.getImageUrl(), task.getTargetFormat());
// 模擬處理時(shí)間
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void sendEmail(EmailTask task) {
// 實(shí)現(xiàn)郵件發(fā)送邏輯
log.info("Sending email to: {}, subject: {}", task.getTo(), task.getSubject());
// 模擬發(fā)送時(shí)間
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}微服務(wù)間通信
在微服務(wù)架構(gòu)中，RabbitMQ可以作為服務(wù)間異步通信的消息總線。
事件驅(qū)動(dòng)架構(gòu)配置：
@Configuration
public class EventDrivenConfig {
public static final String EVENT_EXCHANGE = "event.exchange";
public static final String USER_SERVICE_QUEUE = "user.service.queue";
public static final String ORDER_SERVICE_QUEUE = "order.service.queue";
public static final String NOTIFICATION_SERVICE_QUEUE = "notification.service.queue";
@Bean
public TopicExchange eventExchange() {
return new TopicExchange(EVENT_EXCHANGE);
}
@Bean
public Queue userServiceQueue() {
return QueueBuilder.durable(USER_SERVICE_QUEUE).build();
}
@Bean
public Queue orderServiceQueue() {
return QueueBuilder.durable(ORDER_SERVICE_QUEUE).build();
}
@Bean
public Queue notificationServiceQueue() {
return QueueBuilder.durable(NOTIFICATION_SERVICE_QUEUE).build();
}
// 用戶服務(wù)監(jiān)聽(tīng)用戶相關(guān)事件
@Bean
public Binding userServiceBinding() {
return BindingBuilder.bind(userServiceQueue())
.to(eventExchange()).with("user.*");
}
// 訂單服務(wù)監(jiān)聽(tīng)訂單和用戶事件
@Bean
public Binding orderServiceBinding1() {
return BindingBuilder.bind(orderServiceQueue())
.to(eventExchange()).with("order.*");
}
@Bean
public Binding orderServiceBinding2() {
return BindingBuilder.bind(orderServiceQueue())
.to(eventExchange()).with("user.registered");
}
// 通知服務(wù)監(jiān)聽(tīng)所有事件
@Bean
public Binding notificationServiceBinding() {
return BindingBuilder.bind(notificationServiceQueue())
.to(eventExchange()).with("*.*");
}
}事件處理服務(wù)：
@Service
@Slf4j
public class EventService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 發(fā)布用戶注冊(cè)事件
public void publishUserRegisteredEvent(UserRegisteredEvent event) {
rabbitTemplate.convertAndSend(
EventDrivenConfig.EVENT_EXCHANGE,
"user.registered",
event
);
log.info("Published user registered event: {}", event.getUserId());
}
// 發(fā)布訂單創(chuàng)建事件
public void publishOrderCreatedEvent(OrderCreatedEvent event) {
rabbitTemplate.convertAndSend(
EventDrivenConfig.EVENT_EXCHANGE,
"order.created",
event
);
log.info("Published order created event: {}", event.getOrderId());
}
// 用戶服務(wù)處理用戶事件
@RabbitListener(queues = EventDrivenConfig.USER_SERVICE_QUEUE)
public void handleUserEvents(
@Payload Object event,
@Header("amqp_receivedRoutingKey") String routingKey) {
log.info("User service received event: {} with routing key: {}",
event.getClass().getSimpleName(), routingKey);
switch (routingKey) {
case "user.registered":
handleUserRegistered((UserRegisteredEvent) event);
break;
case "user.updated":
handleUserUpdated((UserUpdatedEvent) event);
break;
default:
log.warn("Unknown user event: {}", routingKey);
}
}
// 訂單服務(wù)處理訂單和用戶事件
@RabbitListener(queues = EventDrivenConfig.ORDER_SERVICE_QUEUE)
public void handleOrderEvents(
@Payload Object event,
@Header("amqp_receivedRoutingKey") String routingKey) {
log.info("Order service received event: {} with routing key: {}",
event.getClass().getSimpleName(), routingKey);
switch (routingKey) {
case "user.registered":
// 為新用戶創(chuàng)建優(yōu)惠券
createWelcomeCoupon((UserRegisteredEvent) event);
break;
case "order.created":
handleOrderCreated((OrderCreatedEvent) event);
break;
case "order.paid":
handleOrderPaid((OrderPaidEvent) event);
break;
default:
log.warn("Unknown order event: {}", routingKey);
}
}
// 通知服務(wù)處理所有事件
@RabbitListener(queues = EventDrivenConfig.NOTIFICATION_SERVICE_QUEUE)
public void handleNotificationEvents(
@Payload Object event,
@Header("amqp_receivedRoutingKey") String routingKey) {
log.info("Notification service received event: {} with routing key: {}",
event.getClass().getSimpleName(), routingKey);
// 根據(jù)事件類型發(fā)送不同的通知
switch (routingKey) {
case "user.registered":
sendWelcomeNotification((UserRegisteredEvent) event);
break;
case "order.created":
sendOrderConfirmationNotification((OrderCreatedEvent) event);
break;
case "order.shipped":
sendShippingNotification((OrderShippedEvent) event);
break;
default:
log.debug("No notification needed for event: {}", routingKey);
}
}
private void handleUserRegistered(UserRegisteredEvent event) {
log.info("Handling user registration: {}", event.getUserId());
// 用戶服務(wù)內(nèi)部處理邏輯
}
private void handleUserUpdated(UserUpdatedEvent event) {
log.info("Handling user update: {}", event.getUserId());
// 用戶更新處理邏輯
}
private void createWelcomeCoupon(UserRegisteredEvent event) {
log.info("Creating welcome coupon for user: {}", event.getUserId());
// 創(chuàng)建新用戶優(yōu)惠券邏輯
}
private void handleOrderCreated(OrderCreatedEvent event) {
log.info("Handling order creation: {}", event.getOrderId());
// 訂單創(chuàng)建處理邏輯
}
private void handleOrderPaid(OrderPaidEvent event) {
log.info("Handling order payment: {}", event.getOrderId());
// 訂單支付處理邏輯
}
private void sendWelcomeNotification(UserRegisteredEvent event) {
log.info("Sending welcome notification to user: {}", event.getUserId());
// 發(fā)送歡迎通知邏輯
}
private void sendOrderConfirmationNotification(OrderCreatedEvent event) {
log.info("Sending order confirmation notification for order: {}", event.getOrderId());
// 發(fā)送訂單確認(rèn)通知邏輯
}
private void sendShippingNotification(OrderShippedEvent event) {
log.info("Sending shipping notification for order: {}", event.getOrderId());
// 發(fā)送發(fā)貨通知邏輯
}
}
// 事件類定義
public class UserRegisteredEvent {
private Long userId;
private String email;
private String username;
private Date registrationTime;
// 構(gòu)造函數(shù)、getter、setter...
}
public class UserUpdatedEvent {
private Long userId;
private Map<String, Object> updatedFields;
private Date updateTime;
// 構(gòu)造函數(shù)、getter、setter...
}
public class OrderCreatedEvent {
private Long orderId;
private Long userId;
private BigDecimal totalAmount;
private Date createTime;
// 構(gòu)造函數(shù)、getter、setter...
}
public class OrderPaidEvent {
private Long orderId;
private Long userId;
private BigDecimal paidAmount;
private String paymentMethod;
private Date paidTime;
// 構(gòu)造函數(shù)、getter、setter...
}
public class OrderShippedEvent {
private Long orderId;
private String trackingNumber;
private String shippingCompany;
private Date shippedTime;
// 構(gòu)造函數(shù)、getter、setter...
}總結(jié)
通過(guò)本文的詳細(xì)介紹，我們深入了解了RabbitMQ在Java應(yīng)用中的使用。從基礎(chǔ)概念到高級(jí)特性，從簡(jiǎn)單的點(diǎn)對(duì)點(diǎn)通信到復(fù)雜的事件驅(qū)動(dòng)架構(gòu)，RabbitMQ都能提供強(qiáng)大的支持。
關(guān)鍵要點(diǎn)回顧
- 基礎(chǔ)概念掌握：理解Exchange、Queue、Binding等核心概念是使用RabbitMQ的基礎(chǔ)。
- 消息模式選擇：根據(jù)業(yè)務(wù)場(chǎng)景選擇合適的消息模式，如簡(jiǎn)單隊(duì)列用于任務(wù)分發(fā)，發(fā)布訂閱用于廣播通知。
- Spring Boot集成：使用Spring AMQP可以大大簡(jiǎn)化RabbitMQ的使用，提供了豐富的注解和配置選項(xiàng)。
- 可靠性保證：通過(guò)消息確認(rèn)、死信隊(duì)列、消息持久化等機(jī)制確保消息的可靠傳遞。
- 性能優(yōu)化：合理配置連接池、批量處理、并發(fā)控制等參數(shù)來(lái)優(yōu)化系統(tǒng)性能。
- 監(jiān)控運(yùn)維：建立完善的監(jiān)控和日志體系，及時(shí)發(fā)現(xiàn)和解決問(wèn)題。
最佳實(shí)踐建議
- 設(shè)計(jì)原則：遵循單一職責(zé)原則，每個(gè)隊(duì)列處理特定類型的消息
- 錯(cuò)誤處理：建立完善的錯(cuò)誤處理和重試機(jī)制
- 資源管理：合理管理連接和通道資源，避免資源泄露
- 安全考慮：使用適當(dāng)?shù)恼J(rèn)證和授權(quán)機(jī)制保護(hù)消息安全
- 測(cè)試策略：編寫完善的單元測(cè)試和集成測(cè)試
RabbitMQ作為一個(gè)成熟的消息中間件，在微服務(wù)架構(gòu)、異步處理、系統(tǒng)解耦等場(chǎng)景中發(fā)揮著重要作用。掌握其使用方法和最佳實(shí)踐，將有助于構(gòu)建更加可靠、可擴(kuò)展的分布式系統(tǒng)。
本文涵蓋了RabbitMQ在Java中的主要使用場(chǎng)景和實(shí)踐方法。在實(shí)際應(yīng)用中，還需要根據(jù)具體的業(yè)務(wù)需求和系統(tǒng)架構(gòu)進(jìn)行適當(dāng)?shù)恼{(diào)整和優(yōu)化。
到此這篇關(guān)于Java中的RabbitMQ完全指南的文章就介紹到這了,更多相關(guān)Java RabbitMQ內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
spring security獲取用戶信息的實(shí)現(xiàn)代碼
這篇文章主要介紹了spring security獲取用戶信息的實(shí)現(xiàn)代碼，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-12-12
java反射機(jī)制及beanUtils的實(shí)現(xiàn)原理分析
本文介紹了Java的反射機(jī)制、VO、DTO、PO的概念以及BeanUtils的實(shí)現(xiàn)原理和簡(jiǎn)單示例,通過(guò)反射可以在運(yùn)行時(shí)動(dòng)態(tài)操作類、方法和字段,BeanUtils用于在不同bean之間進(jìn)行屬性復(fù)制2024-12-12
mybatis查詢數(shù)據(jù)賦值到model里面為空的解決
這篇文章主要介紹了mybatis查詢數(shù)據(jù)賦值到model里面為空的解決，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-01-01
這篇文章主要介紹了Jenkins分布式集群配置方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2025-07-07
關(guān)于MyBatis中映射對(duì)象關(guān)系的舉例
這篇文章主要介紹了關(guān)于MyBatis中映射對(duì)象關(guān)系的舉例，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-06-06
Java實(shí)現(xiàn)指定線程執(zhí)行順序的三種方式示例
這篇文章主要介紹了Java實(shí)現(xiàn)指定線程執(zhí)行順序的三種方式,包括通過(guò)共享對(duì)象鎖加上可見(jiàn)變量,通過(guò)主線程Join()以及通過(guò)線程執(zhí)行時(shí)Join()等三種實(shí)現(xiàn)方法,需要的朋友可以參考下2019-01-01
mybatis-plus自動(dòng)裝配時(shí)間失效的解決
本文主要介紹了mybatis-plus自動(dòng)裝配時(shí)間失效，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2022-04-04
Spring運(yùn)行時(shí)動(dòng)態(tài)注冊(cè)bean的方法
這篇文章主要介紹了Spring運(yùn)行時(shí)動(dòng)態(tài)注冊(cè)bean的方法,小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2018-08-08
在日常開(kāi)發(fā)時(shí),可能需要升級(jí)一下自己的jdk版本,那么在升級(jí)之前肯定要先卸載舊的,這篇文章主要介紹了JDK的卸載與安裝步驟的相關(guān)資料,文中通過(guò)圖文介紹的非常詳細(xì),需要的朋友可以參考下2025-07-07![Java中main函數(shù)的String[]?args用法舉例詳解](http://img.jbzj.com/images/xgimg/bcimg9.png)
Java中main函數(shù)的String[]?args用法舉例詳解
這篇文章主要給大家介紹了關(guān)于Java中main函數(shù)的String[]?args用法的相關(guān)資料,JAVA類中main函數(shù)的參數(shù)String[]?args指的是運(yùn)行時(shí)給main函數(shù)傳遞的參數(shù),文中通過(guò)圖文以及代碼介紹的非常詳細(xì),需要的朋友可以參考下2023-12-12