Spring Cloud Stream v2至v4中Sink接口迁移方案咨询

问题根源 #

v4 版本彻底移除了 Sink、Source 这类预定义通道接口，DirectWithAttributesChannel 是底层通道实现类，并非可直接注入的容器 Bean，所以直接替换会导致找不到 Bean 的错误。以下是两种可行的迁移方案：

方案一：兼容原有通道绑定模型 #

如果想保留原有基于通道的编程风格，可通过自定义输入通道接口替代 Sink：

1. 定义自定义输入通道接口 #

创建自己的输入通道接口，用 @Input 标记通道：

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.MessageChannel;

public interface CustomSink {
    String INPUT = "customSinkInput";

    @Input(INPUT)
    MessageChannel input();
}

2. 启用通道绑定 #

在启动类或配置类上添加 @EnableBinding 注解，绑定自定义接口：

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableBinding(CustomSink.class)
public class ProductionApplication {
    public static void main(String[] args) {
        SpringApplication.run(ProductionApplication.class, args);
    }
}

3. 修改依赖注入逻辑 #

将原来注入 Sink 的代码替换为注入自定义的 CustomSink：

import org.springframework.stereotype.Component;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;

@Component
public class StreamUtils {
    private final CustomSink customSink;

    // 推荐使用构造器注入
    @Autowired
    public StreamUtils(CustomSink customSink) {
        this.customSink = customSink;
    }

    // 示例：发送消息到输入通道
    public void sendMessage(Object payload) {
        customSink.input().send(MessageBuilder.withPayload(payload).build());
    }
}

4. 配置通道绑定属性 #

在 application.yml 中配置通道的消息目的地和消费组：

spring:
  cloud:
    stream:
      bindings:
        customSinkInput:
          destination: your-production-topic # 替换为实际消息主题/队列名
          group: production-consumer-group # 必填，避免重复消费

5. 单元测试调整 #

替换测试类中的 Sink 为 CustomSink，配合 MessageCollector 做测试：

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.messaging.Message;

import static org.junit.jupiter.api.Assertions.assertEquals;

@SpringBootTest
public class StreamUtilsTest {
    @Autowired
    private CustomSink customSink;

    @Autowired
    private MessageCollector messageCollector;

    @Autowired
    private StreamUtils streamUtils;

    @Test
    public void testSendMessage() {
        String testPayload = "test-production-event";
        streamUtils.sendMessage(testPayload);

        Message<?> receivedMsg = messageCollector.forChannel(customSink.input()).poll();
        assertEquals(testPayload, receivedMsg.getPayload());
    }
}

方案二：迁移到函数式编程模型（v4 推荐） #

v4 官方推荐使用函数式编程模型替代传统通道绑定，更简洁高效：

1. 编写消费函数 #

创建配置类，定义消息消费的函数：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.function.Consumer;

@Configuration
public class StreamFunctionConfig {
    @Bean
    public Consumer<String> productionEventConsumer() {
        return payload -> {
            // 这里编写原有的消息处理逻辑
            System.out.println("Received production event: " + payload);
        };
    }
}

2. 配置函数绑定 #

在 application.yml 中配置函数对应的通道属性：

spring:
  cloud:
    stream:
      bindings:
        productionEventConsumer-in-0:
          destination: your-production-topic
          group: production-consumer-group

3. 单元测试调整 #

针对函数式模型编写测试：

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.messaging.support.MessageBuilder;

@SpringBootTest(classes = {ProductionApplication.class, TestChannelBinderConfiguration.class})
public class ProductionConsumerTest {
    @Autowired
    private InputDestination inputDestination;

    @Test
    public void testConsumer() {
        String testPayload = "test-production-event";
        inputDestination.send(MessageBuilder.withPayload(testPayload).build(), "productionEventConsumer-in-0");
        
        // 这里添加断言，验证消息处理逻辑是否执行
    }
}

内容的提问来源于stack exchange，提问作者ThrowsError