Как подключить входящий адаптер AWS SQS к каналу сообщений с поддержкой AMQP?

Я успешно настроил через Java адаптер входящего канала для AWS SQS с использованием прямого канала. В этом проекте используется комбинация JDBC и RabbitMQ с SQS в качестве входящего потока для 3 отдельных очередей. Мне нужны надежные сообщения, поэтому я пытаюсь понять, как использовать RabbitMQ. Я не понимаю, как ссылаться на канал AMQP. Как мне сослаться на канал сообщений AMQP из setOutPutChannel? Моя цель - удалить сообщение из SQS только в том случае, если сообщение успешно опубликовано в устойчивой очереди RabbitMQ.

@Bean
public MessageProducer getSQSChannel() {
    SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(this.amazonSqs, MY_SQS_QUEUE);
    adapter.setOutputChannel(????);  
    return adapter;
}

@Bean
public AmqpChannelFactoryBean messageDriven(ConnectionFactory connectionFactory) {
    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true);
    factoryBean.setConnectionFactory(connectionFactory);
    factoryBean.setQueueName("bar");
    factoryBean.setPubSub(false);
    return factoryBean;
}

spring-integration spring-integration-aws spring-integration-amqp
person Jim Hankins    schedule 12.01.2019    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Использовать

adapter.setOutputChannelName("messageDriven");

и канал, созданный фабричным компонентом, будет определяться по его имени во время выполнения.

ИЗМЕНИТЬ

Или, как сказал Артем ...

MessageProducer getSQSChannel(MessageChannel messageDriven) { 
    ... 
}
person Gary Russell    schedule 12.01.2019
comment
Вы также можете выполнить инъекцию аргумента метода MessageProducer getSQSChannel(MessageChannel messageDriven) { ... } - person Artem Bilan; 12.01.2019
comment
Спасибо! Чтобы подтвердить мое понимание. Если у меня есть поток интеграции, который, например, опрашивает сообщения SQS, преобразует, а затем отправляет RabbitMQ с DirectChannel в качестве входящего канала, существуют сценарии, при которых сообщение может быть потеряно? SQS ››› InboundChannelAdapter ››› DirectChannel ›› ObJTransformer ›› PayloadEnricher ›› RabbitOutboundAdapter В моем тестировании выполнения таких действий, как убийство кролика, установка политики удаления адаптеров sqs на onSuccess, сообщение не удаляется правильно (но увеличивает счетчик прочтений ). Так безопасно ли использовать прямой канал в этом сценарии? - person Jim Hankins; 13.01.2019
comment
Вы не должны задавать несвязанные вопросы в комментариях к существующим ответам; это не помогает сообществу находить вопросы / ответы. DirectChannel безопасен в том смысле, что публикация будет происходить в вызывающем потоке. Я не знаком с расширением SQS, поэтому я не могу комментировать, когда применяется политика onSuccess и будет ли она работать в этом случае (я предполагаю, что это может быть нормально, потому что это адаптер, управляемый сообщениями, поэтому поток выполняется на адаптере нить). - person Gary Russell; 13.01.2019
comment
Тем не менее, публикация RabbitMQ в любом случае является асинхронной, поэтому вам нужно будет использовать транзакцию RabbitMQ, чтобы брокер принял сообщение; транзакции являются значительными накладными расходами при публикации RabbitMQ; особенно при публикации только одного сообщения. - person Gary Russell; 13.01.2019
comment
Спасибо, Гэри. Я буду иметь это в виду. Честно говоря, я указал в своей цели вопроса о желании не потерять сообщение, поэтому я бы сказал, что это связано с вопросом. Но я понимаю, что по этой конкретной теме может быть отдельная тема, поэтому я буду иметь это в виду. Еще раз спасибо! - person Jim Hankins; 13.01.2019