Spring AMQP транзакционный канал тупик - PullRequest
Новая
       15

Spring AMQP транзакционный канал тупик

0 голосов
/ 25 марта 2019

У меня есть требование отправить сообщение экземпляру RabbitMQ с сущностью JPA после его сохранения / очистки, что заставило меня настроить rabbitTemplate как channelTransacted.

Потребитель является внешним, но для создания интеграционного теста я добавил встроенный брокер (Apache QPid) и прослушиватель, чтобы иметь возможность проверять отправку сообщений.

Как указано в документации , я, похоже, зашел в тупик:

Если у нас есть производители и потребители в одном приложении, мы можем оказаться в тупике, когда производители блокируют соединение (потому что у Брокера больше нет ресурсов), и потребители не могут их освободить (потому что соединение заблокировано ). [...]

Отдельная CachingConnectionFactory невозможна для транзакционных производителей, которые выполняются в потоке потребителя, поскольку они должны повторно использовать канал, связанный с транзакциями потребителя.

Если я установлю rabbitTemplate.channelTransacted = false, слушатель будет вызываться нормально, в противном случае harness.getNextInvocationDataFor просто ждет, пока не истечет время ожидания.

Я надеюсь, что такой интеграционный тест еще есть, и, возможно, я что-то настроил неправильно.

Я пробовал как с simple, так и с direct типами слушателей, без разницы: 

queues:
  transactions: 'transactions'

spring:
  rabbitmq:
    host: rabbitmq
    username: guest
    password: guest

    dynamic: true # automatically create queues and exchanges on the RabbitMQ server
    template:
      routing-key: ${queues.transactions}
      retry.enabled: true
      # mandatory: true # interesting only for cases where a reply is expected

    # publisher-confirms: true # does not work in transacted mode
    publisher-returns: true # required to get notifications in case of send problems

    # used for integration tests
    listener:
      type: direct
      direct:
        retry:
          enabled: true
          stateless: false # needed when transacted mode is enabled
          max-attempts: 1 # since this is used just for integration tests, we don't want more

Я использую Spring Boot 2.1.3 с spring-boot-starter-amqp, который использует spring-rabbit-2.1.4 и Apache Qpid 7.1.1 в качестве встроенного брокера для теста: 

@RunWith(SpringRunner.class)
@SpringBootTest(properties = "spring.main.allow-bean-definition-overriding=true")
@AutoConfigureTestDatabase
@Transactional
@ActiveProfiles("test")
public class SalesTransactionGatewayTest {

    private static final String TRANSACTIONS_LISTENER = "transactions";

    @TestConfiguration
    @RabbitListenerTest(spy = false, capture = true)
    public static class Config {
        @Bean
        public SystemLauncher broker() throws Exception {
            SystemLauncher broker = new SystemLauncher();
            Map<String, Object> attributes = new HashMap<>();
            attributes.put(SystemConfig.TYPE, "Memory");
            attributes.put(SystemConfig.INITIAL_CONFIGURATION_LOCATION, "classpath:qpid-config.json");
            attributes.put(SystemConfig.STARTUP_LOGGED_TO_SYSTEM_OUT, false);
            broker.startup(attributes);
            return broker;
        }

        @Bean
        public Listener listener() {
            return new Listener();
        }
    }

    public static class Listener {
        @RabbitListener(id = TRANSACTIONS_LISTENER, queues = "${queues.transactions}")
        public void receive(SalesTransaction transaction) {
            Logger.getLogger(Listener.class.getName()).log(Level.INFO, "Received tx: {0}", transaction);
        }
    }

    @Before
    public void setUp() {
        // this makes the test work, setting it to `true` makes it deadlock
        rabbitTemplate.setChannelTransacted(false);
    }

    @Test
    public void shouldBeSentToGateway() throws Exception {
        SalesTransaction savedTransaction = service.saveTransaction(salesTransaction);

        InvocationData invocationData = this.harness.getNextInvocationDataFor(TRANSACTIONS_LISTENER, 10, TimeUnit.SECONDS);
        assertNotNull(invocationData);
        assertEquals(salesTransaction, invocationData.getArguments()[0]);
    }
}

11:02:56.844 [SimpleAsyncTaskExecutor-1] INFO  org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer - SimpleConsumer [queue=transactions, consumerTag=sgen_1 identity=16ef3497] started
Mar 25, 2019 11:02:57 AM AmqpSalesTransactionGateway send
INFO: Sending transaction: 01a92a56-c93b-4d02-af66-66ef007c2817 w/ status COMPLETED
11:02:57.961 [main] INFO  org.springframework.amqp.rabbit.connection.CachingConnectionFactory - Attempting to connect to: [localhost:5672]
11:02:57.972 [main] INFO  org.springframework.amqp.rabbit.connection.CachingConnectionFactory - Created new connection: rabbitConnectionFactory.publisher#6d543ec2:0/SimpleConnection@79dd79eb [delegate=amqp://guest@127.0.0.1:5672/, localPort= 56501]

java.lang.AssertionError
    at org.junit.Assert.fail(Assert.java:86)
    at org.junit.Assert.assertTrue(Assert.java:41)
    at org.junit.Assert.assertNotNull(Assert.java:712)
    at org.junit.Assert.assertNotNull(Assert.java:722)
...