Смотрите также мой ответ.
работал асинхронные службы Spring Cloud Stream и пытаюсь разработать пограничный сервис, который использует @MessagingGateway для обеспечения синхронного доступа к асинхронным службам по своей природе.
В настоящее время я получаю следующую трассировку стека:
Caused by: org.springframework.messaging.core.DestinationResolutionException: no output-channel or replyChannel header available
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:355)
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:271)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:188)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:115)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
... 47 common frames omitted
Мой @MessagingGateway:
@EnableBinding(AccountChannels.class)
@MessagingGateway
public interface AccountService {
@Gateway(requestChannel = AccountChannels.CREATE_ACCOUNT_REQUEST,replyChannel = AccountChannels.ACCOUNT_CREATED, replyTimeout = 60000, requestTimeout = 60000)
Account createAccount(@Payload Account account, @Header("Authorization") String authorization);
}
Если я получаю сообщение в ответном канале через @StreamListener, оно работает просто отлично:
@HystrixCommand(commandKey = "acounts-edge:accountCreated", fallbackMethod = "accountCreatedFallback", commandProperties = {@HystrixProperty(name = "execution.isolation.strategy", value = "SEMAPHORE")}, ignoreExceptions = {ClientException.class})
@StreamListener(AccountChannels.ACCOUNT_CREATED)
public void accountCreated(Account account, @Header(name = "spanTraceId", required = false) String traceId) {
try {
if (log.isInfoEnabled()) {
log.info(new StringBuilder("Account created: ").append(objectMapper.writeValueAsString(account)).toString());
}
} catch (JsonProcessingException e) {
log.error(e.getMessage(), e);
}
}
На стороне производителя я настраиваюrequiredGroups
чтобы несколько потребителей могли обработать сообщение, и соответственноgroup
конфигурации.
Потребитель:
spring:
cloud:
stream:
bindings:
create-account-request:
binder: rabbit1
contentType: application/json
destination: create-account-request
requiredGroups: accounts-service-create-account-request
account-created:
binder: rabbit1
contentType: application/json
destination: account-created
group: accounts-edge-account-created
Режиссер:
spring:
cloud:
stream:
bindings:
create-account-request:
binder: rabbit1
contentType: application/json
destination: create-account-request
group: accounts-service-create-account-request
account-created:
binder: rabbit1
contentType: application/json
destination: account-created
requiredGroups: accounts-edge-account-created
Бит кода на стороне производителя, который обрабатывает запрос и отправляет ответ:
accountChannels.accountCreated().send(MessageBuilder.withPayload(accountService.createAccount(account)).build());
Я могу отладить и увидеть, что запрос получен и обработан, но когда ответ отправляется на канал ответа, именно тогда происходит ошибка.
Чтобы заставить @MessagingGateway работать, какие конфигурации и / или код мне не хватает? Я знаю, что объединяю Spring Integration и Spring Cloud Gateway, поэтому не уверен, что их совместное использование вызывает проблемы.