Spring Cloud Stream总结

概念

一、group：

组内只有1个实例消费。若是不设置group，则stream会自动为每一个实例建立匿名且独立的group——因而每一个实例都会消费

组内单次只有1个实例消费，而且会轮询负载均衡。一般，在将应用程序绑定到给定目标时，最好始终指定consumer group

二、destination binder：

与外部消息系统通讯的组件，为构造 Binding提供了 2 个方法，分别是 bindConsumer 和 bindProducer ，它们分别用于构造生产者和消费者。Binder使Spring Cloud Stream应用程序能够灵活地链接到中间件，目前spring为kafka、rabbitmq提供binder

三、destination binding：

Binding 是链接应用程序跟消息中间件的桥梁，用于消息的消费和生产，由binder建立

四、partition

一个或多个生产者将数据发送到多个消费者，并确保有共同特征标识的数据由同一个消费者处理。默认是对消息进行hashCode，而后根据分区个数取余，因此对于相同的消息，总会落到同一个消费者上

注：严格来讲partition不属于概念，而是一种Stream提升伸缩性、吞吐量的一种方式

---

注解

一、@Input，使用示例：

public interface MySink {
    @Input("my-input")
    SubscribableChannel input();
}

做用：

• 用于接收消息
• 为每一个binding生成channel实例
• 指定input channel的名称
• 在spring容器中生成一个名为my-input，类型为SubscribableChannel的bean
• 在spring容器中生成一个类，实现MySink接口。

二、@Output，使用示例：

public interface MySource {
    @Output("my-output")
    MessageChannel output();
}

做用：

• 与@Input相似，只不过是用来生产消息

三、@StreamListener，使用示例：

@StreamListener(value = Sink.INPUT, condition = "headers['type']=='dog'")
public void receive(String messageBody) {
    log.info("Received: {}", messageBody);
}

做用：

• 用于消费消息
• condition的做用：用于过滤消息，只有符合条件表达式的消息才会被处理
• condition起做用的两个条件：
  • 注解的方法没有返回值
  • 方法是一个独立方法，不支持Reactive API

四、@SendTo，使用示例：

// 接收INPUT这个channel的消息，并将返回值发送到OUTPUT这个channel
@StreamListener(Sink.INPUT)
@SendTo(Source.OUTPUT)
public String receive(String receiveMsg) {
   return "handle...";
}

做用：

• 用于发送消息

四、@InboundChannelAdapter，使用示例：

@Bean
@InboundChannelAdapter(value = Source.OUTPUT,
        poller = @Poller(fixedDelay = "10", maxMessagesPerPoll = "1"))
public MessageSource<String> producer() {
    return () -> new GenericMessage<>("Hello Spring Cloud Stream");
}

做用：

• 让添加该注解的方法生产消息
• fixedDelay：多少毫秒发送1次
• maxMessagesPerPoll：每次发送多少条消息

五、@ServiceActivator，使用示例：

@ServiceActivator(inputChannel = Sink.INPUT, outputChannel = Source.OUTPUT)
public String transform(String payload) {
    return payload.toUpperCase();
}

做用：

• 标注该注解的方法可以处理消息或消息的有效内容，经过监听input消息，用方法体的代码处理后，输出到output中

六、@Transformer，使用示例：

@Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public Object transform(String message) {
  return message.toUpperCase();
}

做用：

• 与@ServiceActivator相似，标注该注解的方法可以转换消息，消息头，或消息有效内容

---

PollableMessageSource

PollableMessageSource容许消费者能够控制消费速率。举个例子简单演示一下，首先定义一个接口：

public interface PolledProcessor {
    @Input("pollable-input")
    PollableMessageSource input();
}

使用示例：

@Autowired
private PolledProcessor polledProcessor;

@Scheduled(fixedDelay = 5_000)
public void poll() {
    polledProcessor.input().poll(message -> {
        byte[] bytes = (byte[]) message.getPayload();
        String payload = new String(bytes);
        System.out.println(payload);
    });
}

---

参考：

https://spring.io/blog/2018/02/27/spring-cloud-stream-2-0-polled-consumers