如何在Spring Cloud Stream中结合Java函数式API使用轮询消费者

明确结论 #

完全可以在Spring Cloud Stream的轮询消费者中使用函数式开发风格，这也是当前官方主推的开发模式。旧版的注解式API（如org.springframework.cloud.stream.annotation.Input）从3.x版本开始就已标记为弃用，后续版本会逐步移除，完全不建议继续使用。

具体实现方式 #

有两种常用实现方案，可根据你的业务需求选择：

方案1：固定间隔自动轮询 #

直接声明函数式Bean即可，框架会按照配置的规则自动执行轮询拉取消息：

1. 编写业务逻辑Bean

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import java.util.function.Consumer;

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

    // Consumer类型函数，用于消费拉取到的消息
    @Bean
    public Consumer<String> polledMessageConsumer() {
        return payload -> {
            // 此处写你的业务处理逻辑
            System.out.println("接收到消息：" + payload);
        };
    }
}

2. 添加配置
   在application.yml中添加轮询规则和绑定配置：

spring:
  cloud:
    stream:
      bindings:
        polledMessageConsumer-in-0:
          destination: 你要消费的topic/队列名称
          group: 消费者组名称
      # 全局轮询配置，也可针对单个绑定单独配置
      poller:
        fixed-delay: 2000 # 轮询间隔，单位为毫秒，可按需调整
        max-messages-per-poll: 5 # 每次轮询最多拉取的消息数

方案2：手动控制拉取时机 #

如果你的业务需要在特定时机触发拉取，而非固定间隔自动轮询，可以直接注入PollableMessageSource实例手动调用拉取方法：

import org.springframework.cloud.stream.binder.PollableMessageSource;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

@Service
public class CustomPollService {
    private final PollableMessageSource pollableMessageSource;

    // 框架会自动注入对应绑定的PollableMessageSource实例
    public CustomPollService(PollableMessageSource pollableMessageSource) {
        this.pollableMessageSource = pollableMessageSource;
    }

    // 你可以在任意业务场景下调用该方法触发消息拉取
    public void pullMessage() {
        Message<?> receivedMessage = pollableMessageSource.poll();
        if (receivedMessage != null) {
            String payload = (String) receivedMessage.getPayload();
            // 此处写你的业务处理逻辑
        }
    }
}

注意事项 #

• 函数式模型下的绑定名称规则为函数名 + -in/out + 序号，序号从0开始，和函数的输入输出参数数量对应，配置时不要写错
• 如果同时存在多个函数式Bean，需要在配置中指定spring.cloud.function.definition参数，声明要生效的函数名，多个用分号分隔

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