这是为 Spring Cloud Stream 2.0.0.RELEASE 发布准备的一系列预发布博客中的第二篇。

前言

Spring Cloud Stream 2.0 引入了轮询消费者，应用程序可以控制消息处理速率。

简介

Spring Cloud Stream 包含生产者和消费者的概念；在使用消息传递范例时，MessageChannel 会绑定到目标（例如 Kafka 主题、Rabbit 交换机/队列）。到目前为止，在消费者端，只要有空闲的消费者可用，就会传递消息。实际上，代理控制着传递速率；通常，在当前消息处理完成后，立即传递下一条消息。

2.0 引入了轮询消费者，应用程序可以控制消息消费的速率。Kafka 和 RabbitMQ 绑定器支持轮询消费者。

详细信息

使用轮询消费者，我们不将 MessageChannel 绑定到目标，而是绑定 PollableMessageSource；例如，PolledProcessor 绑定可能配置如下

public interface PolledProcessor {

    @Input
    PollableMessageSource destIn();

    @Output
    MessageChannel destOut();

}

消息源有一个方法

boolean poll(MessageHandler handler);

在处理程序的 handleRequest 方法退出之前，不会确认消息。

MessageHandler 是来自 spring-messaging 的接口；您可以提供一个标准的 Spring Integration 消息处理程序，或您自己的实现（通常是 lambda）。由于 handleMessage 方法采用 Message<?> 参数，因此没有类型信息，并且不会转换有效负载。

但是，消息转换如本系列中的第一篇博客中所述也可以应用于轮询消费者。为了将类型信息传达给转换服务，我们在重载的 poll() 方法中提供了一个参数化类型引用

boolean poll(MessageHandler handler, ParameterizedTypeReference<?> type)

并且消息有效负载将转换为该类型，该类型可以很简单，例如内容类型为 text/plain

• new ParameterizedTypeReference<String>() {}

或者更复杂，例如 JSON 内容类型

• new ParameterizedTypeReference<Map<String, Foo>>() {}

整合在一起

以下简单的 Spring Boot 应用程序提供了一个完整的示例；它接收字符串有效负载，将其转换为大写，并将结果转发到另一个目标。

@SpringBootApplication
@EnableBinding(Blog2Application.PolledProcessor.class)
public class Blog2Application {

  private final Logger logger =
  	  LoggerFactory.getLogger(Blog2Application.class);

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

  @Bean
  public ApplicationRunner runner(PollableMessageSource source,
  	    MessageChannel dest) {
    return args -> {
      while (true) {
        boolean result = source.poll(m -> {
          String payload = (String) m.getPayload();
          logger.info("Received: " + payload);
          dest.send(MessageBuilder.withPayload(payload.toUpperCase())
              .copyHeaders(m.getHeaders())
              .build());
        }, new ParameterizedTypeReference<String>() { });
        if (result) {
          logger.info("Processed a message");
        }
        else {
          logger.info("Nothing to do");
        }
        Thread.sleep(5_000);
      }
    };
  }

  public static interface PolledProcessor {

    @Input
    PollableMessageSource source();

    @Output
    MessageChannel dest();

  }

}

结论

应用程序现在可以控制消息的消费速率。

有关更多信息，请参阅参考手册中的使用轮询消费者。

我们鼓励您使用以下任一设施提供反馈

• 项目的 GitHub Issues
• Stack Overflow 频道
• Gitter 频道

享受！