java - 屏障组件在 Spring 集成中如何工作?
问题描述
并使用 java DSL 重写它。
首先,我想了解该示例中发生了什么。当然,我已经阅读了 github 的解释,但我仍然不明白。我在这里问是因为我知道 SI 团队负责人(该示例的作者)回答了所有标记为 spring-integration 的问题。
服务器配置:
<int-http:inbound-gateway request-channel="receiveChannel"
path="/postGateway"
error-channel="errorChannel"
supported-methods="POST"/>
<int-http:inbound-gateway request-channel="createPayload"
path="/getGateway"
error-channel="errorChannel"
supported-methods="GET"/>
<int:transformer input-channel="createPayload" output-channel="receiveChannel" expression="'A,B,C'" />
<int:channel id="receiveChannel" />
<int:header-enricher input-channel="receiveChannel" output-channel="processChannel">
<int:header name="ackCorrelation" expression="headers['id']" />
</int:header-enricher>
<int:publish-subscribe-channel id="processChannel" />
<int:chain input-channel="processChannel" order="1">
<int:header-filter header-names="content-type, content-length" />
<int:splitter delimiters="," />
<int-amqp:outbound-channel-adapter amqp-template="rabbitTemplate"
exchange-name="barrier.sample.exchange" routing-key="barrier.sample.key"
confirm-ack-channel="confirmations"
confirm-nack-channel="confirmations"
return-channel="errorChannel"
confirm-correlation-expression="#this"/>
</int:chain>
<!-- Suspend the HTTP thread until the publisher confirms are asynchronously received -->
<int:barrier id="barrier" input-channel="processChannel" order="2"
correlation-strategy-expression="headers['ackCorrelation']"
output-channel="transform" timeout="10000" />
<int:transformer input-channel="transform" expression="payload[1]" />
<!-- Aggregate the publisher confirms and send the result to the barrier release channel -->
<int:chain input-channel="confirmations" output-channel="release">
<int:header-filter header-names="replyChannel, errorChannel" />
<int:service-activator expression="payload" /> <!-- INT-3791; use s-a to retain ack header -->
<int:aggregator>
<bean class="org.springframework.integration.samples.barrier.AckAggregator" />
</int:aggregator>
</int:chain>
<int:channel id="release" />
<int:outbound-channel-adapter channel="release" ref="barrier.handler" method="trigger" />
<!-- Consumer -> nullChannel -->
<int-amqp:inbound-channel-adapter channel="nullChannel"
queue-names="barrier.sample.queue"
connection-factory="rabbitConnectionFactory" />
<!-- Infrastructure -->
<rabbit:queue name="barrier.sample.queue" auto-delete="true" />
<rabbit:direct-exchange name="barrier.sample.exchange" auto-delete="true">
<rabbit:bindings>
<rabbit:binding queue="barrier.sample.queue" key="barrier.sample.key" />
</rabbit:bindings>
</rabbit:direct-exchange>
据我了解有3个派对:
- 客户
- 服务器
- 兔子Mq
客户端发送A,B,C
到服务器
RequestGateway requestGateway = client.getBean("requestGateway", RequestGateway.class);
String request = "A,B,C";
System.out.println("\n\n++++++++++++ Sending: " + request + " ++++++++++++\n");
String reply = requestGateway.echo(request);
服务器接受该请求:
<int-http:inbound-gateway request-channel="receiveChannel"
path="/postGateway"
error-channel="errorChannel"
supported-methods="POST"
然后将一些值添加到消息头中:
<int:header-enricher input-channel="receiveChannel" output-channel="processChannel">
<int:header name="ackCorrelation" expression="headers['id']" />
</int:header-enricher>
一些处理:
<int:chain input-channel="processChannel" order="1">
<int:header-filter header-names="content-type, content-length" />
<int:splitter delimiters="," />
<int-amqp:outbound-channel-adapter amqp-template="rabbitTemplate"
exchange-name="barrier.sample.exchange" routing-key="barrier.sample.key"
confirm-ack-channel="confirmations"
confirm-nack-channel="confirmations"
return-channel="errorChannel"
confirm-correlation-expression="#this"/>
</int:chain>
如果我没记错的话 - 消息被分割,,
所以我们有 3 条消息并发A
B
送到C
rabbitMq 交换barrier.sample.exchange
keybarrier.sample.key
这么。下一步我不清楚。你能澄清一下吗?
附言
我知道我们希望等待来自 rabbit 的所有 ACK 消息,以确保我们传递给 rabbit 的消息以及当我们从 rabbit 获得所有 ACK 时,我们想要回复客户端。但我不了解导致该结果的 SI 组件定义。您能否提供任何模式/图片/解释它是如何发生的?
更具体的问题:
为什么我们需要丰富的标题以及我们如何做到这一点?
receiveChannel的目的是什么?- 愚蠢的问题
2 个 http 服务在哪里 -/postGateway
和/getGateway
. /postGateway
用于从客户端获取 msg,拆分为部分,发送给 rabbit,获取 Acks,然后回复客户端。但目的是/getGateway
什么?
附言
现在我的流程声明看起来像这样:
@Bean
public IntegrationFlow integrationFlow() {
return IntegrationFlows.from(Http.inboundGateway("/spring_integration_post")
.requestMapping(m -> m.methods(HttpMethod.POST))
.requestPayloadType(String.class))
.enrich(enricherSpec -> {
enricherSpec.header("correlationId", 1); //or ackCorrelationId ?
})
.split(s -> s.applySequence(false).get().getT2().setDelimiters(","))
.log()
//.barrier(1000L) is it correct place for barrier?
.log()
.handle(Amqp.outboundAdapter(amqpTemplate())
.exchangeName("barrierExchange")
.routingKey("barrierKey"))
.get();
}
我不知道如何在这里使用屏障。
从服务器端,一切都在部分工作 - 我看到队列中的消息(在rabbitMq Web界面上)但从客户端我开始看到以下堆栈跟踪:
2019-08-28 22:38:43.432 ERROR 12936 --- [ask-scheduler-8] o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageHandlingException: HTTP request execution failed for URI [http://localhost:8080/spring_integration_post]; nested exception is org.springframework.web.client.HttpServerErrorException$InternalServerError: 500 null, failedMessage=GenericMessage [payload=6, headers={id=36781a7f-3d4f-e17d-60e6-33450c9307e4, timestamp=1567021122424}]
at org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler.exchange(HttpRequestExecutingMessageHandler.java:171)
at org.springframework.integration.http.outbound.AbstractHttpRequestExecutingMessageHandler.handleRequestMessage(AbstractHttpRequestExecutingMessageHandler.java:289)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:123)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:132)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:401)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
at org.springframework.integration.endpoint.SourcePollingChannelAdapter.handleMessage(SourcePollingChannelAdapter.java:234)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:390)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:329)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$1(AbstractPollingEndpoint.java:277)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$2(AbstractPollingEndpoint.java:274)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.springframework.web.client.HttpServerErrorException$InternalServerError: 500 null
at org.springframework.web.client.HttpServerErrorException.create(HttpServerErrorException.java:79)
at org.springframework.web.client.DefaultResponseErrorHandler.handleError(DefaultResponseErrorHandler.java:124)
at org.springframework.web.client.DefaultResponseErrorHandler.handleError(DefaultResponseErrorHandler.java:102)
at org.springframework.web.client.ResponseErrorHandler.handleError(ResponseErrorHandler.java:63)
at org.springframework.web.client.RestTemplate.handleResponse(RestTemplate.java:778)
at org.springframework.web.client.RestTemplate.doExecute(RestTemplate.java:736)
at org.springframework.web.client.RestTemplate.execute(RestTemplate.java:710)
at org.springframework.web.client.RestTemplate.exchange(RestTemplate.java:598)
at org.springframework.integration.http.outbound.HttpRequestExecutingMessageHandler.exchange(HttpRequestExecutingMessageHandler.java:165)
... 30 more
解决方案
这很简单;不清楚你不明白什么。
- 我们收到来自 http 的请求
- 我们向消息中添加相关标头(标头丰富器)
- 我们向rabbitmq发送3条消息
- 我们需要等待确认 - 因此 HTTP 线程在屏障中暂停
- 3个确认异步返回
- 我们将响应聚合到一条消息中(使用相关标头)
- 我们将该消息发送到释放 HTTP 线程的屏障
这receiveChannel
是标题丰富器的输入通道。
您需要添加流以接收确认、聚合和触发屏障。
推荐阅读
- angular - 如何在 2019 年 10 月使用 Jest 设置 Angular?
- python - 依赖于 (k-1) 个先前项的递归函数
- javascript - 为什么我的侧边导航菜单不会被这个 jquery 幻灯片效果隐藏?
- python - 如何加密字典数据?
- java - JMS 队列是 Java Util 队列的实现吗?使用 Java Util Queue 实现的 Java 包中有哪些类?
- ubuntu - Ubuntu 14.04.5 LTS:监视警报 -- 连接失败 sshd
- javascript - JavaScript 在添加时崩溃
- excel - 为不同的 Excel 工作簿执行相同的代码时发生运行时错误 1004
- postgresql - postgres 中排除约束的要点
- sql-server - 在 SQL Server 的一个过程中使用 UNION ALL 从不同数据库中的两个不同函数获取记录