Spring Cloud Stream 高階特性使用詳解

2022-09-08 18:04:56

重試

Consumer端可以設定重試次數，當訊息消費失敗的時候會進行重試。

底層使用Spring Retry去重試，重試次數可自定義設定。

# 預設重試次數為3，設定大於1時才會生效
spring.cloud.stream.bindings.<channelName>.consumer.maxAttempte=3

訊息傳送失敗的處理

Producer傳送訊息出錯的情況下，可以設定錯誤處理，將錯誤資訊傳送給對應ID的MessageChannel

訊息傳送失敗的場景下，會將訊息傳送到一個MessageChannel。這個MessageChannel會取ApplicationContext中name為topic.errors（topic就是設定的destination）的Bean。
如果找不到就會自動構建一個PublishSubscribeChannel。
然後使用BridgeHandler訂閱這個MessageChannel，同時再設定ApplicationContext中name為errorChannel的PublishSubscribeChannel訊息通道為BridgeHandler的outputChannel。

    public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel"
    private SubscribableChannel registerErrorInfrastructure(
            ProducerDestination destination) {
        // destination.getName() + ".errors"
        String errorChannelName = errorsBaseName(destination);
        SubscribableChannel errorChannel;
        if (getApplicationContext().containsBean(errorChannelName)) {
            Object errorChannelObject = getApplicationContext().getBean(errorChannelName);
            if (!(errorChannelObject instanceof SubscribableChannel)) {
                throw new IllegalStateException("Error channel '" + errorChannelName
                        + "' must be a SubscribableChannel");
            }
            errorChannel = (SubscribableChannel) errorChannelObject;
        }
        else {
            errorChannel = new PublishSubscribeChannel();
            ((GenericApplicationContext) getApplicationContext()).registerBean(
                    errorChannelName, SubscribableChannel.class, () -> errorChannel);
        }
        MessageChannel defaultErrorChannel = null;
        if (getApplicationContext()
                .containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
            defaultErrorChannel = getApplicationContext().getBean(
                    IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,
                    MessageChannel.class);
        }
        if (defaultErrorChannel != null) {
            BridgeHandler errorBridge = new BridgeHandler();
            errorBridge.setOutputChannel(defaultErrorChannel);
            errorChannel.subscribe(errorBridge);
            String errorBridgeHandlerName = getErrorBridgeName(destination);
            ((GenericApplicationContext) getApplicationContext()).registerBean(
                    errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
        }
        return errorChannel;
    }

範例程式碼

spring.cloud.stream.bindings.output.destination=test-output
# 訊息傳送失敗的處理邏輯預設是關閉的
spring.cloud.stream.bindings.output.producer.errorChannelEnabled=true

    @Bean("test-output.errors")
    MessageChannel testOutputErrorChannel() {
        return new PublishSubscribeChannel();
    }
    @Service
    class ErrorProduceService {
        @ServiceActivator(inputChannel = "test-output.errors")
        public void receiveProduceError(Message receiveMsg) {
            System.out.println("receive error msg: " + receiveMsg);
        }
    }

消費錯誤處理

Consumer消費訊息出錯的情況下，可以設定錯誤處理，將錯誤資訊發給對應ID的MessageChannel

訊息錯誤處理與生產錯誤處理大致相同。錯誤的MessageChannel對應的name為topic.group.errors，還會加上多個MessageHandler訂閱的一些判斷，使用ErrorMessageStrategy建立錯誤訊息等內容。

範例程式碼

spring.cloud.stream.bindings.input.destination=test-input
spring.cloud.stream.bindings.input.group=test-input-group

@StreamListener(Sink.INPUT)
public void receive(String receiveMsg) {
    throw new RuntimeException("Oops");
}
@ServiceActivator(inputChannel = "test-input.test-input-group.errors")
public void receiveConsumeError(Message receiveMsg) {
    System.out.println("receive error msg: " + receiveMsg);
}

建議直接使用topic.group.errors這個訊息通道，並設定傳送到單播模式的DirectChannel訊息通道中（使用@ServiceActivator註解接收會直接構成DirectChannel），這樣會確保只會被唯一的一個訂閱了topic.group.errors的MessageHandler處理，否則可能會被多個MessageHandler處理，導致出現一些意想不到的結果。

自定義MessageHandler型別

預設情況下，Output Binding對應的MessageChannel和Input Binding對應的SubscribeChannel會被構造成DirectChannel。

SCS提供了BindingTargetFactory介面進行擴充套件，比如可以擴充套件構造PublishSubscribeChannel這種廣播型別的MessageChannel。

BindingTargetFactory介面只有兩個實現類

SubscribableChannelBindingTargetFactory：針對Input Binding和Output Binding都會構造成DirectWithAttributesChannel型別的MessageChannel（一種帶有HashMap屬性的DirectChannel）。
MessageSourceBindingTargetFactory：不支援Output Binding，Input Binding會構造成DefaultPollableMessageSource。DefaultPollableMessageSource內部維護著MessageSource屬性，該屬性用於拉取訊息。

Endpoint端點

SCS提供了BindingsEndpoint，可以獲取Binding資訊或對Binding生命週期進行修改，比如start、stop、pause或resume。

BindingsEndpoint的ID是bindings，對外暴露了一下3個操作：

修改Binding狀態，可以改成STARTED、STOPPED、PAUSED和RESUMED，對應Binding介面的4個操作。
查詢單個Binding的狀態資訊。
查詢所有Binding的狀態資訊。

@Endpoint(id = "bindings")
public class BindingsEndpoint {
  ...
  @WriteOperation
    public void changeState(@Selector String name, State state) {
        Binding<?> binding = BindingsEndpoint.this.locateBinding(name);
        if (binding != null) {
            switch (state) {
            case STARTED:
                binding.start();
                break;
            case STOPPED:
                binding.stop();
                break;
            case PAUSED:
                binding.pause();
                break;
            case RESUMED:
                binding.resume();
                break;
            default:
                break;
            }
        }
    }
    @ReadOperation
    public List<?> queryStates() {
        List<Binding<?>> bindings = new ArrayList<>(gatherInputBindings());
        bindings.addAll(gatherOutputBindings());
        return this.objectMapper.convertValue(bindings, List.class);
    }
    @ReadOperation
    public Binding<?> queryState(@Selector String name) {
        Assert.notNull(name, "'name' must not be null");
        return this.locateBinding(name);
    }
  ...
}

Metrics指標

該功能自動與micrometer整合進行Metrics統計，可以通過字首spring.cloud.stream.metrics進行相關設定，設定項spring.cloud.stream.bindings.applicationMetrics.destination會構造MetersPublisherBinding，將相關的metrics傳送到MQ中。

Serverless

預設與Spring Cloud Function整合。

可以使用Function處理訊息。組態檔需要加上function設定。

spring.cloud.stream.function.definition=uppercase | addprefix

  @Bean
  public Function<String, String> uppercase() {
      return x -> x.toUpperCase();
  }
  @Bean
  public Function<String, String> addprefix() {
      return x -> "prefix-" + x;
  }

Partition統一

SCS統一Partition相關的設定，可以遮蔽不同MQ Partition的設定。

Producer Binding提供的ProducerProperties提供了一些Partition相關的設定：

partitionKeyExpression：partition key提取表示式。
partitionKeyExtractorName：是一個實現PartitionKeyExtractorStrategy介面的Bean name。PartitionKeyExtractorStrategy是一個根據Message獲取partition key的介面。如果兩者都設定，優先順序高於partitionKeyExtractorName。
partitionSelectorName：是一個實現PartitionSelectorStrategy介面的Bean name。PartitionSelectorStrategy是一個根據partition key決定選擇哪個partition 的介面。
partitionSelectorExpression：partition 選擇表示式，會根據表示式和partition key得到最終的partition。如果兩者都設定，優先partitionSelectorExpression表示式解析partition。
partitionCount：partition 個數。該屬性不一定會生效，Kafka Binder 和RocketMQ Binder會使用topic上的partition 個數覆蓋該屬性。

public final class PartitioningInterceptor implements ChannelInterceptor {
      ...
      @Override
      public Message<?> preSend(Message<?> message, MessageChannel channel) {
      if (!message.getHeaders().containsKey(BinderHeaders.PARTITION_OVERRIDE)) {
        int partition = this.partitionHandler.determinePartition(message);
        return MessageConverterConfigurer.this.messageBuilderFactory
          .fromMessage(message)
          .setHeader(BinderHeaders.PARTITION_HEADER, partition).build();
      }
      else {
        return MessageConverterConfigurer.this.messageBuilderFactory
          .fromMessage(message)
          .setHeader(BinderHeaders.PARTITION_HEADER,
                     message.getHeaders()
                     .get(BinderHeaders.PARTITION_OVERRIDE))
          .removeHeader(BinderHeaders.PARTITION_OVERRIDE).build();
      }
    }
}
public class PartitionHandler {
      ...
      public int determinePartition(Message<?> message) {
        Object key = extractKey(message);
        int partition;
        if (this.producerProperties.getPartitionSelectorExpression() != null) {
            partition = this.producerProperties.getPartitionSelectorExpression()
                    .getValue(this.evaluationContext, key, Integer.class);
        }
        else {
            partition = this.partitionSelectorStrategy.selectPartition(key,
                    this.partitionCount);
        }
        // protection in case a user selector returns a negative.
        return Math.abs(partition % this.partitionCount);
    }
    private Object extractKey(Message<?> message) {
        Object key = invokeKeyExtractor(message);
        if (key == null && this.producerProperties.getPartitionKeyExpression() != null) {
            key = this.producerProperties.getPartitionKeyExpression()
                    .getValue(this.evaluationContext, message);
        }
        Assert.notNull(key, "Partition key cannot be null");
        return key;
    }
      ...
}

Polling Consumer

實現MessageSource進行polling操作的Consumer。

普通的Pub/Sub模式需要定義SubscribeableChannel型別的返回值，Polling Consumer需要定義PollableMessageSource型別的返回值。

public interface PollableSink {
    /**
     * Input channel name.
     */
    String INPUT = "input";
    /**
     * @return input channel.
     */
    @Input(Sink.INPUT)
    PollableMessageSource input();
}

支援多個Binder同時使用

支援多個Binder同時使用，在設定Binding的時候需要指定對應的Binder。

設定全域性預設的Binder：spring.cloud.stream.default-binder=rocketmq。

設定各個Binder內部的設定資訊：

spring.cloud.stream.binders.rocketmq.environment.<xx>=xx

spring.cloud.stream.binders.rocketmq.type=rocketmq

設定Binding對應的Binder：

spring.cloud.stream.bindings.<channelName>.binder=kafka

spring.cloud.stream.bindings.<channelName>.binder=rocketmq

spring.cloud.stream.bindings.<channelName>.binder=rabbit