【发布时间】：2018-06-05 12:25:23
【问题描述】：
关于为How can @MessagingGateway be configured with Spring Cloud Stream MessageChannels? 发布的答案，处理可以从 Spring Cloud Stream 服务返回的@MessagingGateway 错误的正确方法是什么？
回顾一下，我有一个@MessagingGateway，它提供对使用 Spring Cloud Stream 构建的异步服务的同步访问。当 Spring Cloud Stream 服务层中发生错误时，我会创建一个错误响应并通过 SubscribableChannel 将其发送到处理错误的其他 @StreamListener 服务。
例如，当创建帐户时，我向accountCreated 频道发送消息。当发生错误时，我会向accountNotCreated 频道发送错误响应。
这很好用，但我也想向@MessagingGateway 的客户端发送错误响应，以便他们同步接收错误响应。 @MessagingGateway 注释具有errorChannel 属性，但@Gateway 注释没有。因此，@MessagingGateway 的客户端应该能够阻止并等待 1) 创建帐户或 2) 错误响应。
同样，这里的目标是构建“后端”服务，将 Spring Cloud Stream 用于事务服务（即创建、更新或删除数据的服务），同时为我们的客户提供“网关”访问该块并等待返回响应。 Artem Bilan 为我提供的解决方案适用于快乐的道路，但是当发生错误时，我不清楚 Spring Integration 如何最适合处理此问题。
更新代码示例
GatewayApplication.java
package com.example.demo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.dsl.HeaderEnricherSpec;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
@EnableBinding({GatewayApplication.GatewayChannels.class})
@SpringBootApplication
public class GatewayApplication {
@Component
public interface GatewayChannels {
String TO_UPPERCASE_REPLY = "to-uppercase-reply";
String TO_UPPERCASE_REQUEST = "to-uppercase-request";
@Input(TO_UPPERCASE_REPLY)
SubscribableChannel toUppercaseReply();
@Output(TO_UPPERCASE_REQUEST)
SubscribableChannel toUppercaseRequest();
}
@MessagingGateway
public interface StreamGateway {
public static final String ENRICH = "enrich";
@Gateway(requestChannel = ENRICH, replyChannel = GatewayChannels.TO_UPPERCASE_REPLY)
StringWrapper process(StringWrapper payload) throws MyException;
}
@RestController
public class UppercaseController {
@Autowired
StreamGateway gateway;
@GetMapping(value = "/string/{string}",
produces = {MediaType.APPLICATION_JSON_VALUE, MediaType.APPLICATION_XML_VALUE})
public ResponseEntity<StringWrapper> getUser(@PathVariable("string") String string) {
try {
StringWrapper result = gateway.process(new StringWrapper(string));
// Instead of catching the exception in the below catch clause, here we have just a string
// representation of the stack trace when an exception occurs.
return new ResponseEntity<StringWrapper>(result, HttpStatus.OK);
} catch (MyException e) {
// Why is the exception not caught here?
return new ResponseEntity<StringWrapper>(new StringWrapper("An error has occurred"),
HttpStatus.INTERNAL_SERVER_ERROR);
}
}
}
public static void main(String[] args) {
SpringApplication.run(GatewayApplication.class, args);
}
@Bean
public IntegrationFlow headerEnricherFlow() {
return IntegrationFlows.from(StreamGateway.ENRICH)
.enrichHeaders(HeaderEnricherSpec::headerChannelsToString)
.channel(GatewayChannels.TO_UPPERCASE_REQUEST).get();
}
}
application.yml
spring:
cloud:
stream:
bindings:
to-uppercase-request:
destination: to-uppercase-request
content-type: application/json
group: stream-to-uppercase-request
to-uppercase-reply:
destination: to-uppercase-reply
content-type: application/json
producer:
required-groups: gateway-to-uppercase-reply, stream-listener-to-uppercase-reply
server:
port: 8088
StringWrapper.java（在所有三个项目中使用）
package com.example.demo;
import com.fasterxml.jackson.annotation.JsonProperty;
public class StringWrapper {
@JsonProperty
private String string;
@JsonProperty
private long time = System.currentTimeMillis();
public StringWrapper() {
super();
}
public StringWrapper(String string) {
this.string = string;
}
public String getString() {
return string;
}
public long getTime() {
return time;
}
public void setString(String string) {
this.string = string;
}
@Override
public String toString() {
return "StringWrapper [string=" + string + ", time=" + time + "]";
}
}
CloudStreamApplication.java
package com.example.demo;
import java.util.Random;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
@EnableBinding({CloudStreamApplication.CloudStreamChannels.class})
@SpringBootApplication
public class CloudStreamApplication {
@Component
interface CloudStreamChannels {
String TO_UPPERCASE_REPLY = "to-uppercase-reply";
String TO_UPPERCASE_REQUEST = "to-uppercase-request";
@Output(TO_UPPERCASE_REPLY)
SubscribableChannel toUppercaseReply();
@Input(TO_UPPERCASE_REQUEST)
SubscribableChannel toUppercaseRequest();
}
@Component
public class Processor {
@Autowired
CloudStreamChannels channels;
@StreamListener(CloudStreamChannels.TO_UPPERCASE_REQUEST)
public void process(Message<StringWrapper> request) {
StringWrapper uppercase = null;
try {
uppercase = toUppercase(request);
} catch (MyException e) {
channels.toUppercaseReply()
.send(MessageBuilder.withPayload(e).setHeader("__TypeId__", e.getClass().getName())
.copyHeaders(request.getHeaders()).build());
}
if (uppercase != null) {
channels.toUppercaseReply()
.send(MessageBuilder.withPayload(uppercase)
.setHeader("__TypeId__", StringWrapper.class.getName())
.copyHeaders(request.getHeaders()).build());
}
}
private StringWrapper toUppercase(Message<StringWrapper> request) throws MyException {
Random random = new Random();
int number = random.nextInt(50) + 1;
if (number > 25) {
throw new MyException("An error occurred.");
}
return new StringWrapper(request.getPayload().getString().toUpperCase());
}
}
public static void main(String[] args) {
SpringApplication.run(CloudStreamApplication.class, args);
}
}
application.yml
spring:
cloud:
stream:
bindings:
to-uppercase-request:
destination: to-uppercase-request
content-type: application/json
group: stream-to-uppercase-request
to-uppercase-reply:
destination: to-uppercase-reply
content-type: application/json
producer:
required-groups: gateway-to-uppercase-reply, stream-listener-to-uppercase-reply
server:
port: 8088
StreamListenerApplication.java
package com.example.demo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;
@EnableBinding({StreamListenerApplication.CloudStreamChannels.class})
@SpringBootApplication
public class StreamListenerApplication {
@Component
interface CloudStreamChannels {
String TO_UPPERCASE_REPLY = "to-uppercase-reply";
@Input(TO_UPPERCASE_REPLY)
SubscribableChannel toUppercaseReply();
}
public static void main(String[] args) {
SpringApplication.run(StreamListenerApplication.class, args);
}
@Autowired
CloudStreamChannels channels;
@StreamListener(CloudStreamChannels.TO_UPPERCASE_REPLY)
public void processToUppercaseReply(Message<StringWrapper> message) {
System.out.println("Processing message: " + message.getPayload());
}
}
application.yml
spring:
cloud:
stream:
bindings:
to-uppercase-reply:
destination: to-uppercase-reply
content-type: application/json
group: stream-listener-to-uppercase-reply
server:
port: 8089
【问题讨论】：
-
您应该接受 Artem 对其他问题的回答（或者至少给他投票，这样如果您决定接受自己的总结最终解决方案的回答，他就会得到一些信任）。见What should I do when someone answers my question?。
标签： spring-integration spring-cloud-stream