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