Cao Note消息队列
RabbitMQ 与 Kafka 两大主流消息中间件，含对比与选型建议。本文从 Java 开发者视角出发，涵盖核心概念、Spring 集成、进阶特性和生产实践。
目录
1. MQ 基础
1.1 什么是消息队列
消息队列（Message Queue，简称 MQ）是一种跨进程的通信机制，用于在分布式系统中传递数据。生产者将消息发送到队列，消费者从队列中获取并处理消息。
核心作用：
| 作用 | 说明 |
|---|---|
| 解耦 | 生产者和消费者各自独立演进，不直接依赖对方的接口 |
| 异步 | 非阻塞通信，生产者发送后立即返回，消费者异步处理 |
| 削峰填谷 | 瞬时流量先进入队列，消费者按自身能力慢慢消费，避免系统被突发流量冲垮 |
1.2 核心概念
| 概念 | 说明 |
|---|---|
| Producer（生产者） | 发送消息的一方 |
| Consumer（消费者） | 接收并处理消息的一方 |
| Broker（代理/中间件服务器） | 消息队列的服务端，负责存储和转发消息 |
| Topic（主题） | 消息的逻辑分类，Kafka 中核心概念 |
| Queue（队列） | 消息的存储载体，RabbitMQ 中核心概念 |
| Message（消息） | 传输的数据单元，通常包含业务数据和元信息 |
1.3 消息模型
点对点模型（Point-to-Point）
Producer --> Queue --> ConsumerProducer --> Queue --> Consumer- 一条消息只会被一个消费者消费
- 消息消费后从队列中移除
- 典型代表：RabbitMQ 的直连模式
发布订阅模型（Publish / Subscribe）
Producer --> Topic --> Subscriber A
--> Subscriber B
--> Subscriber CProducer --> Topic --> Subscriber A
--> Subscriber B
--> Subscriber C- 一条消息可以被多个消费者订阅
- 发布者不直接与订阅者通信
- 典型代表：Kafka 的 Topic & Consumer Group
2. RabbitMQ（重点）
RabbitMQ 是基于 AMQP 0-9-1 协议的开源消息中间件，使用 Erlang 语言开发。在 Java 生态中通过 Spring AMQP / RabbitMQ Java Client 集成。
2.1 基础概念
核心组件
| 组件 | 说明 |
|---|---|
| Exchange（交换机） | 生产者发送消息的目标，根据路由规则将消息分发到队列 |
| Queue（队列） | 存储消息的缓冲区，消费者从队列拉取消息 |
| Binding（绑定） | 将 Exchange 与 Queue 关联，并指定 Routing Key |
| Routing Key（路由键） | 消息携带的标签，Exchange 根据它和 Binding Key 匹配决定投递到哪个队列 |
| Virtual Host（虚拟主机） | 逻辑隔离单元，一个 Broker 下可建多个 VHost，每个 VHost 有自己的 Exchange/Queue/Binding |
消息流转流程
Producer --> Exchange --> (Binding) --> Queue --> ConsumerProducer --> Exchange --> (Binding) --> Queue --> Consumer- 生产者发送消息到 Exchange
- Exchange 根据 类型 和 Routing Key 匹配 Binding
- 消息被路由到一个或多个 Queue
- 消费者从 Queue 中拉取或等待推送消息
2.2 Exchange 类型
RabbitMQ 提供四种 Exchange 类型，理解它们是使用 RabbitMQ 的基础。
Direct Exchange（直连交换机）
路由规则：Routing Key 完全匹配 Binding Key。
java
// 声明 Direct Exchange
channel.exchangeDeclare("direct.exchange", BuiltinExchangeType.DIRECT);
// 声明队列
channel.queueDeclare("order.queue", true, false, false, null);
channel.queueDeclare("payment.queue", true, false, false, null);
// 绑定队列到 Exchange，指定 Binding Key
channel.queueBind("order.queue", "direct.exchange", "order");
channel.queueBind("payment.queue", "direct.exchange", "payment");
// 发送消息：Routing Key = "order" --> order.queue
channel.basicPublish("direct.exchange", "order", null, "订单消息".getBytes());// 声明 Direct Exchange
channel.exchangeDeclare("direct.exchange", BuiltinExchangeType.DIRECT);
// 声明队列
channel.queueDeclare("order.queue", true, false, false, null);
channel.queueDeclare("payment.queue", true, false, false, null);
// 绑定队列到 Exchange，指定 Binding Key
channel.queueBind("order.queue", "direct.exchange", "order");
channel.queueBind("payment.queue", "direct.exchange", "payment");
// 发送消息：Routing Key = "order" --> order.queue
channel.basicPublish("direct.exchange", "order", null, "订单消息".getBytes());适用场景：根据业务类型精确路由，如日志级别（info / warn / error）。
Topic Exchange（主题交换机）
路由规则：Routing Key 按通配符匹配 Binding Key。
*：匹配一个单词#：匹配零个或多个单词
java
channel.exchangeDeclare("topic.exchange", BuiltinExchangeType.TOPIC);
channel.queueBind("order.queue", "topic.exchange", "order.#");
channel.queueBind("log.queue", "topic.exchange", "log.*.error");
// 匹配 order.# --> order.queue
channel.basicPublish("topic.exchange", "order.created", null, "订单创建".getBytes());
// 匹配 log.*.error --> log.queue
channel.basicPublish("topic.exchange", "log.service.error", null, "服务异常".getBytes());channel.exchangeDeclare("topic.exchange", BuiltinExchangeType.TOPIC);
channel.queueBind("order.queue", "topic.exchange", "order.#");
channel.queueBind("log.queue", "topic.exchange", "log.*.error");
// 匹配 order.# --> order.queue
channel.basicPublish("topic.exchange", "order.created", null, "订单创建".getBytes());
// 匹配 log.*.error --> log.queue
channel.basicPublish("topic.exchange", "log.service.error", null, "服务异常".getBytes());适用场景：按业务分类的多级路由。
Fanout Exchange（扇出交换机）
路由规则：忽略 Routing Key，广播到所有绑定的队列。
java
channel.exchangeDeclare("fanout.exchange", BuiltinExchangeType.FANOUT);
channel.queueBind("queue.A", "fanout.exchange", "");
channel.queueBind("queue.B", "fanout.exchange", "");
// 所有绑定队列都会收到消息
channel.basicPublish("fanout.exchange", "", null, "广播消息".getBytes());channel.exchangeDeclare("fanout.exchange", BuiltinExchangeType.FANOUT);
channel.queueBind("queue.A", "fanout.exchange", "");
channel.queueBind("queue.B", "fanout.exchange", "");
// 所有绑定队列都会收到消息
channel.basicPublish("fanout.exchange", "", null, "广播消息".getBytes());适用场景：全局通知、配置刷新、缓存失效等广播场景。
Headers Exchange（头部交换机）
路由规则：匹配消息的 headers 属性（键值对），不依赖 Routing Key。
java
Map<String, Object> headers = new HashMap<>();
headers.put("x-match", "all"); // all = 所有键匹配; any = 任一匹配
headers.put("format", "pdf");
headers.put("type", "report");
channel.exchangeDeclare("headers.exchange", BuiltinExchangeType.HEADERS);
channel.queueBind("pdf.report.queue", "headers.exchange", "", headers);
// 发送消息，headers 包含 format=pdf 和 type=report
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.headers(Map.of("format", "pdf", "type", "report"))
.build();
channel.basicPublish("headers.exchange", "", props, "头部匹配消息".getBytes());Map<String, Object> headers = new HashMap<>();
headers.put("x-match", "all"); // all = 所有键匹配; any = 任一匹配
headers.put("format", "pdf");
headers.put("type", "report");
channel.exchangeDeclare("headers.exchange", BuiltinExchangeType.HEADERS);
channel.queueBind("pdf.report.queue", "headers.exchange", "", headers);
// 发送消息，headers 包含 format=pdf 和 type=report
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.headers(Map.of("format", "pdf", "type", "report"))
.build();
channel.basicPublish("headers.exchange", "", props, "头部匹配消息".getBytes());2.3 Spring AMQP
Spring AMQP 提供 RabbitTemplate 和 @RabbitListener 两大核心 API，极大地简化了 RabbitMQ 在 Spring Boot 中的使用。
依赖配置
xml
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency><dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>application.yml 配置
yaml
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
# 发布确认
publisher-confirm-type: correlated
# 发布回退
publisher-returns: true
template:
mandatory: true
listener:
simple:
# 手动 ack
acknowledge-mode: manual
# 预取数量
prefetch: 1
retry:
enabled: true
max-attempts: 3
initial-interval: 1000spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
# 发布确认
publisher-confirm-type: correlated
# 发布回退
publisher-returns: true
template:
mandatory: true
listener:
simple:
# 手动 ack
acknowledge-mode: manual
# 预取数量
prefetch: 1
retry:
enabled: true
max-attempts: 3
initial-interval: 1000Java 配置
java
@Configuration
@EnableRabbit
public class RabbitConfig {
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
// 设置 JSON 序列化
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
// 开启 mandatory（消息无法路由时触发 ReturnCallback）
rabbitTemplate.setMandatory(true);
// 确认回调
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
System.out.println("消息发送成功: " + correlationData);
} else {
System.err.println("消息发送失败: " + cause);
}
});
// 回退回调
rabbitTemplate.setReturnsCallback(returned -> {
System.err.println("消息被退回: " + returned.getMessage()
+ ", replyCode: " + returned.getReplyCode()
+ ", replyText: " + returned.getReplyText()
+ ", exchange: " + returned.getExchange()
+ ", routingKey: " + returned.getRoutingKey());
});
return rabbitTemplate;
}
@Bean
public Jackson2JsonMessageConverter messageConverter() {
ObjectMapper mapper = new ObjectMapper();
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
return new Jackson2JsonMessageConverter(mapper);
}
@Bean
public Queue orderQueue() {
return QueueBuilder.durable("order.queue")
.deadLetterExchange("dlx.exchange")
.deadLetterRoutingKey("order.dead")
.ttl(30000) // 30 秒未消费进入死信
.maxLength(10000)
.build();
}
@Bean
public DirectExchange orderExchange() {
return new DirectExchange("order.exchange");
}
@Bean
public Binding orderBinding() {
return BindingBuilder.bind(orderQueue())
.to(orderExchange())
.with("order.create");
}
}@Configuration
@EnableRabbit
public class RabbitConfig {
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
// 设置 JSON 序列化
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
// 开启 mandatory（消息无法路由时触发 ReturnCallback）
rabbitTemplate.setMandatory(true);
// 确认回调
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
System.out.println("消息发送成功: " + correlationData);
} else {
System.err.println("消息发送失败: " + cause);
}
});
// 回退回调
rabbitTemplate.setReturnsCallback(returned -> {
System.err.println("消息被退回: " + returned.getMessage()
+ ", replyCode: " + returned.getReplyCode()
+ ", replyText: " + returned.getReplyText()
+ ", exchange: " + returned.getExchange()
+ ", routingKey: " + returned.getRoutingKey());
});
return rabbitTemplate;
}
@Bean
public Jackson2JsonMessageConverter messageConverter() {
ObjectMapper mapper = new ObjectMapper();
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
return new Jackson2JsonMessageConverter(mapper);
}
@Bean
public Queue orderQueue() {
return QueueBuilder.durable("order.queue")
.deadLetterExchange("dlx.exchange")
.deadLetterRoutingKey("order.dead")
.ttl(30000) // 30 秒未消费进入死信
.maxLength(10000)
.build();
}
@Bean
public DirectExchange orderExchange() {
return new DirectExchange("order.exchange");
}
@Bean
public Binding orderBinding() {
return BindingBuilder.bind(orderQueue())
.to(orderExchange())
.with("order.create");
}
}发送消息
java
@Service
public class OrderProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 发送订单创建消息
*/
public void sendOrderCreated(OrderDTO order) {
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 设置回调数据
correlationData.getFuture().whenComplete((confirm, ex) -> {
if (ex != null) {
System.err.println("确认异常: " + ex.getMessage());
} else if (confirm.isAck()) {
System.out.println("Broker 已确认: " + correlationData.getId());
} else {
System.err.println("Broker 未确认: " + confirm.getReason());
}
});
rabbitTemplate.convertAndSend("order.exchange", "order.create", order, correlationData);
}
}@Service
public class OrderProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 发送订单创建消息
*/
public void sendOrderCreated(OrderDTO order) {
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// 设置回调数据
correlationData.getFuture().whenComplete((confirm, ex) -> {
if (ex != null) {
System.err.println("确认异常: " + ex.getMessage());
} else if (confirm.isAck()) {
System.out.println("Broker 已确认: " + correlationData.getId());
} else {
System.err.println("Broker 未确认: " + confirm.getReason());
}
});
rabbitTemplate.convertAndSend("order.exchange", "order.create", order, correlationData);
}
}接收消息
java
@Component
public class OrderConsumer {
private static final Logger log = LoggerFactory.getLogger(OrderConsumer.class);
@RabbitListener(queues = "order.queue", concurrency = "3-5")
public void handleOrder(
@Payload OrderDTO order,
@Headers Map<String, Object> headers,
Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
try {
log.info("收到订单消息: {}", order);
// 业务处理
processOrder(order);
// 手动确认
channel.basicAck(deliveryTag, false);
} catch (BusinessException e) {
log.warn("业务异常，拒绝消息: {}", e.getMessage());
// 拒绝并重新入队
channel.basicNack(deliveryTag, false, true);
} catch (Exception e) {
log.error("处理失败，拒绝消息不入队: {}", e.getMessage());
// 拒绝且不入队（进入死信队列）
channel.basicNack(deliveryTag, false, false);
}
}
@RabbitListener(queues = "order.dlq.queue")
public void handleDeadLetter(OrderDTO order) {
log.warn("处理死信消息: {}", order);
// 死信处理：记录日志、告警、补偿等
}
}@Component
public class OrderConsumer {
private static final Logger log = LoggerFactory.getLogger(OrderConsumer.class);
@RabbitListener(queues = "order.queue", concurrency = "3-5")
public void handleOrder(
@Payload OrderDTO order,
@Headers Map<String, Object> headers,
Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
try {
log.info("收到订单消息: {}", order);
// 业务处理
processOrder(order);
// 手动确认
channel.basicAck(deliveryTag, false);
} catch (BusinessException e) {
log.warn("业务异常，拒绝消息: {}", e.getMessage());
// 拒绝并重新入队
channel.basicNack(deliveryTag, false, true);
} catch (Exception e) {
log.error("处理失败，拒绝消息不入队: {}", e.getMessage());
// 拒绝且不入队（进入死信队列）
channel.basicNack(deliveryTag, false, false);
}
}
@RabbitListener(queues = "order.dlq.queue")
public void handleDeadLetter(OrderDTO order) {
log.warn("处理死信消息: {}", order);
// 死信处理：记录日志、告警、补偿等
}
}2.4 进阶特性
2.4.1 死信队列（DLX + TTL）
死信（Dead Letter）：消息在队列中出现以下情况时会变成死信：
- 消费者使用
basicNack/basicReject且 requeue = false - 消息 TTL 到期
- 队列达到最大长度
死信交换机（DLX）：当消息变成死信后，会被重新发布到指定的 DLX，再路由到死信队列。
java
// 普通队列，绑定死信交换机
@Bean
public Queue businessQueue() {
return QueueBuilder.durable("business.queue")
.deadLetterExchange("dlx.exchange") // 死信交换机
.deadLetterRoutingKey("business.dead") // 死信路由键
.ttl(60000) // 消息 TTL：60 秒
.maxLength(5000) // 队列最大长度
.build();
}
// 死信交换机
@Bean
public DirectExchange dlxExchange() {
return new DirectExchange("dlx.exchange");
}
// 死信队列
@Bean
public Queue dlxQueue() {
return QueueBuilder.durable("dlx.queue").build();
}
// 绑定死信队列到死信交换机
@Bean
public Binding dlxBinding() {
return BindingBuilder.bind(dlxQueue())
.to(dlxExchange())
.with("business.dead");
}// 普通队列，绑定死信交换机
@Bean
public Queue businessQueue() {
return QueueBuilder.durable("business.queue")
.deadLetterExchange("dlx.exchange") // 死信交换机
.deadLetterRoutingKey("business.dead") // 死信路由键
.ttl(60000) // 消息 TTL：60 秒
.maxLength(5000) // 队列最大长度
.build();
}
// 死信交换机
@Bean
public DirectExchange dlxExchange() {
return new DirectExchange("dlx.exchange");
}
// 死信队列
@Bean
public Queue dlxQueue() {
return QueueBuilder.durable("dlx.queue").build();
}
// 绑定死信队列到死信交换机
@Bean
public Binding dlxBinding() {
return BindingBuilder.bind(dlxQueue())
.to(dlxExchange())
.with("business.dead");
}延迟队列实现：利用 DLX + TTL 可以模拟延迟队列效果——消息先进入一个无消费者的队列，TTL 到期后成为死信，被投递到实际处理队列。
java
// 延迟队列（无消费者，等待 TTL 到期）
@Bean
public Queue delayQueue() {
return QueueBuilder.durable("delay.queue")
.deadLetterExchange("business.exchange")
.deadLetterRoutingKey("business.process")
.ttl(30000) // 30 秒延迟
.build();
}
// 实际处理队列
@Bean
public Queue processQueue() {
return new Queue("process.queue");
}// 延迟队列（无消费者，等待 TTL 到期）
@Bean
public Queue delayQueue() {
return QueueBuilder.durable("delay.queue")
.deadLetterExchange("business.exchange")
.deadLetterRoutingKey("business.process")
.ttl(30000) // 30 秒延迟
.build();
}
// 实际处理队列
@Bean
public Queue processQueue() {
return new Queue("process.queue");
}注意：同一个队列中的消息 TTL 相同。如果需要不同延迟时间，需要为不同的延迟时间创建不同的队列，或者使用延迟消息插件。
2.4.2 延迟消息（插件方式）
RabbitMQ 官方提供了延迟消息插件 rabbitmq_delayed_message_exchange，安装后可以使用延迟交换机，按消息级别设置延迟时间。
安装插件（在 RabbitMQ 服务器上执行）：
bash
# 下载插件
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/latest/download/rabbitmq_delayed_message_exchange-3.13.0.ez
# 复制到 plugins 目录
cp rabbitmq_delayed_message_exchange-3.13.0.ez /usr/lib/rabbitmq/lib/rabbitmq_server-3.13.0/plugins/
# 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
# 重启 RabbitMQ
systemctl restart rabbitmq-server# 下载插件
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/latest/download/rabbitmq_delayed_message_exchange-3.13.0.ez
# 复制到 plugins 目录
cp rabbitmq_delayed_message_exchange-3.13.0.ez /usr/lib/rabbitmq/lib/rabbitmq_server-3.13.0/plugins/
# 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
# 重启 RabbitMQ
systemctl restart rabbitmq-serverJava 代码使用延迟交换机：
java
@Bean
public CustomExchange delayExchange() {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct"); // 底层行为类似 direct exchange
return new CustomExchange("delay.exchange", "x-delayed-message", true, false, args);
}
@Bean
public Binding delayBinding() {
return BindingBuilder.bind(delayQueue())
.to(delayExchange())
.with("delay.routing.key")
.noargs();
}@Bean
public CustomExchange delayExchange() {
Map<String, Object> args = new HashMap<>();
args.put("x-delayed-type", "direct"); // 底层行为类似 direct exchange
return new CustomExchange("delay.exchange", "x-delayed-message", true, false, args);
}
@Bean
public Binding delayBinding() {
return BindingBuilder.bind(delayQueue())
.to(delayExchange())
.with("delay.routing.key")
.noargs();
}java
@Service
public class DelayMessageProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendDelayMessage(OrderDTO order, long delayMillis) {
MessagePostProcessor processor = message -> {
// 设置延迟时间（毫秒）
message.getMessageProperties().setDelayLong(delayMillis);
return message;
};
rabbitTemplate.convertAndSend(
"delay.exchange", "delay.routing.key", order, processor);
}
}@Service
public class DelayMessageProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendDelayMessage(OrderDTO order, long delayMillis) {
MessagePostProcessor processor = message -> {
// 设置延迟时间（毫秒）
message.getMessageProperties().setDelayLong(delayMillis);
return message;
};
rabbitTemplate.convertAndSend(
"delay.exchange", "delay.routing.key", order, processor);
}
}java
@Component
public class DelayMessageConsumer {
@RabbitListener(queues = "delay.queue")
public void handle(OrderDTO order) {
System.out.println("收到延迟消息: " + order + ", 时间: " + LocalDateTime.now());
}
}@Component
public class DelayMessageConsumer {
@RabbitListener(queues = "delay.queue")
public void handle(OrderDTO order) {
System.out.println("收到延迟消息: " + order + ", 时间: " + LocalDateTime.now());
}
}2.4.3 消息确认
生产端确认：
| 类型 | 说明 |
|---|---|
| Publisher Confirm | 消息成功到达 Exchange 后 Broker 回传确认 |
| Return Callback | 消息无法路由到任何队列时的回退通知 |
yaml
spring:
rabbitmq:
publisher-confirm-type: correlated # 异步确认
publisher-returns: true # 开启回退
template:
mandatory: true # 必须开启 mandatoryspring:
rabbitmq:
publisher-confirm-type: correlated # 异步确认
publisher-returns: true # 开启回退
template:
mandatory: true # 必须开启 mandatory消费端确认：
| 模式 | 说明 |
|---|---|
| 自动 ack（AUTO） | 框架自动确认，可能丢消息 |
| 手动 ack（MANUAL） | 业务处理成功后才确认，安全 |
| NONE（无确认） | 不确认，消息立即丢弃（不推荐） |
yaml
spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual # 手动确认
prefetch: 1 # 每次只取一条spring:
rabbitmq:
listener:
simple:
acknowledge-mode: manual # 手动确认
prefetch: 1 # 每次只取一条java
@RabbitListener(queues = "business.queue")
public void handleMessage(
@Payload String msg,
Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
try {
// 业务处理
doBusiness(msg);
// 确认
channel.basicAck(deliveryTag, false);
} catch (RetryableException e) {
// 可重试的异常，拒绝并重新入队
channel.basicNack(deliveryTag, false, true);
} catch (FatalException e) {
// 不可重试的异常，拒绝不入队
channel.basicNack(deliveryTag, false, false);
}
}@RabbitListener(queues = "business.queue")
public void handleMessage(
@Payload String msg,
Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
try {
// 业务处理
doBusiness(msg);
// 确认
channel.basicAck(deliveryTag, false);
} catch (RetryableException e) {
// 可重试的异常，拒绝并重新入队
channel.basicNack(deliveryTag, false, true);
} catch (FatalException e) {
// 不可重试的异常，拒绝不入队
channel.basicNack(deliveryTag, false, false);
}
}QoS 预取（Prefetch）：控制单个消费者未确认消息的最大数量。prefetch=1 确保消费者处理完一条再取下一条，实现公平调度。
yaml
spring:
rabbitmq:
listener:
simple:
prefetch: 1spring:
rabbitmq:
listener:
simple:
prefetch: 12.4.4 批量发送
java
@Bean
public RabbitTemplate batchRabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(new Jackson2JsonMessageConverter());
// 启用批量化（按大小或条数触发）
BatchingStrategy strategy = new SimpleBatchingStrategy(
100, // 批次大小上限（字节）
10, // 每批最大消息数
5000 // 超时时间（毫秒）
);
template.setBatchingStrategy(strategy);
return template;
}@Bean
public RabbitTemplate batchRabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(new Jackson2JsonMessageConverter());
// 启用批量化（按大小或条数触发）
BatchingStrategy strategy = new SimpleBatchingStrategy(
100, // 批次大小上限（字节）
10, // 每批最大消息数
5000 // 超时时间（毫秒）
);
template.setBatchingStrategy(strategy);
return template;
}2.5 RabbitMQ 集群
普通集群（Classic Cluster）
- 元数据同步：所有节点同步 Exchange、Queue、Binding 的元信息
- 消息数据非共享：消息只存储在其归属节点上
- 节点间转发：消费者连接非存储节点时，该节点从存储节点拉取消息
bash
# 假设已搭建三节点集群
# 节点 1：rabbit1
# 节点 2：rabbit2
# 节点 3：rabbit3
# 将所有节点加入集群
# 在 rabbit2 上执行
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@rabbit1
rabbitmqctl start_app
# 在 rabbit3 上执行
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@rabbit1
rabbitmqctl start_app
# 查看集群状态
rabbitmqctl cluster_status# 假设已搭建三节点集群
# 节点 1：rabbit1
# 节点 2：rabbit2
# 节点 3：rabbit3
# 将所有节点加入集群
# 在 rabbit2 上执行
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@rabbit1
rabbitmqctl start_app
# 在 rabbit3 上执行
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@rabbit1
rabbitmqctl start_app
# 查看集群状态
rabbitmqctl cluster_status镜像队列（Mirrored Queues）
高可用模式：将队列数据在集群节点间复制，一个节点宕机后其他节点仍提供服务。
bash
# 使用策略配置镜像队列（推荐方式）
# 将所有以 "ha." 开头的队列镜像到所有节点
rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}' --priority 1
# 镜像到 2 个节点
rabbitmqctl set_policy ha-two "^ha\." '{"ha-mode":"exactly","ha-params":2}' --priority 1# 使用策略配置镜像队列（推荐方式）
# 将所有以 "ha." 开头的队列镜像到所有节点
rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}' --priority 1
# 镜像到 2 个节点
rabbitmqctl set_policy ha-two "^ha\." '{"ha-mode":"exactly","ha-params":2}' --priority 1Quorum Queue
现代高可用方案（RabbitMQ 3.8+），基于 Raft 共识算法，替代镜像队列。
java
@Bean
public Queue quorumQueue() {
return QueueBuilder.durable("quorum.queue")
.quorum() // 声明为 Quorum Queue
.deliveryLimit(3) // 最大投递次数，超限进入死信
.build();
}@Bean
public Queue quorumQueue() {
return QueueBuilder.durable("quorum.queue")
.quorum() // 声明为 Quorum Queue
.deliveryLimit(3) // 最大投递次数，超限进入死信
.build();
}镜像队列 vs Quorum Queue：
| 对比维度 | 镜像队列 | Quorum Queue |
|---|---|---|
| 算法 | GM（Guaranteed Multicast） | Raft 共识 |
| 一致性 | 最终一致 | 强一致 |
| 数据安全 | 脑裂可能丢数据 | Leader 故障自动选举，数据不丢 |
| 性能 | 高（异步复制） | 中（需多数派写入） |
| 建议 | 旧项目兼容 | 新项目首选 |
3. Kafka（重点）
Apache Kafka 是分布式流处理平台，核心设计为高吞吐、持久化、可分区、可复制的发布订阅系统。在 Java 生态中通过 Spring Kafka 集成。
3.1 基础概念
核心组件
| 组件 | 说明 |
|---|---|
| Topic（主题） | 消息的逻辑分类，对应 RabbitMQ 的 Exchange + Routing Key 组合 |
| Partition（分区） | Topic 的物理分片，每个分区是一个有序的不可变日志文件 |
| Offset（偏移量） | 分区内消息的唯一递增序号，由 Kafka 自动分配 |
| Consumer Group（消费者组） | 一组消费者的逻辑分组，一条消息只能被组内一个消费者消费 |
| Broker（代理节点） | Kafka 服务进程 |
| Controller（控制器） | 集群中负责管理分区 Leader 选举的节点 |
| ISR（In-Sync Replica） | 与 Leader 保持同步的副本集合 |
| Leader / Follower | 分区的读写都由 Leader 处理，Follower 被动复制 |
消息在 Kafka 中的流转
Producer --> Topic (Partition 0) --> Offset 0, 1, 2, 3...
--> Topic (Partition 1) --> Offset 0, 1, 2, 3...
--> Topic (Partition 2) --> Offset 0, 1, 2, 3...
Consumer Group A --> Consumer A-0 (消费 Partition 0)
--> Consumer A-1 (消费 Partition 1, 2)
Consumer Group B --> Consumer B-0 (消费 Partition 0, 1, 2)Producer --> Topic (Partition 0) --> Offset 0, 1, 2, 3...
--> Topic (Partition 1) --> Offset 0, 1, 2, 3...
--> Topic (Partition 2) --> Offset 0, 1, 2, 3...
Consumer Group A --> Consumer A-0 (消费 Partition 0)
--> Consumer A-1 (消费 Partition 1, 2)
Consumer Group B --> Consumer B-0 (消费 Partition 0, 1, 2)关键特性：
- 同一分区的消息严格有序（追加到日志尾部）
- 消息持久化到磁盘，可配置保留时长或大小
- 消费者通过 offset 控制消费位置，可重放消费
- 吞吐量随分区数线性扩展
Kafka 架构
Producer A Producer B
| |
+------+---+----------+--------+
| | |
+--------+ +--------+ +--------+
|Broker 1 | |Broker 2 | ... |Broker N |
|Leader P0| |Leader P1| |Leader P2|
|Follower | |Follower | |Follower |
+--------+ +--------+ +--------+
| \ /
+------+----------+------+
| Zookeeper | (或 KRaft)
+--------------------------+
+--------------------------+
| Consumer Group A |
| Consumer A-0 A-1 |
+--------------------------+ Producer A Producer B
| |
+------+---+----------+--------+
| | |
+--------+ +--------+ +--------+
|Broker 1 | |Broker 2 | ... |Broker N |
|Leader P0| |Leader P1| |Leader P2|
|Follower | |Follower | |Follower |
+--------+ +--------+ +--------+
| \ /
+------+----------+------+
| Zookeeper | (或 KRaft)
+--------------------------+
+--------------------------+
| Consumer Group A |
| Consumer A-0 A-1 |
+--------------------------+3.2 Spring Kafka
Spring Kafka 提供 KafkaTemplate 和 @KafkaListener 两大核心 API。
依赖配置
xml
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency><dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>application.yml 配置
yaml
spring:
kafka:
bootstrap-servers: localhost:9092,localhost:9093,localhost:9094
# 生产者配置
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
acks: all
retries: 3
compression-type: snappy
batch-size: 16384
linger-ms: 5
buffer-memory: 33554432
# 幂等生产者
properties:
enable:
idempotence: true
max:
in:
flight:
requests:
per:
connection: 5
# 消费者配置
consumer:
group-id: order-service-group
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
auto-offset-reset: earliest
enable-auto-commit: false
max-poll-records: 500
properties:
spring:
json:
trusted:
packages: '*'
isolation:
level: read_committed
# 监听器配置
listener:
ack-mode: manual_immediate
concurrency: 3
missing-topics-fatal: falsespring:
kafka:
bootstrap-servers: localhost:9092,localhost:9093,localhost:9094
# 生产者配置
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
acks: all
retries: 3
compression-type: snappy
batch-size: 16384
linger-ms: 5
buffer-memory: 33554432
# 幂等生产者
properties:
enable:
idempotence: true
max:
in:
flight:
requests:
per:
connection: 5
# 消费者配置
consumer:
group-id: order-service-group
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
auto-offset-reset: earliest
enable-auto-commit: false
max-poll-records: 500
properties:
spring:
json:
trusted:
packages: '*'
isolation:
level: read_committed
# 监听器配置
listener:
ack-mode: manual_immediate
concurrency: 3
missing-topics-fatal: false发送消息
java
@Service
public class OrderEventProducer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
private static final Logger log = LoggerFactory.getLogger(OrderEventProducer.class);
/**
* 发送订单创建事件
*/
public void sendOrderCreated(OrderEvent event) {
// 使用订单 ID 作为 key，确保同一订单进入同一分区
ListenableFuture<SendResult<String, Object>> future =
kafkaTemplate.send("order-events", event.getOrderId(), event);
future.whenComplete((result, ex) -> {
if (ex == null) {
RecordMetadata metadata = result.getRecordMetadata();
log.info("消息发送成功: topic={}, partition={}, offset={}, key={}",
metadata.topic(), metadata.partition(),
metadata.offset(), event.getOrderId());
} else {
log.error("消息发送失败: key={}, error={}", event.getOrderId(), ex.getMessage());
// 可以写入本地重试表或发送到死信
handleSendFailure(event, ex);
}
});
}
/**
* 发送带回调的异步消息
*/
public void sendWithCallback(OrderEvent event) {
kafkaTemplate.send("order-events", event.getOrderId(), event)
.addCallback(new ListenableFutureCallback<>() {
@Override
public void onFailure(Throwable ex) {
log.error("发送失败: {}", ex.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> result) {
log.info("发送成功: offset={}", result.getRecordMetadata().offset());
}
});
}
/**
* 同步发送（不推荐用于核心业务，会阻塞）
*/
public void sendSync(OrderEvent event) throws ExecutionException, InterruptedException {
SendResult<String, Object> result =
kafkaTemplate.send("order-events", event.getOrderId(), event).get();
log.info("同步发送成功: offset={}", result.getRecordMetadata().offset());
}
}@Service
public class OrderEventProducer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
private static final Logger log = LoggerFactory.getLogger(OrderEventProducer.class);
/**
* 发送订单创建事件
*/
public void sendOrderCreated(OrderEvent event) {
// 使用订单 ID 作为 key，确保同一订单进入同一分区
ListenableFuture<SendResult<String, Object>> future =
kafkaTemplate.send("order-events", event.getOrderId(), event);
future.whenComplete((result, ex) -> {
if (ex == null) {
RecordMetadata metadata = result.getRecordMetadata();
log.info("消息发送成功: topic={}, partition={}, offset={}, key={}",
metadata.topic(), metadata.partition(),
metadata.offset(), event.getOrderId());
} else {
log.error("消息发送失败: key={}, error={}", event.getOrderId(), ex.getMessage());
// 可以写入本地重试表或发送到死信
handleSendFailure(event, ex);
}
});
}
/**
* 发送带回调的异步消息
*/
public void sendWithCallback(OrderEvent event) {
kafkaTemplate.send("order-events", event.getOrderId(), event)
.addCallback(new ListenableFutureCallback<>() {
@Override
public void onFailure(Throwable ex) {
log.error("发送失败: {}", ex.getMessage());
}
@Override
public void onSuccess(SendResult<String, Object> result) {
log.info("发送成功: offset={}", result.getRecordMetadata().offset());
}
});
}
/**
* 同步发送（不推荐用于核心业务，会阻塞）
*/
public void sendSync(OrderEvent event) throws ExecutionException, InterruptedException {
SendResult<String, Object> result =
kafkaTemplate.send("order-events", event.getOrderId(), event).get();
log.info("同步发送成功: offset={}", result.getRecordMetadata().offset());
}
}消费消息
java
@Component
public class OrderEventConsumer {
private static final Logger log = LoggerFactory.getLogger(OrderEventConsumer.class);
/**
* 简单消费：单条消息处理
*/
@KafkaListener(
topics = "order-events",
groupId = "order-service-group",
containerFactory = "kafkaListenerContainerFactory"
)
public void handleOrderEvent(
@Payload OrderEvent event,
@Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
@Header(KafkaHeaders.OFFSET) long offset,
Acknowledgment acknowledgment) {
try {
log.info("收到订单事件: partition={}, offset={}, event={}",
partition, offset, event);
// 业务处理
processEvent(event);
// 手动提交 offset
acknowledgment.acknowledge();
} catch (Exception e) {
log.error("处理事件异常: {}", e.getMessage());
// 不提交 offset，下次重新消费
}
}
/**
* 批量消费
*/
@KafkaListener(
topics = "order-events",
groupId = "order-batch-group",
containerFactory = "batchFactory"
)
public void handleBatch(List<OrderEvent> events, Acknowledgment acknowledgment) {
log.info("批量收到 {} 条事件", events.size());
for (OrderEvent event : events) {
try {
processEvent(event);
} catch (Exception e) {
log.error("批量处理异常: {}", e.getMessage());
// 批量模式异常处理策略由业务决定
}
}
acknowledgment.acknowledge();
}
@Autowired
private OrderEventProducer eventProducer;
/**
* 兜底处理：发送到 DLT
*/
private void processEvent(OrderEvent event) {
try {
orderService.process(event);
} catch (Exception e) {
// 发送到死信主题
eventProducer.sendToDlt(event, e.getMessage());
}
}
}@Component
public class OrderEventConsumer {
private static final Logger log = LoggerFactory.getLogger(OrderEventConsumer.class);
/**
* 简单消费：单条消息处理
*/
@KafkaListener(
topics = "order-events",
groupId = "order-service-group",
containerFactory = "kafkaListenerContainerFactory"
)
public void handleOrderEvent(
@Payload OrderEvent event,
@Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
@Header(KafkaHeaders.OFFSET) long offset,
Acknowledgment acknowledgment) {
try {
log.info("收到订单事件: partition={}, offset={}, event={}",
partition, offset, event);
// 业务处理
processEvent(event);
// 手动提交 offset
acknowledgment.acknowledge();
} catch (Exception e) {
log.error("处理事件异常: {}", e.getMessage());
// 不提交 offset，下次重新消费
}
}
/**
* 批量消费
*/
@KafkaListener(
topics = "order-events",
groupId = "order-batch-group",
containerFactory = "batchFactory"
)
public void handleBatch(List<OrderEvent> events, Acknowledgment acknowledgment) {
log.info("批量收到 {} 条事件", events.size());
for (OrderEvent event : events) {
try {
processEvent(event);
} catch (Exception e) {
log.error("批量处理异常: {}", e.getMessage());
// 批量模式异常处理策略由业务决定
}
}
acknowledgment.acknowledge();
}
@Autowired
private OrderEventProducer eventProducer;
/**
* 兜底处理：发送到 DLT
*/
private void processEvent(OrderEvent event) {
try {
orderService.process(event);
} catch (Exception e) {
// 发送到死信主题
eventProducer.sendToDlt(event, e.getMessage());
}
}
}KafkaListenerContainerFactory 配置
java
@Configuration
public class KafkaConsumerConfig {
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object>
kafkaListenerContainerFactory(ConsumerFactory<String, Object> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
// 并发消费者数（等于 consumer 线程数）
factory.setConcurrency(3);
// 手动 ack
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
// 批量消费
// factory.setBatchListener(true);
// 异常重试
factory.setCommonErrorHandler(new DefaultErrorHandler(
new FixedBackOff(1000L, 3)
));
return factory;
}
@Bean
public ConsumerFactory<String, Object> consumerFactory(
KafkaProperties properties) {
Map<String, Object> props = properties.buildConsumerProperties();
props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
return new DefaultKafkaConsumerFactory<>(props);
}
}@Configuration
public class KafkaConsumerConfig {
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object>
kafkaListenerContainerFactory(ConsumerFactory<String, Object> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
// 并发消费者数（等于 consumer 线程数）
factory.setConcurrency(3);
// 手动 ack
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
// 批量消费
// factory.setBatchListener(true);
// 异常重试
factory.setCommonErrorHandler(new DefaultErrorHandler(
new FixedBackOff(1000L, 3)
));
return factory;
}
@Bean
public ConsumerFactory<String, Object> consumerFactory(
KafkaProperties properties) {
Map<String, Object> props = properties.buildConsumerProperties();
props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
return new DefaultKafkaConsumerFactory<>(props);
}
}3.3 进阶特性
3.3.1 重试与死信主题
Spring Kafka 通过 @RetryableTopic 注解实现重试机制。
java
@Slf4j
@Component
public class RetryableConsumer {
/**
* 自动重试 + 死信主题
* - 尝试 3 次
* - 重试间隔 1 秒
* - 最终失败进入死信主题
*/
@RetryableTopic(
attempts = "3",
backoff = @Backoff(delay = 1000, multiplier = 2.0),
autoCreateTopics = "true",
dltTopicSuffix = "-dlt",
kafkaTemplate = "kafkaTemplate"
)
@KafkaListener(topics = "retryable-topic", groupId = "retry-group")
public void handleRetryable(String message) {
log.info("收到消息: {}", message);
if (message.contains("error")) {
throw new RuntimeException("模拟处理异常");
}
log.info("处理成功: {}", message);
}
/**
* 死信主题处理器
*/
@DltHandler
public void handleDlt(String message) {
log.error("进入死信: {}, 时间: {}", message, LocalDateTime.now());
// 告警、记录日志、触发补偿
}
}@Slf4j
@Component
public class RetryableConsumer {
/**
* 自动重试 + 死信主题
* - 尝试 3 次
* - 重试间隔 1 秒
* - 最终失败进入死信主题
*/
@RetryableTopic(
attempts = "3",
backoff = @Backoff(delay = 1000, multiplier = 2.0),
autoCreateTopics = "true",
dltTopicSuffix = "-dlt",
kafkaTemplate = "kafkaTemplate"
)
@KafkaListener(topics = "retryable-topic", groupId = "retry-group")
public void handleRetryable(String message) {
log.info("收到消息: {}", message);
if (message.contains("error")) {
throw new RuntimeException("模拟处理异常");
}
log.info("处理成功: {}", message);
}
/**
* 死信主题处理器
*/
@DltHandler
public void handleDlt(String message) {
log.error("进入死信: {}, 时间: {}", message, LocalDateTime.now());
// 告警、记录日志、触发补偿
}
}手动实现重试 + DLT：
java
@Component
public class ManualRetryConsumer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
private static final String DLT_TOPIC = "order-events-dlt";
@KafkaListener(topics = "order-events", groupId = "retry-group")
public void consume(ConsumerRecord<String, OrderEvent> record,
Acknowledgment ack) {
try {
process(record.value());
ack.acknowledge();
} catch (RetryableException e) {
int retryCount = getRetryCount(record.headers());
if (retryCount < 3) {
// 发送到重试主题（延迟主题）
kafkaTemplate.send("order-events-retry", record.key(), record.value());
} else {
// 超过重试次数，发送到 DLT
kafkaTemplate.send(DLT_TOPIC, record.key(), record.value());
}
ack.acknowledge();
}
}
private int getRetryCount(Headers headers) {
Header header = headers.lastHeader("retry-count");
if (header == null) return 0;
return Integer.parseInt(new String(header.value()));
}
private void process(OrderEvent event) {
// 业务处理
}
}@Component
public class ManualRetryConsumer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
private static final String DLT_TOPIC = "order-events-dlt";
@KafkaListener(topics = "order-events", groupId = "retry-group")
public void consume(ConsumerRecord<String, OrderEvent> record,
Acknowledgment ack) {
try {
process(record.value());
ack.acknowledge();
} catch (RetryableException e) {
int retryCount = getRetryCount(record.headers());
if (retryCount < 3) {
// 发送到重试主题（延迟主题）
kafkaTemplate.send("order-events-retry", record.key(), record.value());
} else {
// 超过重试次数，发送到 DLT
kafkaTemplate.send(DLT_TOPIC, record.key(), record.value());
}
ack.acknowledge();
}
}
private int getRetryCount(Headers headers) {
Header header = headers.lastHeader("retry-count");
if (header == null) return 0;
return Integer.parseInt(new String(header.value()));
}
private void process(OrderEvent event) {
// 业务处理
}
}3.3.2 幂等生产者
enable.idempotence=true 保证生产者发送消息的幂等性，避免网络重试导致的消息重复。
yaml
spring:
kafka:
producer:
properties:
enable:
idempotence: true
max:
in:
flight:
requests:
per:
connection: 5spring:
kafka:
producer:
properties:
enable:
idempotence: true
max:
in:
flight:
requests:
per:
connection: 5原理：
- 每个 Producer 启动时被分配一个 Producer ID（PID）
- 每条消息带一个自增的 Sequence Number
- Broker 端去重：同一 PID + Sequence Number 的消息只写入一次
3.3.3 事务
Kafka 事务支持跨分区和跨主题的原子写入，实现 Exactly-Once Semantics（EOS）。
java
@Configuration
public class KafkaTransactionConfig {
@Bean
public KafkaTransactionManager<String, Object> transactionManager(
ProducerFactory<String, Object> producerFactory) {
return new KafkaTransactionManager<>(producerFactory);
}
@Bean
public KafkaTemplate<String, Object> kafkaTemplate(
ProducerFactory<String, Object> producerFactory) {
return new KafkaTemplate<>(producerFactory);
}
}@Configuration
public class KafkaTransactionConfig {
@Bean
public KafkaTransactionManager<String, Object> transactionManager(
ProducerFactory<String, Object> producerFactory) {
return new KafkaTransactionManager<>(producerFactory);
}
@Bean
public KafkaTemplate<String, Object> kafkaTemplate(
ProducerFactory<String, Object> producerFactory) {
return new KafkaTemplate<>(producerFactory);
}
}java
@Service
public class TransactionalProducer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
/**
* 事务性发送
* 要么所有消息都成功，要么都失败
*/
@Transactional("kafkaTransactionManager")
public void sendInTransaction(OrderEvent event) {
// 写入订单事件
kafkaTemplate.send("order-events", event.getOrderId(), event);
// 同时写入积分事件
kafkaTemplate.send("points-events", event.getUserId(), buildPointsEvent(event));
// 写入审计日志
kafkaTemplate.send("audit-log", event.getOrderId(), buildAuditLog(event));
// 这三条消息在同一事务中，原子提交
}
/**
* 跨系统事务（Kafka + JDBC）
*/
@Transactional(rollbackFor = Exception.class)
@Autowired
private JdbcTemplate jdbcTemplate;
public void crossSystemTransaction(OrderEvent event) {
// 1. 写数据库
jdbcTemplate.update("INSERT INTO orders (id, user_id, amount) VALUES (?, ?, ?)",
event.getOrderId(), event.getUserId(), event.getAmount());
// 2. 发消息到 Kafka
kafkaTemplate.send("order-events", event.getOrderId(), event);
// 3. 如果 send 失败，数据库操作也会回滚
// 注意：此处需要 ChainedKafkaTransactionManager
}
}@Service
public class TransactionalProducer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
/**
* 事务性发送
* 要么所有消息都成功，要么都失败
*/
@Transactional("kafkaTransactionManager")
public void sendInTransaction(OrderEvent event) {
// 写入订单事件
kafkaTemplate.send("order-events", event.getOrderId(), event);
// 同时写入积分事件
kafkaTemplate.send("points-events", event.getUserId(), buildPointsEvent(event));
// 写入审计日志
kafkaTemplate.send("audit-log", event.getOrderId(), buildAuditLog(event));
// 这三条消息在同一事务中，原子提交
}
/**
* 跨系统事务（Kafka + JDBC）
*/
@Transactional(rollbackFor = Exception.class)
@Autowired
private JdbcTemplate jdbcTemplate;
public void crossSystemTransaction(OrderEvent event) {
// 1. 写数据库
jdbcTemplate.update("INSERT INTO orders (id, user_id, amount) VALUES (?, ?, ?)",
event.getOrderId(), event.getUserId(), event.getAmount());
// 2. 发消息到 Kafka
kafkaTemplate.send("order-events", event.getOrderId(), event);
// 3. 如果 send 失败，数据库操作也会回滚
// 注意：此处需要 ChainedKafkaTransactionManager
}
}事务配置要求：
yaml
spring:
kafka:
producer:
acks: all
properties:
enable:
idempotence: true
transaction:
id: tx-order- # 每个生产者实例有唯一的 transactional.id
consumer:
properties:
isolation:
level: read_committed # 只读取已提交的消息spring:
kafka:
producer:
acks: all
properties:
enable:
idempotence: true
transaction:
id: tx-order- # 每个生产者实例有唯一的 transactional.id
consumer:
properties:
isolation:
level: read_committed # 只读取已提交的消息3.3.4 分区分配策略
当消费者组内成员变更（加入/离开/崩溃）时会触发再均衡（Rebalance），需要重新分配分区。
| 策略 | 说明 | 特点 |
|---|---|---|
| Range | 按主题范围分配（默认） | 可能导致分配不均，一个消费者可能承担过多主题的尾部分区 |
| RoundRobin | 轮询分配 | 相对均匀，但未考虑消费者地理位置 |
| Sticky | 粘性分配 | 尽量保持现有分配不变，减少再均衡时的分区变动 |
| CooperativeSticky | 协作粘性 | 推荐，支持增量再均衡，避免 Stop-The-World |
yaml
spring:
kafka:
consumer:
properties:
partition:
assignment:
strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignorspring:
kafka:
consumer:
properties:
partition:
assignment:
strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor3.3.5 手动提交 Offset
java
@Component
public class ManualOffsetConsumer {
// 方式一：使用 Acknowledgment
@KafkaListener(topics = "manual-offset-topic", groupId = "manual-group")
public void listenWithAck(ConsumerRecord<String, String> record,
Acknowledgment ack) {
try {
process(record);
ack.acknowledge(); // 提交 offset
} catch (Exception e) {
// 不提交，再次消费
}
}
// 方式二：使用 Consumer 直接提交
@KafkaListener(topics = "manual-offset-topic", groupId = "manual-group-v2")
public void listenWithConsumer(ConsumerRecord<String, String> record,
Consumer<?, ?> consumer) {
try {
process(record);
// 同步提交
consumer.commitSync();
} catch (Exception e) {
// 异步提交
consumer.commitAsync((offsets, exception) -> {
if (exception != null) {
log.error("提交 offset 失败: {}", offsets, exception);
}
});
}
}
private void process(ConsumerRecord<String, String> record) {
// 业务处理
}
}@Component
public class ManualOffsetConsumer {
// 方式一：使用 Acknowledgment
@KafkaListener(topics = "manual-offset-topic", groupId = "manual-group")
public void listenWithAck(ConsumerRecord<String, String> record,
Acknowledgment ack) {
try {
process(record);
ack.acknowledge(); // 提交 offset
} catch (Exception e) {
// 不提交，再次消费
}
}
// 方式二：使用 Consumer 直接提交
@KafkaListener(topics = "manual-offset-topic", groupId = "manual-group-v2")
public void listenWithConsumer(ConsumerRecord<String, String> record,
Consumer<?, ?> consumer) {
try {
process(record);
// 同步提交
consumer.commitSync();
} catch (Exception e) {
// 异步提交
consumer.commitAsync((offsets, exception) -> {
if (exception != null) {
log.error("提交 offset 失败: {}", offsets, exception);
}
});
}
}
private void process(ConsumerRecord<String, String> record) {
// 业务处理
}
}**enable-auto-commit 为 false 搭配 ** AckMode 选项：
| AckMode | 说明 |
|---|---|
RECORD | 每处理一条消息就提交 |
BATCH | poll 一次的所有消息处理完提交（默认） |
TIME | 按时间间隔提交 |
COUNT | 按消息数量提交 |
COUNT_TIME | 数量和时间的组合 |
MANUAL | 手动调用 acknowledge() 触发 |
MANUAL_IMMEDIATE | 手动调用后立即提交 |
3.3.6 消费者再均衡
再均衡发生时可以注册回调，执行清理或状态保存操作。
java
@Component
public class RebalanceAwareConsumer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String>
rebalanceAwareFactory(ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.getContainerProperties().setConsumerRebalanceListener(
new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
log.warn("分区被回收: {}", partitions);
// 保存当前处理状态，提交 offset
saveStateAndCommit(partitions);
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.info("新分配分区: {}", partitions);
// 恢复状态或 seek 到指定位置
restoreState(partitions);
}
@Override
public void onPartitionsLost(Collection<TopicPartition> partitions) {
log.warn("分区丢失（可能消费者超时）: {}", partitions);
// 清理本地状态
cleanLocalState(partitions);
}
}
);
return factory;
}
// 手动 seek 到指定 offset（实现精确重放）
@KafkaListener(topics = "rebalance-topic", groupId = "rebalance-group")
public void consume(ConsumerRecord<String, String> record,
Consumer<?, ?> consumer) {
// 处理消息
process(record);
// 假如需要跳过某些 offset
if (shouldSkip(record.offset())) {
consumer.seek(new TopicPartition(record.topic(), record.partition()),
record.offset() + 1);
}
}
}@Component
public class RebalanceAwareConsumer {
@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String>
rebalanceAwareFactory(ConsumerFactory<String, String> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.getContainerProperties().setConsumerRebalanceListener(
new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
log.warn("分区被回收: {}", partitions);
// 保存当前处理状态，提交 offset
saveStateAndCommit(partitions);
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.info("新分配分区: {}", partitions);
// 恢复状态或 seek 到指定位置
restoreState(partitions);
}
@Override
public void onPartitionsLost(Collection<TopicPartition> partitions) {
log.warn("分区丢失（可能消费者超时）: {}", partitions);
// 清理本地状态
cleanLocalState(partitions);
}
}
);
return factory;
}
// 手动 seek 到指定 offset（实现精确重放）
@KafkaListener(topics = "rebalance-topic", groupId = "rebalance-group")
public void consume(ConsumerRecord<String, String> record,
Consumer<?, ?> consumer) {
// 处理消息
process(record);
// 假如需要跳过某些 offset
if (shouldSkip(record.offset())) {
consumer.seek(new TopicPartition(record.topic(), record.partition()),
record.offset() + 1);
}
}
}3.4 Kafka 集群
集群搭建
bash
# 假设三台机器：kafka1, kafka2, kafka3
# 每台机器上的 config/server.properties 配置（以 kafka1 为例）
cat > /opt/kafka/config/server.properties << 'EOF'
broker.id=1
listeners=PLAINTEXT://kafka1:9092
log.dirs=/var/lib/kafka/data
num.partitions=3
default.replication.factor=2
min.insync.replicas=2
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=2
transaction.state.log.min.isr=2
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181
auto.create.topics.enable=true
delete.topic.enable=true
EOF
# 启动 Kafka
systemctl start kafka
systemctl enable kafka
# 验证集群
kafka-broker-api-versions.sh --bootstrap-server kafka1:9092# 假设三台机器：kafka1, kafka2, kafka3
# 每台机器上的 config/server.properties 配置（以 kafka1 为例）
cat > /opt/kafka/config/server.properties << 'EOF'
broker.id=1
listeners=PLAINTEXT://kafka1:9092
log.dirs=/var/lib/kafka/data
num.partitions=3
default.replication.factor=2
min.insync.replicas=2
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=2
transaction.state.log.min.isr=2
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181
auto.create.topics.enable=true
delete.topic.enable=true
EOF
# 启动 Kafka
systemctl start kafka
systemctl enable kafka
# 验证集群
kafka-broker-api-versions.sh --bootstrap-server kafka1:9092常用配置
properties
# 核心配置
broker.id=1
log.dirs=/var/lib/kafka/data
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181/kafka
# 主题配置
num.partitions=3
default.replication.factor=2
min.insync.replicas=2
# 日志保留策略
log.retention.hours=168 # 7 天
log.retention.bytes=1073741824 # 1 GB
log.segment.bytes=1073741824 # 单个 segment 大小
log.retention.check.interval.ms=300000
# 网络和性能
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
# 压缩
compression.type=producer# 核心配置
broker.id=1
log.dirs=/var/lib/kafka/data
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181/kafka
# 主题配置
num.partitions=3
default.replication.factor=2
min.insync.replicas=2
# 日志保留策略
log.retention.hours=168 # 7 天
log.retention.bytes=1073741824 # 1 GB
log.segment.bytes=1073741824 # 单个 segment 大小
log.retention.check.interval.ms=300000
# 网络和性能
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
# 压缩
compression.type=producer常用 CLI 命令
bash
# 创建主题
kafka-topics.sh --bootstrap-server localhost:9092 \
--create --topic my-topic \
--partitions 6 --replication-factor 2
# 查看主题列表
kafka-topics.sh --bootstrap-server localhost:9092 --list
# 查看主题详情
kafka-topics.sh --bootstrap-server localhost:9092 \
--describe --topic my-topic
# 修改主题
kafka-topics.sh --bootstrap-server localhost:9092 \
--alter --topic my-topic --partitions 12
# 删除主题
kafka-topics.sh --bootstrap-server localhost:9092 \
--delete --topic my-topic
# 控制台生产者
kafka-console-producer.sh --bootstrap-server localhost:9092 \
--topic my-topic \
--property parse.key=true \
--property key.separator=:
# 控制台消费者
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic my-topic --group my-group \
--from-beginning \
--property print.key=true \
--property print.timestamp=true
# 查看消费者组
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--list
# 查看消费者组详情（含 lag）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group my-group
# 重置消费者组 offset
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--group my-group --reset-offsets \
--to-earliest --execute --all-topics
# 查看分区副本状态
kafka-replica-verification.sh --broker-list localhost:9092 \
--topic-white-list my-topic# 创建主题
kafka-topics.sh --bootstrap-server localhost:9092 \
--create --topic my-topic \
--partitions 6 --replication-factor 2
# 查看主题列表
kafka-topics.sh --bootstrap-server localhost:9092 --list
# 查看主题详情
kafka-topics.sh --bootstrap-server localhost:9092 \
--describe --topic my-topic
# 修改主题
kafka-topics.sh --bootstrap-server localhost:9092 \
--alter --topic my-topic --partitions 12
# 删除主题
kafka-topics.sh --bootstrap-server localhost:9092 \
--delete --topic my-topic
# 控制台生产者
kafka-console-producer.sh --bootstrap-server localhost:9092 \
--topic my-topic \
--property parse.key=true \
--property key.separator=:
# 控制台消费者
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic my-topic --group my-group \
--from-beginning \
--property print.key=true \
--property print.timestamp=true
# 查看消费者组
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--list
# 查看消费者组详情（含 lag）
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group my-group
# 重置消费者组 offset
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--group my-group --reset-offsets \
--to-earliest --execute --all-topics
# 查看分区副本状态
kafka-replica-verification.sh --broker-list localhost:9092 \
--topic-white-list my-topicKafka Streams 简介
Kafka Streams 是一个轻量级的流处理库，用于构建处理 Kafka 数据的应用程序。
java
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
StreamsBuilder builder = new StreamsBuilder();
// 从输入主题读取流
KStream<String, String> source = builder.stream("input-topic");
// 流处理：过滤、转换、聚合
KTable<String, Long> counts = source
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\s+")))
.groupBy((key, word) -> word)
.count();
// 写入到输出主题
counts.toStream().to("output-topic",
Produced.with(Serdes.String(), Serdes.Long()));
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();
// 优雅关闭
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
StreamsBuilder builder = new StreamsBuilder();
// 从输入主题读取流
KStream<String, String> source = builder.stream("input-topic");
// 流处理：过滤、转换、聚合
KTable<String, Long> counts = source
.flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\s+")))
.groupBy((key, word) -> word)
.count();
// 写入到输出主题
counts.toStream().to("output-topic",
Produced.with(Serdes.String(), Serdes.Long()));
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();
// 优雅关闭
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));4. RabbitMQ vs Kafka 对比
4.1 全面对比表
| 对比维度 | RabbitMQ | Kafka |
|---|---|---|
| 定位 | 通用消息中间件，面向消息 | 分布式流处理平台，面向事件 |
| 协议 | AMQP 0-9-1 / MQTT / STOMP | 自定义 TCP 协议 |
| 消息模型 | Exchange → Queue → Consumer | Producer → Topic → Partition → Consumer Group |
| 消息顺序 | 单队列内有序 | 分区内有序，分区间无序 |
| 吞吐量 | 万级/秒（单机） | 百万级/秒（分区扩展） |
| 延迟 | 微秒级（低延迟） | 毫秒级（默认批量发送） |
| 消息存储 | 消费后默认删除 | 持久化到磁盘，按策略保留 |
| 消息路由 | 灵活（4 种 Exchange + Binding） | 仅按 Topic 名称 |
| 消费方式 | Push（Broker 推送给消费者） | Pull（消费者拉取 + 长轮询） |
| 消费确认 | ACK / NACK / Reject | 手动/自动提交 Offset |
| 重试机制 | DLX + TTL（灵活的死信重试） | @RetryableTopic 或重试主题 |
| 事务支持 | 基于 channel 的事务（性能有限） | 跨分区事务（EOS） |
| 消息回溯 | 不支持（消费后移除） | 支持（按 Offset/Timestamp 重放） |
| 消息顺序保证 | 单消费者单队列严格有序 | 单分区内严格有序 |
| 集群高可用 | 镜像队列 / Quorum Queue | Partition Replication + ISR |
| 运维复杂度 | 中等 | 较高 |
| 学习曲线 | 较低（概念直观） | 中等（概念较多） |
| Java 集成 | Spring AMQP / Spring Rabbit | Spring Kafka / Kafka Client |
| 社区活跃度 | 非常活跃 | 非常活跃 |
4.2 选型建议
选 RabbitMQ 的场景
- 需要灵活的路由策略：复杂的消息路由规则（如按内容、通配符匹配）
- 低延迟敏感应用：金融交易、实时通知，要求毫秒级甚至微秒级延迟
- 传统企业集成：大量使用 AMQP 协议的系统
- 消息数量可控：吞吐量需求在万级/秒以内
- 需要延迟消息/死信队列：RabbitMQ 原生支持 DLX + TTL，实现简单
- 需求快速迭代验证：RabbitMQ 概念简单，上手快
选 Kafka 的场景
- 超高吞吐量：需要百万级/秒的消息处理能力
- 日志收集与监控：海量日志/指标的采集和聚合
- 事件溯源 / CQRS：需要消息持久化和回溯重放
- 流处理：Kafka Streams / ksqlDB 实现复杂流计算
- 数据管道 / ETL：大数据生态集成（Flink、Spark、Hadoop）
- 削峰填谷并重放：消费能力不足时可以先积压，恢复后重放
- 多消费者独立消费：不同业务线独立消费同一份数据
混合架构
在实际的大型项目中，RabbitMQ 和 Kafka 可以并存：
| 场景 | 推荐消息队列 | 原因 |
|---|---|---|
| 用户注册 → 发送欢迎邮件 | RabbitMQ | 路由简单，延迟低 |
| 支付成功 → 更新订单状态 | RabbitMQ | 需要死信队列进行补偿 |
| 订单创建 → 触发风控分析 | Kafka | 上下游多个系统订阅 |
| 应用日志收集 | Kafka | 吞吐量大，按时间保留 |
| 实时数仓 ODS 层 | Kafka | 对接 Flink/Spark |
| 秒杀流量削峰 | Kafka | 超高瞬时流量 |
| 管理后台实时通知 | RabbitMQ | 低延迟，广播推送 |
5. 实战场景
5.1 异步解耦（订单系统）
+------------------+
| 通知服务（邮件） | <-- 30ms
+------------------+
+------------------+
下单请求 --> 订单服务 -->| 积分服务 | <-- 100ms
+------------------+
+------------------+
| 审计日志 | <-- 20ms
+------------------+ +------------------+
| 通知服务（邮件） | <-- 30ms
+------------------+
+------------------+
下单请求 --> 订单服务 -->| 积分服务 | <-- 100ms
+------------------+
+------------------+
| 审计日志 | <-- 20ms
+------------------+同步调用的问题：
- 总响应时间 = 30 + 100 + 20 = 150ms（串行）
- 任一依赖服务故障，下单失败
- 新增订阅方需修改订单服务代码
引入 MQ 后：
下单请求 --> 订单服务 --> [MQ: order.events]
|
+---------+---------+
| | |
v v v
通知服务 积分服务 审计日志下单请求 --> 订单服务 --> [MQ: order.events]
|
+---------+---------+
| | |
v v v
通知服务 积分服务 审计日志- 订单服务只负责写消息到 MQ，响应时间 < 10ms
- 下游服务各自独立消费，互不影响
- 新增服务只需订阅 Topic，不修改订单服务
- 下游宕机不影响下单核心链路
代码示例：
java
// 订单控制器
@RestController
@RequestMapping("/orders")
public class OrderController {
@Autowired
private OrderService orderService;
@PostMapping
public ResponseEntity<OrderVO> create(@RequestBody OrderCreateRequest request) {
OrderVO order = orderService.createOrder(request);
return ResponseEntity.ok(order);
}
}
// 订单服务
@Service
public class OrderService {
@Autowired
private OrderMapper orderMapper;
@Autowired
private OrderEventPublisher eventPublisher;
@Transactional
public OrderVO createOrder(OrderCreateRequest request) {
// 1. 写数据库
Order order = new Order();
order.setUserId(request.getUserId());
order.setAmount(request.getAmount());
order.setStatus(OrderStatus.CREATED);
orderMapper.insert(order);
// 2. 发布事件（此时 MQ 消息与会话绑定，事务提交后才真正发送）
eventPublisher.publishOrderCreated(order);
return OrderVO.from(order);
}
}// 订单控制器
@RestController
@RequestMapping("/orders")
public class OrderController {
@Autowired
private OrderService orderService;
@PostMapping
public ResponseEntity<OrderVO> create(@RequestBody OrderCreateRequest request) {
OrderVO order = orderService.createOrder(request);
return ResponseEntity.ok(order);
}
}
// 订单服务
@Service
public class OrderService {
@Autowired
private OrderMapper orderMapper;
@Autowired
private OrderEventPublisher eventPublisher;
@Transactional
public OrderVO createOrder(OrderCreateRequest request) {
// 1. 写数据库
Order order = new Order();
order.setUserId(request.getUserId());
order.setAmount(request.getAmount());
order.setStatus(OrderStatus.CREATED);
orderMapper.insert(order);
// 2. 发布事件（此时 MQ 消息与会话绑定，事务提交后才真正发送）
eventPublisher.publishOrderCreated(order);
return OrderVO.from(order);
}
}5.2 削峰填谷（秒杀系统）
+-------------------------------+
| 秒杀流量（10万 QPS） |
+-------------------------------+
|
v
+-------------------------------+
| Nginx / SLB |
+-------------------------------+
|
v
+-------------------------------+
| 秒杀网关（限流+前置过滤） |
+-------------------------------+
|
v
+-------------------------------+
| [Kafka: seckill.orders] |
| 削峰缓冲区（10万 TPS） |
+-------------------------------+
|
v
+-------------------------------+
| 秒杀 Worker（N 个消费者） |
| 按库存扣减，500 QPS/实例 |
+-------------------------------+
|
v
+-------------------------------+
| MySQL / Redis 库存 |
+-------------------------------+ +-------------------------------+
| 秒杀流量（10万 QPS） |
+-------------------------------+
|
v
+-------------------------------+
| Nginx / SLB |
+-------------------------------+
|
v
+-------------------------------+
| 秒杀网关（限流+前置过滤） |
+-------------------------------+
|
v
+-------------------------------+
| [Kafka: seckill.orders] |
| 削峰缓冲区（10万 TPS） |
+-------------------------------+
|
v
+-------------------------------+
| 秒杀 Worker（N 个消费者） |
| 按库存扣减，500 QPS/实例 |
+-------------------------------+
|
v
+-------------------------------+
| MySQL / Redis 库存 |
+-------------------------------+优势：
- 瞬时 10 万 QPS 先进入 Kafka，不会打垮数据库
- Worker 按自身处理能力拉取消息（500 QPS/实例）
- 可根据队列长度自动扩容 Worker
java
// 秒杀入口
@RestController
@RequestMapping("/seckill")
public class SeckillController {
@Autowired
private KafkaTemplate<String, SeckillRequest> kafkaTemplate;
@PostMapping("/order")
public ResponseEntity<?> createSeckillOrder(@Valid @RequestBody SeckillRequest request) {
// 前置校验（参数验证 + 幂等性）通过后直接发到 MQ
kafkaTemplate.send("seckill.orders", request.getUserId(), request);
return ResponseEntity.accepted().body("排队中");
}
}
// 秒杀消费者
@Component
public class SeckillConsumer {
@Autowired
private InventoryService inventoryService;
@KafkaListener(topics = "seckill.orders", groupId = "seckill-worker",
concurrency = "10")
public void processSeckill(SeckillRequest request, Acknowledgment ack) {
try {
// 尝试扣减库存
boolean success = inventoryService.deductStock(
request.getSkuId(), request.getQuantity());
if (success) {
// 创建订单
orderService.createSeckillOrder(request);
}
ack.acknowledge();
} catch (Exception e) {
// 重试或记录失败
log.error("秒杀处理失败: {}", e.getMessage());
ack.acknowledge(); // 跳过当前消息，避免阻塞
// 可发送到死信主题人工处理
}
}
}// 秒杀入口
@RestController
@RequestMapping("/seckill")
public class SeckillController {
@Autowired
private KafkaTemplate<String, SeckillRequest> kafkaTemplate;
@PostMapping("/order")
public ResponseEntity<?> createSeckillOrder(@Valid @RequestBody SeckillRequest request) {
// 前置校验（参数验证 + 幂等性）通过后直接发到 MQ
kafkaTemplate.send("seckill.orders", request.getUserId(), request);
return ResponseEntity.accepted().body("排队中");
}
}
// 秒杀消费者
@Component
public class SeckillConsumer {
@Autowired
private InventoryService inventoryService;
@KafkaListener(topics = "seckill.orders", groupId = "seckill-worker",
concurrency = "10")
public void processSeckill(SeckillRequest request, Acknowledgment ack) {
try {
// 尝试扣减库存
boolean success = inventoryService.deductStock(
request.getSkuId(), request.getQuantity());
if (success) {
// 创建订单
orderService.createSeckillOrder(request);
}
ack.acknowledge();
} catch (Exception e) {
// 重试或记录失败
log.error("秒杀处理失败: {}", e.getMessage());
ack.acknowledge(); // 跳过当前消息，避免阻塞
// 可发送到死信主题人工处理
}
}
}5.3 最终一致性（本地消息表 + MQ）
+----------+
| 业务服务 |
+----+-----+
|
1. 写业务表 + 消息表（同一本地事务）
|
2. 定时任务扫描消息表，发送到 MQ
|
+----v----+
| MQ |
+----+----+
|
3. 下游服务消费处理
|
+----v----+
| 下游服务 |
+----+-----+
|
4. 处理成功，回调删除消息表记录 +----------+
| 业务服务 |
+----+-----+
|
1. 写业务表 + 消息表（同一本地事务）
|
2. 定时任务扫描消息表，发送到 MQ
|
+----v----+
| MQ |
+----+----+
|
3. 下游服务消费处理
|
+----v----+
| 下游服务 |
+----+-----+
|
4. 处理成功，回调删除消息表记录java
// 本地消息表实体
@Data
@TableName("local_message")
public class LocalMessage {
private Long id;
private String businessKey;
private String messageBody;
private String exchange;
private String routingKey;
@TableField("`status`")
private Integer status; // 0: 待发送, 1: 已发送, 2: 已完成
private Integer retryCount;
private LocalDateTime createTime;
private LocalDateTime updateTime;
}
// 服务：写业务数据 + 写本地消息（同一事务）
@Service
public class BusinessService {
@Autowired
private BusinessMapper businessMapper;
@Autowired
private LocalMessageMapper messageMapper;
@Transactional
public void doBusiness(BusinessDTO dto) {
// 1. 写业务表
Business business = new Business();
business.setId(dto.getId());
business.setData(dto.getData());
businessMapper.insert(business);
// 2. 写本地消息表
LocalMessage message = new LocalMessage();
message.setBusinessKey(dto.getId());
message.setMessageBody(JSON.toJSONString(dto));
message.setExchange("business.exchange");
message.setRoutingKey("business.route");
message.setStatus(0);
messageMapper.insert(message);
}
}
// 定时任务：扫描并发送消息
@Component
public class MessageRelayTask {
@Autowired
private LocalMessageMapper messageMapper;
@Autowired
private RabbitTemplate rabbitTemplate;
@Scheduled(fixedDelay = 5000) // 每 5 秒扫描一次
public void relayMessages() {
List<LocalMessage> pending = messageMapper.selectList(
new LambdaQueryWrapper<LocalMessage>()
.eq(LocalMessage::getStatus, 0)
.lt(LocalMessage::getRetryCount, 5)
.last("limit 100"));
for (LocalMessage msg : pending) {
try {
rabbitTemplate.convertAndSend(
msg.getExchange(),
msg.getRoutingKey(),
msg.getMessageBody(),
new CorrelationData(msg.getBusinessKey()));
msg.setStatus(1);
messageMapper.updateById(msg);
} catch (Exception e) {
msg.setRetryCount(msg.getRetryCount() + 1);
msg.setUpdateTime(LocalDateTime.now());
messageMapper.updateById(msg);
}
}
}
}// 本地消息表实体
@Data
@TableName("local_message")
public class LocalMessage {
private Long id;
private String businessKey;
private String messageBody;
private String exchange;
private String routingKey;
@TableField("`status`")
private Integer status; // 0: 待发送, 1: 已发送, 2: 已完成
private Integer retryCount;
private LocalDateTime createTime;
private LocalDateTime updateTime;
}
// 服务：写业务数据 + 写本地消息（同一事务）
@Service
public class BusinessService {
@Autowired
private BusinessMapper businessMapper;
@Autowired
private LocalMessageMapper messageMapper;
@Transactional
public void doBusiness(BusinessDTO dto) {
// 1. 写业务表
Business business = new Business();
business.setId(dto.getId());
business.setData(dto.getData());
businessMapper.insert(business);
// 2. 写本地消息表
LocalMessage message = new LocalMessage();
message.setBusinessKey(dto.getId());
message.setMessageBody(JSON.toJSONString(dto));
message.setExchange("business.exchange");
message.setRoutingKey("business.route");
message.setStatus(0);
messageMapper.insert(message);
}
}
// 定时任务：扫描并发送消息
@Component
public class MessageRelayTask {
@Autowired
private LocalMessageMapper messageMapper;
@Autowired
private RabbitTemplate rabbitTemplate;
@Scheduled(fixedDelay = 5000) // 每 5 秒扫描一次
public void relayMessages() {
List<LocalMessage> pending = messageMapper.selectList(
new LambdaQueryWrapper<LocalMessage>()
.eq(LocalMessage::getStatus, 0)
.lt(LocalMessage::getRetryCount, 5)
.last("limit 100"));
for (LocalMessage msg : pending) {
try {
rabbitTemplate.convertAndSend(
msg.getExchange(),
msg.getRoutingKey(),
msg.getMessageBody(),
new CorrelationData(msg.getBusinessKey()));
msg.setStatus(1);
messageMapper.updateById(msg);
} catch (Exception e) {
msg.setRetryCount(msg.getRetryCount() + 1);
msg.setUpdateTime(LocalDateTime.now());
messageMapper.updateById(msg);
}
}
}
}5.4 事件驱动架构（领域事件）
+------------------------+
| Command Side |
| （Order Service） |
| order.created |
+----------+-------------+
|
v
+------------------------+
| [Kafka: order.events] |
+---+---------+---------+
| | |
v v v
+---------+ +---------+ +----------+
|Payment | |Inventory| |Analytics |
|Service | |Service | |Service |
+---------+ +---------+ +----------+
+------------------------+
| Query Side |
| （Order Read Model） |
| 订单详情查询（最终一致） |
+------------------------+ +------------------------+
| Command Side |
| （Order Service） |
| order.created |
+----------+-------------+
|
v
+------------------------+
| [Kafka: order.events] |
+---+---------+---------+
| | |
v v v
+---------+ +---------+ +----------+
|Payment | |Inventory| |Analytics |
|Service | |Service | |Service |
+---------+ +---------+ +----------+
+------------------------+
| Query Side |
| （Order Read Model） |
| 订单详情查询（最终一致） |
+------------------------+事件模式实现：
java
// 领域事件基类
public abstract class DomainEvent {
private final String eventId = UUID.randomUUID().toString();
private final LocalDateTime occurredOn = LocalDateTime.now();
public String getEventId() { return eventId; }
public LocalDateTime getOccurredOn() { return occurredOn; }
}
// 具体事件
public class OrderCreatedEvent extends DomainEvent {
private final String orderId;
private final String userId;
private final BigDecimal amount;
public OrderCreatedEvent(String orderId, String userId, BigDecimal amount) {
this.orderId = orderId;
this.userId = userId;
this.amount = amount;
}
// getters...
}
public class OrderPaidEvent extends DomainEvent {
private final String orderId;
private final String paymentId;
// ...
}
// 事件发布器
@Component
public class DomainEventPublisher {
@Autowired
private KafkaTemplate<String, DomainEvent> kafkaTemplate;
public void publish(String topic, DomainEvent event, String key) {
kafkaTemplate.send(topic, key, event);
}
public void publishOrderCreated(OrderCreatedEvent event) {
publish("order.events", event, event.getOrderId());
}
public void publishOrderPaid(OrderPaidEvent event) {
publish("order.events", event, event.getOrderId());
}
}
// 订单聚合根
@Service
public class OrderAggregate {
@Autowired
private DomainEventPublisher eventPublisher;
@Transactional
public Order createOrder(CreateOrderCommand cmd) {
// 1. 验证
validate(cmd);
// 2. 创建订单实体
Order order = new Order(cmd.getUserId(), cmd.getItems());
save(order);
// 3. 发布领域事件
eventPublisher.publishOrderCreated(new OrderCreatedEvent(
order.getId(), order.getUserId(), order.getTotalAmount()));
return order;
}
public void payOrder(PayOrderCommand cmd) {
Order order = findOrder(cmd.getOrderId());
order.pay(cmd.getPaymentId());
save(order);
eventPublisher.publishOrderPaid(new OrderPaidEvent(
order.getId(), cmd.getPaymentId()));
}
}
// 事件订阅者：积分服务
@Component
public class PointsEventSubscriber {
@Autowired
private PointsService pointsService;
@KafkaListener(topics = "order.events", groupId = "points-service")
public void handleOrderPaid(OrderPaidEvent event) {
// 订单已支付，给用户加积分
pointsService.addPoints(event.getUserId(), 100);
}
}
// 事件订阅者：通知服务
@Component
public class NotificationEventSubscriber {
@Autowired
private NotificationService notificationService;
@KafkaListener(topics = "order.events", groupId = "notification-service")
public void handleOrderCreated(OrderCreatedEvent event) {
notificationService.sendOrderConfirmation(event.getUserId(), event.getOrderId());
}
}// 领域事件基类
public abstract class DomainEvent {
private final String eventId = UUID.randomUUID().toString();
private final LocalDateTime occurredOn = LocalDateTime.now();
public String getEventId() { return eventId; }
public LocalDateTime getOccurredOn() { return occurredOn; }
}
// 具体事件
public class OrderCreatedEvent extends DomainEvent {
private final String orderId;
private final String userId;
private final BigDecimal amount;
public OrderCreatedEvent(String orderId, String userId, BigDecimal amount) {
this.orderId = orderId;
this.userId = userId;
this.amount = amount;
}
// getters...
}
public class OrderPaidEvent extends DomainEvent {
private final String orderId;
private final String paymentId;
// ...
}
// 事件发布器
@Component
public class DomainEventPublisher {
@Autowired
private KafkaTemplate<String, DomainEvent> kafkaTemplate;
public void publish(String topic, DomainEvent event, String key) {
kafkaTemplate.send(topic, key, event);
}
public void publishOrderCreated(OrderCreatedEvent event) {
publish("order.events", event, event.getOrderId());
}
public void publishOrderPaid(OrderPaidEvent event) {
publish("order.events", event, event.getOrderId());
}
}
// 订单聚合根
@Service
public class OrderAggregate {
@Autowired
private DomainEventPublisher eventPublisher;
@Transactional
public Order createOrder(CreateOrderCommand cmd) {
// 1. 验证
validate(cmd);
// 2. 创建订单实体
Order order = new Order(cmd.getUserId(), cmd.getItems());
save(order);
// 3. 发布领域事件
eventPublisher.publishOrderCreated(new OrderCreatedEvent(
order.getId(), order.getUserId(), order.getTotalAmount()));
return order;
}
public void payOrder(PayOrderCommand cmd) {
Order order = findOrder(cmd.getOrderId());
order.pay(cmd.getPaymentId());
save(order);
eventPublisher.publishOrderPaid(new OrderPaidEvent(
order.getId(), cmd.getPaymentId()));
}
}
// 事件订阅者：积分服务
@Component
public class PointsEventSubscriber {
@Autowired
private PointsService pointsService;
@KafkaListener(topics = "order.events", groupId = "points-service")
public void handleOrderPaid(OrderPaidEvent event) {
// 订单已支付，给用户加积分
pointsService.addPoints(event.getUserId(), 100);
}
}
// 事件订阅者：通知服务
@Component
public class NotificationEventSubscriber {
@Autowired
private NotificationService notificationService;
@KafkaListener(topics = "order.events", groupId = "notification-service")
public void handleOrderCreated(OrderCreatedEvent event) {
notificationService.sendOrderConfirmation(event.getUserId(), event.getOrderId());
}
}5.5 消息积压处理
问题：消费者处理速度跟不上生产者发送速度，导致队列/主题中的消息大量积压。
排查步骤：
bash
# RabbitMQ 查看队列积压
rabbitmqctl list_queues name messages messages_ready messages_unacknowledged
# Kafka 查看消费者组 lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group my-group
# 关注 LAG 列：当前 offset 与最新 offset 的差值# RabbitMQ 查看队列积压
rabbitmqctl list_queues name messages messages_ready messages_unacknowledged
# Kafka 查看消费者组 lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group my-group
# 关注 LAG 列：当前 offset 与最新 offset 的差值常见原因和解决方案：
| 原因 | 解决方案 |
|---|---|
| 消费者处理慢 | 增加分区 + 消费者实例数（并发消费） |
| 消费者异常/卡住 | 检查死锁、数据库连接池、外部依赖 |
| 生产者突增流量 | 临时增加消费者实例（Kafka 同组增减） |
| 消息体过大 | 评估是否可压缩或引用存储 |
| 数据库瓶颈 | 批量写入、异步刷盘 |
Kafka 快速扩容消费能力：
yaml
# 方法一：增加单实例消费者线程数
spring:
kafka:
listener:
concurrency: 10 # 原来 3，现在 10
# 方法二：增加消费者组实例
# 启动多个实例，只要 groupId 相同即可
# 注意：消费者数不能超过分区总数# 方法一：增加单实例消费者线程数
spring:
kafka:
listener:
concurrency: 10 # 原来 3，现在 10
# 方法二：增加消费者组实例
# 启动多个实例，只要 groupId 相同即可
# 注意：消费者数不能超过分区总数RabbitMQ 快速扩容：
yaml
# 增加并发消费者数
spring:
rabbitmq:
listener:
simple:
concurrency: 5 # 初始
max-concurrency: 20 # 最大# 增加并发消费者数
spring:
rabbitmq:
listener:
simple:
concurrency: 5 # 初始
max-concurrency: 20 # 最大消息积压补偿方案：
java
@Component
public class LagMonitor {
private static final Logger log = LoggerFactory.getLogger(LagMonitor.class);
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 监控队列积压，触发动态扩缩容或告警
*/
@Scheduled(fixedDelay = 30000)
public void monitorQueueDepth() {
// RabbitMQ 管理 API
RestTemplate rest = new RestTemplate();
String url = "http://localhost:15672/api/queues/%2f/order.queue";
// Basic Auth
String auth = "Basic " + Base64.getEncoder().encodeToString("guest:guest".getBytes());
HttpHeaders headers = new HttpHeaders();
headers.add("Authorization", auth);
ResponseEntity<Map> response = rest.exchange(
url, HttpMethod.GET, new HttpEntity<>(headers), Map.class);
Map<String, Object> body = response.getBody();
int messages = (int) body.getOrDefault("messages", 0);
if (messages > 10000) {
log.warn("队列积压告警: {} 条消息", messages);
// 触发告警：钉钉/邮件
alertService.sendAlert("订单队列积压", messages + " 条消息待处理");
// 动态扩容（通过 K8s API 或手动）
scaleUpConsumers();
}
}
}@Component
public class LagMonitor {
private static final Logger log = LoggerFactory.getLogger(LagMonitor.class);
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 监控队列积压，触发动态扩缩容或告警
*/
@Scheduled(fixedDelay = 30000)
public void monitorQueueDepth() {
// RabbitMQ 管理 API
RestTemplate rest = new RestTemplate();
String url = "http://localhost:15672/api/queues/%2f/order.queue";
// Basic Auth
String auth = "Basic " + Base64.getEncoder().encodeToString("guest:guest".getBytes());
HttpHeaders headers = new HttpHeaders();
headers.add("Authorization", auth);
ResponseEntity<Map> response = rest.exchange(
url, HttpMethod.GET, new HttpEntity<>(headers), Map.class);
Map<String, Object> body = response.getBody();
int messages = (int) body.getOrDefault("messages", 0);
if (messages > 10000) {
log.warn("队列积压告警: {} 条消息", messages);
// 触发告警：钉钉/邮件
alertService.sendAlert("订单队列积压", messages + " 条消息待处理");
// 动态扩容（通过 K8s API 或手动）
scaleUpConsumers();
}
}
}5.6 消息幂等性
问题：MQ 的"至少一次"投递语义可能导致消费者重复收到同一条消息。
解决方案：
java
// 方案一：数据库唯一键约束
@Service
public class IdempotentConsumer {
@Autowired
private JdbcTemplate jdbcTemplate;
@KafkaListener(topics = "order-events", groupId = "idempotent-group")
public void consume(ConsumerRecord<String, OrderEvent> record) {
String messageId = record.key(); // 业务幂等键
try {
// 使用 INSERT IGNORE（或 ON DUPLICATE KEY）
jdbcTemplate.update(
"INSERT INTO idempotent_record (message_id, status) VALUES (?, 'processing')",
messageId);
} catch (DuplicateKeyException e) {
log.info("重复消息，跳过: {}", messageId);
return;
}
// 业务处理
processEvent(record.value());
// 更新状态
jdbcTemplate.update(
"UPDATE idempotent_record SET status = 'done' WHERE message_id = ?",
messageId);
}
}
// 方案二：Redis 分布式锁（轻量级）
@Component
public class RedisIdempotentConsumer {
@Autowired
private StringRedisTemplate redisTemplate;
@RabbitListener(queues = "order.queue")
public void consume(OrderEvent event) {
String dedupKey = "order:dedup:" + event.getEventId();
// SET NX EX 300 保证原子性
Boolean absent = redisTemplate.opsForValue()
.setIfAbsent(dedupKey, "1", Duration.ofMinutes(5));
if (Boolean.FALSE.equals(absent)) {
log.info("重复消息（Redis 去重）: {}", event.getEventId());
return;
}
// 业务处理
process(event);
}
}
// 方案三：业务键去重
@Table(name = "`order`", uniqueConstraints =
@UniqueConstraint(columnNames = "business_id"))
public class OrderEntity {
// business_id 唯一约束 (如支付回调 ID、第三方订单号)
private String businessId;
}// 方案一：数据库唯一键约束
@Service
public class IdempotentConsumer {
@Autowired
private JdbcTemplate jdbcTemplate;
@KafkaListener(topics = "order-events", groupId = "idempotent-group")
public void consume(ConsumerRecord<String, OrderEvent> record) {
String messageId = record.key(); // 业务幂等键
try {
// 使用 INSERT IGNORE（或 ON DUPLICATE KEY）
jdbcTemplate.update(
"INSERT INTO idempotent_record (message_id, status) VALUES (?, 'processing')",
messageId);
} catch (DuplicateKeyException e) {
log.info("重复消息，跳过: {}", messageId);
return;
}
// 业务处理
processEvent(record.value());
// 更新状态
jdbcTemplate.update(
"UPDATE idempotent_record SET status = 'done' WHERE message_id = ?",
messageId);
}
}
// 方案二：Redis 分布式锁（轻量级）
@Component
public class RedisIdempotentConsumer {
@Autowired
private StringRedisTemplate redisTemplate;
@RabbitListener(queues = "order.queue")
public void consume(OrderEvent event) {
String dedupKey = "order:dedup:" + event.getEventId();
// SET NX EX 300 保证原子性
Boolean absent = redisTemplate.opsForValue()
.setIfAbsent(dedupKey, "1", Duration.ofMinutes(5));
if (Boolean.FALSE.equals(absent)) {
log.info("重复消息（Redis 去重）: {}", event.getEventId());
return;
}
// 业务处理
process(event);
}
}
// 方案三：业务键去重
@Table(name = "`order`", uniqueConstraints =
@UniqueConstraint(columnNames = "business_id"))
public class OrderEntity {
// business_id 唯一约束 (如支付回调 ID、第三方订单号)
private String businessId;
}5.7 消息轨迹与全链路追踪
集成 MDC（Mapped Diagnostic Context）实现分布式链路追踪：
java
@Component
public class TracingConsumer {
@KafkaListener(topics = "order-events", groupId = "tracing-group")
public void consume(ConsumerRecord<String, String> record) {
// 获取上游传入的 traceId
String traceId = record.headers()
.lastHeader("trace-id") != null
? new String(record.headers().lastHeader("trace-id").value())
: UUID.randomUUID().toString();
// 设置到日志 MDC
MDC.put("traceId", traceId);
MDC.put("spanId", UUID.randomUUID().toString().substring(0, 8));
try {
log.info("开始处理消息: key={}, offset={}", record.key(), record.offset());
// 业务处理
process(record.value());
log.info("处理完成: key={}", record.key());
} finally {
MDC.clear();
}
}
}@Component
public class TracingConsumer {
@KafkaListener(topics = "order-events", groupId = "tracing-group")
public void consume(ConsumerRecord<String, String> record) {
// 获取上游传入的 traceId
String traceId = record.headers()
.lastHeader("trace-id") != null
? new String(record.headers().lastHeader("trace-id").value())
: UUID.randomUUID().toString();
// 设置到日志 MDC
MDC.put("traceId", traceId);
MDC.put("spanId", UUID.randomUUID().toString().substring(0, 8));
try {
log.info("开始处理消息: key={}, offset={}", record.key(), record.offset());
// 业务处理
process(record.value());
log.info("处理完成: key={}", record.key());
} finally {
MDC.clear();
}
}
}总结
| 维度 | RabbitMQ | Kafka |
|---|---|---|
| 核心优势 | 灵活路由、低延迟、易用性 | 高吞吐、持久化、流处理 |
| 典型场景 | 异步任务、业务解耦、通知推送 | 日志聚合、事件溯源、大数据管道 |
| Java 集成 | Spring AMQP（强类型、注解驱动） | Spring Kafka（Consumer/Producer API） |
| 生产建议 | 新项目用 Quorum Queue 替代镜像队列 | 用幂等生产者 + 事务 + CooperativeSticky |
| 运维要点 | 关注队列积压、连接数、内存 | 关注 ISR、Lag、磁盘使用率 |
| 学习资源 | RabbitMQ 官方文档 | Kafka 官方文档 |
选型并不互斥：在企业架构中，RabbitMQ 和 Kafka 可以共生——RabbitMQ 做业务消息枢纽，Kafka 做数据管道与流处理。