Spring Cloud Stream: @StreamListener обрабатывает сообщения дважды

Question

Я использую Spring Cloud Stream (Edgware.SR5) с Spring Boot (1.5.10.RELEASE). Мой @StreamListener обрабатывает дважды каждое полученное сообщение.

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

Услуги:

@EnableBinding(ExampleBindings.class)
@Service
public class ExampleService {

    @Publisher(channel = ExampleBindings.OUTPUT)
    public String queue(String message){
        return message;
    }

    @StreamListener(ExampleBindings.INPUT)
    public void dequeue(String message){
        System.out.println("New message: " + message);
    }
}

привязки:

public interface ExampleBindings {

    String INPUT = "input1";
    String OUTPUT = "output1";

    @Input(ExampleBindings.INPUT)
    SubscribableChannel input();

    @Output(ExampleBindings.OUTPUT)
    MessageChannel output();
}

application.properties:

spring.cloud.stream.default.group=group1
spring.cloud.stream.default.binder=binder1

spring.cloud.stream.bindings.input1.destination=dest_1
spring.cloud.stream.bindings.output1.destination=dest_1

spring.cloud.stream.binders.binder1.type=rabbit
spring.cloud.stream.binders.binder1.environment.spring.rabbitmq.host=localhost

Конфигурация (для внедрения прокси-сервиса в тесте):

@Configuration
public class ExampleConfig {

    @Bean
    public PublisherAnnotationBeanPostProcessor publisherAnnotationBeanPostProcessor(){
         PublisherAnnotationBeanPostProcessor publisherAnnotationBeanPostProcessor =
            new PublisherAnnotationBeanPostProcessor();
        publisherAnnotationBeanPostProcessor.setProxyTargetClass(true);
        return publisherAnnotationBeanPostProcessor;
    }
}

Тест:

@RunWith(SpringRunner.class)
@SpringBootTest
public class ExampleServiceTest {

    @Autowired
    private ExampleService exampleService;

    @Test
    public void testQueue() throws InterruptedException {
        exampleService.queue("Hello!");
        Thread.sleep(1000);//Wait for message processing
        System.out.println("Ready!");
    }
}

У меня следующий вывод:

17:19:10.230 [dest1.group1-2] DEBUG o.s.c.s.b.StreamListenerMessageHandler - org.springframework.cloud.stream.binding.StreamListenerMessageHandler@575c3e9b received message: GenericMessage [payload=Hello!, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=dest1, amqp_receivedExchange=dest1, amqp_deliveryTag=1, amqp_consumerQueue=dest1.group1, amqp_redelivered=false, id=2f22ce16-bb5a-350c-8b3d-e6c898760888, amqp_consumerTag=amq.ctag-sxu6zQHJTGrsazfwbmol9Q, contentType=text/plain, timestamp=1547583550230}]
New message: Hello!
17:19:10.231 [dest1.group1-1] DEBUG o.s.c.s.b.StreamListenerMessageHandler - handler 'org.springframework.cloud.stream.binding.StreamListenerMessageHandler@575c3e9b' produced no reply for request Message: GenericMessage [payload=Hello!, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=dest1, amqp_receivedExchange=dest1, amqp_deliveryTag=1, amqp_consumerQueue=dest1.group1, amqp_redelivered=false, id=788e8bbf-4ae4-86cc-0859-d4f153cb5807, amqp_consumerTag=amq.ctag-fV0aaDzYUZfq08JsODq6pA, contentType=text/plain, timestamp=1547583550230}]
17:19:10.231 [dest1.group1-1] DEBUG o.s.i.channel.DirectChannel - postSend (sent=true) on channel 'input1', message: GenericMessage [payload=Hello!, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=dest1, amqp_receivedExchange=dest1, amqp_deliveryTag=1, amqp_consumerQueue=dest1.group1, amqp_redelivered=false, id=788e8bbf-4ae4-86cc-0859-d4f153cb5807, amqp_consumerTag=amq.ctag-fV0aaDzYUZfq08JsODq6pA, contentType=text/plain, timestamp=1547583550230}]
New message: Hello!
17:19:10.232 [dest1.group1-2] DEBUG o.s.c.s.b.StreamListenerMessageHandler - handler 'org.springframework.cloud.stream.binding.StreamListenerMessageHandler@575c3e9b' produced no reply for request Message: GenericMessage [payload=Hello!, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=dest1, amqp_receivedExchange=dest1, amqp_deliveryTag=1, amqp_consumerQueue=dest1.group1, amqp_redelivered=false, id=2f22ce16-bb5a-350c-8b3d-e6c898760888, amqp_consumerTag=amq.ctag-sxu6zQHJTGrsazfwbmol9Q, contentType=text/plain, timestamp=1547583550230}]
17:19:10.232 [dest1.group1-2] DEBUG o.s.i.channel.DirectChannel - postSend (sent=true) on channel 'input1', message: GenericMessage [payload=Hello!, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=dest1, amqp_receivedExchange=dest1, amqp_deliveryTag=1, amqp_consumerQueue=dest1.group1, amqp_redelivered=false, id=2f22ce16-bb5a-350c-8b3d-e6c898760888, amqp_consumerTag=amq.ctag-sxu6zQHJTGrsazfwbmol9Q, contentType=text/plain, timestamp=1547583550230}]
Ready!

Я не могу понять, в чем проблема с моей конфигурацией или если это какая-то ошибка, какой-либо совет?

Спасибо!

РЕДАКТИРОВАНИЕ:

Я загрузил (не) рабочий пример здесь

Вы можете создать экземпляр RabbitMQ, используя:

docker run -p 5672:5672 -p 15672:15672 rabbitmq:3-management

Answer 1

Я обнаружил, что @Publisher публикует дважды из-за конфигурации в ExampleConfig. Эта новая конфигурация (заимствованная из здесь ), кажется, работает нормально:

@Bean
public static BeanFactoryPostProcessor bfpp() {
    return bf -> bf.getBean(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME,
        PublisherAnnotationBeanPostProcessor.class).setProxyTargetClass(true);
}

Answer 2

Думаю, из конфига вы пытаетесь опубликовать то же сообщение снова в том же месте назначения dest_1.

spring.cloud.stream.bindings.input1.destination=dest_1
spring.cloud.stream.bindings.output1.destination=dest_1

А из журнала понятно, 2-е сообщение имеет другой идентификатор

id=788e8bbf-4ae4-86cc-0859-d4f153cb5807
id=2f22ce16-bb5a-350c-8b3d-e6c898760888