SpringCloudStream消息驱动编程

wen java案例 2026-07-28 241

Spring Cloud Stream 消息驱动编程完全指南

核心概念

1 什么是 Spring Cloud Stream

Spring Cloud Stream 是一个用于构建消息驱动微服务的框架，它基于 Spring Boot 和 Spring Integration，提供了统一的编程模型,屏蔽了不同消息中间件的差异。

SpringCloudStream消息驱动编程

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.java

2 配置文件

# 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 提供了一套完整的消息驱动解决方案，通过统一的编程模型，让开发者可以专注于业务逻辑，而无需关心底层消息中间件的具体实现，掌握了这些核心概念和最佳实践，你就能构建出健壮、可扩展的消息驱动微服务系统。

本文地址： https://www.bmjup.com/post/32201.html