Spring Cloud Stream 消息驱动编程完全指南
核心概念
1 什么是 Spring Cloud Stream
Spring Cloud Stream 是一个用于构建消息驱动微服务的框架，它基于 Spring Boot 和 Spring Integration，提供了统一的编程模型,屏蔽了不同消息中间件的差异。
2 核心组件
┌──────────────────────────────────────┐
│ Application Core Logic │
├──────────────────────────────────────┤
│ @EnableBinding / @EnableBinding │
│ @StreamListener │
├──────────────────────────────────────┤
│ Source / Sink / Processor │
├──────────────────────────────────────┤
│ Binder 抽象层 │
├──────────────────────────────────────┤
│ RabbitMQ │ Kafka │ 其他 MQ │
└──────────────────────────────────────┘基础配置
1 Maven 依赖
<!-- Spring Cloud Stream 基础依赖 -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!-- 或使用 Kafka -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
</dependency>
2 核心注解配置
@SpringBootApplication
@EnableBinding({Source.class, Sink.class, Processor.class})
public class StreamApplication {
public static void main(String[] args) {
SpringApplication.run(StreamApplication.class, args);
}
}
编程模型详解
1 消息生产者 (Source)
基础生产者
@Component
public class MessageProducer {
@Autowired
private Source source;
public void sendMessage(String message) {
source.output().send(MessageBuilder
.withPayload(message)
.setHeader("content-type", "text/plain")
.build());
}
// 定时发送
@Scheduled(fixedDelay = 5000)
public void sendPeriodicMessage() {
String msg = "Time: " + new Date();
source.output().send(MessageBuilder
.withPayload(msg)
.build());
}
}
使用 StreamBridge (Spring Cloud Stream 3.x+)
@Component
public class StreamBridgeProducer {
@Autowired
private StreamBridge streamBridge;
public void sendMessage(String bindingName, Object payload) {
streamBridge.send(bindingName, MessageBuilder
.withPayload(payload)
.setHeader("messageId", UUID.randomUUID().toString())
.build());
}
// 支持异步发送
@Async
public CompletableFuture<Boolean> sendAsync(String bindingName, Object payload) {
return streamBridge.sendAsync(bindingName, payload);
}
}
2 消息消费者 (Sink)
基础消费者
@Component
@Slf4j
public class MessageConsumer {
@StreamListener(Sink.INPUT)
public void handleMessage(String message) {
log.info("Received message: {}", message);
// 业务处理逻辑
}
// 使用 Function 编程模型 (Spring Cloud Stream 3.x+)
@Bean
public Function<String, String> processMessage() {
return message -> {
log.info("Processing: {}", message);
return message.toUpperCase();
};
}
}
自定义 Sink 接口
public interface CustomSink {
String INPUT = "custom-input";
@Input(CustomSink.INPUT)
SubscribableChannel input();
}
// 使用自定义 Sink
@EnableBinding(CustomSink.class)
@Component
@Slf4j
public class CustomConsumer {
@StreamListener(CustomSink.INPUT)
public void handleCustomMessage(Message<String> message) {
log.info("Headers: {}", message.getHeaders());
log.info("Payload: {}", message.getPayload());
}
}
3 消息处理器 (Processor)
处理器接口
@Component
@EnableBinding(Processor.class)
@Slf4j
public class MessageProcessor {
// 接收 -> 处理 -> 发送
@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public String process(String message) {
log.info("Processing message: {}", message);
// 转换处理
String processed = "Processed: " + message;
return processed;
}
}
多处理器链
@Component
@Slf4j
public class ChainProcessors {
@Bean
public Function<String, String> firstProcessor() {
return message -> {
log.info("First processor: {}", message);
return message + " -> processed by first";
};
}
@Bean
public Function<String, String> secondProcessor() {
return message -> {
log.info("Second processor: {}", message);
return message + " -> processed by second";
};
}
}
高级特性
1 消息分区
spring:
cloud:
stream:
bindings:
output:
destination: partitioned-topic
producer:
partition-key-expression: headers['partitionKey']
partition-count: 3
# 消费者配置
input:
destination: partitioned-topic
consumer:
partitioned: true
instance-count: 3
instance-index: 0 # 0,1,2 分别对应不同实例
2 错误处理机制
重试机制
@Component
@Slf4j
public class RetryConsumer {
@StreamListener(Sink.INPUT)
public void handleWithRetry(String message) {
try {
// 业务处理
processBusinessLogic(message);
} catch (Exception e) {
log.error("Processing failed", e);
// 抛出异常会自动触发重试
throw new RuntimeException(e);
}
}
}
spring:
cloud:
stream:
bindings:
input:
consumer:
max-attempts: 3
back-off-initial-interval: 1000
back-off-multiplier: 2.0
default-retryable: true
死信队列 (DLQ)
spring:
cloud:
stream:
rabbit:
bindings:
input:
consumer:
auto-bind-dlq: true
republish-to-dlq: true
dlq-dead-letter-exchange: dlx
dlq-dead-letter-routing-key: dlq-key
3 消息转换
自定义消息转换
@Bean
public MessageConverter customMessageConverter() {
return new MappingJackson2MessageConverter() {
@Override
protected Object convertFromInternal(Message<?> message,
Class<?> targetClass,
Object conversionHint) {
// 自定义转换逻辑
return super.convertFromInternal(message, targetClass, conversionHint);
}
};
}
JSON 消息处理
@Data
public class OrderEvent {
private String orderId;
private String customerName;
private BigDecimal amount;
private Date createTime;
}
@Component
@Slf4j
public class OrderConsumer {
@StreamListener(Sink.INPUT)
public void handleOrder(OrderEvent orderEvent) {
log.info("Received order: {}", orderEvent);
// 处理订单
}
}
完整实战示例
1 项目结构
src/main/java/com/example/stream/
├── StreamApplication.java
├── channel/
│ └── CustomChannel.java
├── producer/
│ ├── MessageProducer.java
│ └── OrderProducer.java
├── consumer/
│ ├── MessageConsumer.java
│ └── OrderConsumer.java
├── model/
│ └── OrderEvent.java
└── config/
└── StreamConfig.java2 配置文件
# application.yml
spring:
cloud:
stream:
default-binder: rabbit
bindings:
order-output:
destination: order-topic
content-type: application/json
producer:
required-groups: order-group
order-input:
destination: order-topic
group: order-group
consumer:
max-attempts: 3
back-off-initial-interval: 1000
rabbit:
bindings:
order-input:
consumer:
auto-bind-dlq: true
republish-to-dlq: true
dlq-dead-letter-exchange: order-dlx
dlq-dead-letter-routing-key: order-dlq
3 REST API 接口
@RestController
@RequestMapping("/api/messages")
@Slf4j
public class MessageController {
@Autowired
private StreamBridge streamBridge;
@Autowired
private MessageProducer messageProducer;
@PostMapping("/send")
public ResponseEntity<String> sendMessage(@RequestBody String message) {
messageProducer.sendMessage(message);
return ResponseEntity.ok("Message sent: " + message);
}
@PostMapping("/order")
public ResponseEntity<String> sendOrder(@RequestBody OrderEvent order) {
streamBridge.send("order-output", order);
return ResponseEntity.ok("Order sent: " + order.getOrderId());
}
@GetMapping("/test")
public ResponseEntity<String> testStream() {
// 测试流处理
streamBridge.send("test-input", "test message");
return ResponseEntity.ok("Test sent");
}
}
常用绑定器配置
1 RabbitMQ Binder
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
template:
retry:
enabled: true
initial-interval: 2s
max-attempts: 3
2 Kafka Binder
spring:
kafka:
bootstrap-servers: localhost:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: stream-group
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
最佳实践
1 幂等性处理
@Component
@Slf4j
public class IdempotentConsumer {
private final Set<String> processedMessages = ConcurrentHashMap.newKeySet();
@StreamListener(Sink.INPUT)
public void handleMessage(Message<String> message) {
String messageId = message.getHeaders().getId().toString();
// 幂等性检查
if (!processedMessages.add(messageId)) {
log.warn("Duplicate message: {}", messageId);
return;
}
try {
// 业务处理
process(message.getPayload());
} catch (Exception e) {
processedMessages.remove(messageId);
throw e;
}
}
}
2 消息追踪
@Component
@Slf4j
public class MessageTracingInterceptor implements ChannelInterceptor {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
// 添加追踪信息
MessageHeaders headers = message.getHeaders();
String traceId = headers.containsKey("traceId")
? headers.get("traceId").toString()
: UUID.randomUUID().toString();
log.info("Sending message with traceId: {}", traceId);
return MessageBuilder.fromMessage(message)
.setHeader("traceId", traceId)
.build();
}
@Override
public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
log.info("Message sent: {}, status: {}",
message.getHeaders().getId(), sent ? "success" : "failed");
}
}
3 监控指标
management:
endpoints:
web:
exposure:
include: bindings,stream
metrics:
export:
prometheus:
enabled: true
Spring Cloud Stream 提供了一套完整的消息驱动解决方案，通过统一的编程模型，让开发者可以专注于业务逻辑，而无需关心底层消息中间件的具体实现，掌握了这些核心概念和最佳实践，你就能构建出健壮、可扩展的消息驱动微服务系统。
