SpringCloudStream高級(jí)特性使用詳解

第SpringCloudStream高級(jí)特性使用詳解目錄重試消息發(fā)送失敗的處理消費(fèi)錯(cuò)誤處理自定義MessageHandler類型Endpoint端點(diǎn)Metrics指標(biāo)ServerlessPartition統(tǒng)一PollingConsumer支持多個(gè)Binder同時(shí)使用建立事件機(jī)制

重試

Consumer端可以配置重試次數(shù)，當(dāng)消息消費(fèi)失敗的時(shí)候會(huì)進(jìn)行重試。

底層使用SpringRetry去重試，重試次數(shù)可自定義配置。

#默認(rèn)重試次數(shù)為3，配置大于1時(shí)才會(huì)生效

spring.cloud.stream.bindings.channelName.consumer.maxAttempte=3

消息發(fā)送失敗的處理

Producer發(fā)送消息出錯(cuò)的情況下，可以配置錯(cuò)誤處理，將錯(cuò)誤信息發(fā)送給對(duì)應(yīng)ID的MessageChannel

消息發(fā)送失敗的場(chǎng)景下，會(huì)將消息發(fā)送到一個(gè)MessageChannel。這個(gè)MessageChannel會(huì)取ApplicationContext中name為topic.errors（topic就是配置的destination）的Bean。如果找不到就會(huì)自動(dòng)構(gòu)建一個(gè)PublishSubscribeChannel。然后使用BridgeHandler訂閱這個(gè)MessageChannel，同時(shí)再設(shè)置ApplicationContext中name為errorChannel的PublishSubscribeChannel消息通道為BridgeHandler的outputChannel。

publicstaticfinalStringERROR_CHANNEL_BEAN_NAME="errorChannel"

privateSubscribableChannelregisterErrorInfrastructure(

ProducerDestinationdestination){

//destination.getName()+".errors"

StringerrorChannelName=errorsBaseName(destination);

SubscribableChannelerrorChannel;

if(getApplicationContext().containsBean(errorChannelName)){

ObjecterrorChannelObject=getApplicationContext().getBean(errorChannelName);

if(!(errorChannelObjectinstanceofSubscribableChannel)){

thrownewIllegalStateException("Errorchannel'"+errorChannelName

+"'mustbeaSubscribableChannel");

errorChannel=(SubscribableChannel)errorChannelObject;

else{

errorChannel=newPublishSubscribeChannel();

((GenericApplicationContext)getApplicationContext()).registerBean(

errorChannelName,SubscribableChannel.class,()-errorChannel);

MessageChanneldefaultErrorChannel=null;

if(getApplicationContext()

.containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)){

defaultErrorChannel=getApplicationContext().getBean(

IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,

MessageChannel.class);

if(defaultErrorChannel!=null){

BridgeHandlererrorBridge=newBridgeHandler();

errorBridge.setOutputChannel(defaultErrorChannel);

errorChannel.subscribe(errorBridge);

StringerrorBridgeHandlerName=getErrorBridgeName(destination);

((GenericApplicationContext)getApplicationContext()).registerBean(

errorBridgeHandlerName,BridgeHandler.class,()-errorBridge);

returnerrorChannel;

spring.cloud.stream.bindings.output.destination=test-output

#消息發(fā)送失敗的處理邏輯默認(rèn)是關(guān)閉的

ducer.errorChannelEnabled=true

@Bean("test-output.errors")

MessageChanneltestOutputErrorChannel(){

returnnewPublishSubscribeChannel();

@Service

classErrorProduceService{

@ServiceActivator(inputChannel="test-output.errors")

publicvoidreceiveProduceError(MessagereceiveMsg){

System.out.println("receiveerrormsg:"+receiveMsg);

消費(fèi)錯(cuò)誤處理

Consumer消費(fèi)消息出錯(cuò)的情況下，可以配置錯(cuò)誤處理，將錯(cuò)誤信息發(fā)給對(duì)應(yīng)ID的MessageChannel

消息錯(cuò)誤處理與生產(chǎn)錯(cuò)誤處理大致相同。錯(cuò)誤的MessageChannel對(duì)應(yīng)的name為topic.group.errors，還會(huì)加上多個(gè)MessageHandler訂閱的一些判斷，使用ErrorMessageStrategy創(chuàng)建錯(cuò)誤消息等內(nèi)容。

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

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

@StreamListener(Sink.INPUT)

publicvoidreceive(StringreceiveMsg){

thrownewRuntimeException("Oops");

@ServiceActivator(inputChannel="test-input.test-input-group.errors")

publicvoidreceiveConsumeError(MessagereceiveMsg){

System.out.println("receiveerrormsg:"+receiveMsg);

建議直接使用topic.group.errors這個(gè)消息通道，并設(shè)置發(fā)送到單播模式的DirectChannel消息通道中（使用@ServiceActivator注解接收會(huì)直接構(gòu)成DirectChannel），這樣會(huì)確保只會(huì)被唯一的一個(gè)訂閱了topic.group.errors的MessageHandler處理，否則可能會(huì)被多個(gè)MessageHandler處理，導(dǎo)致出現(xiàn)一些意想不到的結(jié)果。

自定義MessageHandler類型

默認(rèn)情況下，OutputBinding對(duì)應(yīng)的MessageChannel和InputBinding對(duì)應(yīng)的SubscribeChannel會(huì)被構(gòu)造成DirectChannel。

SCS提供了BindingTargetFactory接口進(jìn)行擴(kuò)展，比如可以擴(kuò)展構(gòu)造PublishSubscribeChannel這種廣播類型的MessageChannel。

BindingTargetFactory接口只有兩個(gè)實(shí)現(xiàn)類

SubscribableChannelBindingTargetFactory：針對(duì)InputBinding和OutputBinding都會(huì)構(gòu)造成DirectWithAttributesChannel類型的MessageChannel（一種帶有HashMap屬性的DirectChannel）。MessageSourceBindingTargetFactory：不支持OutputBinding，InputBinding會(huì)構(gòu)造成DefaultPollableMessageSource。DefaultPollableMessageSource內(nèi)部維護(hù)著MessageSource屬性，該屬性用于拉取消息。

Endpoint端點(diǎn)

SCS提供了BindingsEndpoint，可以獲取Binding信息或?qū)inding生命周期進(jìn)行修改，比如start、stop、pause或resume。

BindingsEndpoint的ID是bindings，對(duì)外暴露了一下3個(gè)操作：

修改Binding狀態(tài)，可以改成STARTED、STOPPED、PAUSED和RESUMED，對(duì)應(yīng)Binding接口的4個(gè)操作。查詢單個(gè)Binding的狀態(tài)信息。查詢所有Binding的狀態(tài)信息。

@Endpoint(id="bindings")

publicclassBindingsEndpoint{

@WriteOperation

publicvoidchangeState(@SelectorStringname,Statestate){

Bindingbinding=BindingsEndpoint.this.locateBinding(name);

if(binding!=null){

switch(state){

caseSTARTED:

binding.start();

break;

caseSTOPPED:

binding.stop();

break;

casePAUSED:

binding.pause();

break;

caseRESUMED:

binding.resume();

break;

default:

break;

@ReadOperation

publicListqueryStates(){

ListBindingbindings=newArrayList(gatherInputBindings());

bindings.addAll(gatherOutputBindings());

returnthis.objectMapper.convertValue(bindings,List.class);

@ReadOperation

publicBindingqueryState(@SelectorStringname){

Assert.notNull(name,"'name'mustnotbenull");

returnthis.locateBinding(name);

Metrics指標(biāo)

該功能自動(dòng)與micrometer集成進(jìn)行Metrics統(tǒng)計(jì)，可以通過(guò)前綴spring.cloud.stream.metrics進(jìn)行相關(guān)配置，配置項(xiàng)spring.cloud.stream.bindings.applicationMetrics.destination會(huì)構(gòu)造MetersPublisherBinding，將相關(guān)的metrics發(fā)送到MQ中。

Serverless

默認(rèn)與SpringCloudFunction集成。

可以使用Function處理消息。配置文件需要加上function配置。

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

@Bean

publicFunctionString,Stringuppercase(){

returnx-x.toUpperCase();

@Bean

publicFunctionString,Stringaddprefix(){

returnx-"prefix-"+x;

Partition統(tǒng)一

SCS統(tǒng)一Partition相關(guān)的設(shè)置，可以屏蔽不同MQPartition的設(shè)置。

ProducerBinding提供的ProducerProperties提供了一些Partition相關(guān)的配置：

partitionKeyExpression：partitionkey提取表達(dá)式。partitionKeyExtractorName：是一個(gè)實(shí)現(xiàn)PartitionKeyExtractorStrategy接口的Beanname。PartitionKeyExtractorStrategy是一個(gè)根據(jù)Message獲取partitionkey的接口。如果兩者都配置，優(yōu)先級(jí)高于partitionKeyExtractorName。partitionSelectorName：是一個(gè)實(shí)現(xiàn)PartitionSelectorStrategy接口的Beanname。PartitionSelectorStrategy是一個(gè)根據(jù)partitionkey決定選擇哪個(gè)partition的接口。partitionSelectorExpression：partition選擇表達(dá)式，會(huì)根據(jù)表達(dá)式和partitionkey得到最終的partition。如果兩者都配置，優(yōu)先partitionSelectorExpression表達(dá)式解析partition。partitionCount：partition個(gè)數(shù)。該屬性不一定會(huì)生效，KafkaBinder和RocketMQBinder會(huì)使用topic上的partition個(gè)數(shù)覆蓋該屬性。

publicfinalclassPartitioningInterceptorimplementsChannelInterceptor{

@Override

publicMessagepreSend(Messagemessage,MessageChannelchannel){

if(!message.getHeaders().containsKey(BinderHeaders.PARTITION_OVERRIDE)){

intpartition=this.partitionHandler.determinePartition(message);

returnMessageConverterConfigurer.this.messageBuilderFactory

.fromMessage(message)

.setHeader(BinderHeaders.PARTITION_HEADER,partition).build();

else{

returnMessageConverterConfigurer.this.messageBuilderFactory

.fromMessage(message)

.setHeader(BinderHeaders.PARTITION_HEADER,

message.getHeaders()

.get(BinderHeaders.PARTITION_OVERRIDE))

.removeHeader(BinderHeaders.PARTITION_OVERRIDE).build();

publicclassPartitionHandler{

publicintdeterminePartition(Messagemessage){

Objectkey=extractKey(message);

intpartition;

if(ducerProperties.getPartitionSelectorExpression()!=null){

partition=ducerProperties.getPartitionSelectorExpression()

.getValue(this.evaluationContext,key,Integer.class);

else{

partition=this.partitionSelectorStrategy.selectPartition(key,

this.partitionCount);

//protectionincaseauserselectorreturnsanegative.

returnMath.abs(partition%this.partitionCount);

privateObjectextractKey(Messagemessage){

Objectkey=invokeKeyExtractor(message);

if(key==nullducerProperties.getPartitionKeyExpression()!=null){

key=ducerProperties.getPartitionKeyExpression()

.getValue(this.evaluationContext,message);

Assert.notNull(key,"Partitionkeycannotbenull");

returnkey;

PollingConsumer

實(shí)現(xiàn)MessageSource進(jìn)行polling操作的Consumer。

普通的Pub/Sub模式需要定義Subs