Java 由于交换不存在，RabbitMQ 客户端在 nack 之后通过生产者提交回调的线程重新发送时挂起
Java RabbitMQ client hangs on resend via thread of producer commit callback after nack due to non-existent exchange
我目前正在试验通过消息代理 RabbitMQ 进行通信时可能发生的故障场景。目标是评估如何使这种沟通更具弹性。
特别是，我想在 producer-commit mode 中发送消息时触发 nack（非确认）确认。
为此，我通过 Spring AMQP 的 RabbitTemplate.send 向不存在的交换器发送消息。在通过 RabbitTemplate.setConfirmCallback 提供的回调中，我然后通过将消息重新发送到现有交换来处理 ack=false 确认（模拟我处理了 nack 原因）。
下面提供了一个示例class和相关的测试，完整的示例项目可以在my github repository中找到。我使用 RabbitMQ 3.6 和 Spring Boot/AMQP 2.0.2.
当 运行 测试时，按预期使用 ack=false 调用回调。
但是，在重新创建频道时重新发送消息会挂起（10 分钟后超时异常）。下面提供了调用堆栈和日志的转储。
问题的解决方案似乎是按照提议 在不同的线程中发送消息。
如果您在测试中取消注释 service.runInSeparateThread = true; 行，一切正常！
但是，除了上面提到的post，我既不真正理解为什么事情（不）有效，也没有在任何地方读到过这种做法。这是预期的行为还是错误？有人可以解释一下细节吗？
非常感谢您的建议！
调用堆栈快照:
"AMQP Connection 127.0.0.1:5672@3968" prio=5 tid=0xe nid=NA waiting
java.lang.Thread.State: WAITING
at java.lang.Object.wait(Object.java:-1)
at com.rabbitmq.utility.BlockingCell.get(BlockingCell.java:73)
at com.rabbitmq.utility.BlockingCell.uninterruptibleGet(BlockingCell.java:120)
at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:36)
at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:494)
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:288)
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:138)
at com.rabbitmq.client.impl.ChannelN.open(ChannelN.java:133)
at com.rabbitmq.client.impl.ChannelManager.createChannel(ChannelManager.java:176)
at com.rabbitmq.client.impl.AMQConnection.createChannel(AMQConnection.java:542)
at org.springframework.amqp.rabbit.connection.SimpleConnection.createChannel(SimpleConnection.java:57)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.createBareChannel(CachingConnectionFactory.java:1156)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.access0(CachingConnectionFactory.java:1144)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.doCreateBareChannel(CachingConnectionFactory.java:585)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createBareChannel(CachingConnectionFactory.java:568)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.getCachedChannelProxy(CachingConnectionFactory.java:538)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.getChannel(CachingConnectionFactory.java:520)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.access00(CachingConnectionFactory.java:94)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.createChannel(CachingConnectionFactory.java:1161)
at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:1803)
at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1771)
at org.springframework.amqp.rabbit.core.RabbitTemplate.send(RabbitTemplate.java:859)
...
日志:
...
10:21:24.613 [main] DEBUG org.springframework.amqp.rabbit.core.RabbitAdmin - declaring Exchange 'ExistentExchange'
10:21:24.630 [main] INFO com.example.rabbitmq.ProducerService - sending `initial Message`
10:21:24.648 [main] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - Added listener org.springframework.amqp.rabbit.core.RabbitTemplate$MockitoMock2329793@562c877a
10:21:24.648 [main] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Added publisher confirm channel: Cached Rabbit Channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1), conn: Proxy@3013909b Shared Rabbit Connection: SimpleConnection@12db3386 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 1341] to map, size now 1
10:21:24.649 [main] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Executing callback RabbitTemplate$$Lambda5/1694519286 on RabbitMQ Channel: Cached Rabbit Channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1), conn: Proxy@3013909b Shared Rabbit Connection: SimpleConnection@12db3386 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 1341]
10:21:24.649 [main] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Publishing message (Body:'[B@67001148(byte[15])' MessageProperties [headers={}, contentType=application/octet-stream, contentLength=0, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])on exchange [nonExistentExchange], routingKey = [nonExistentQueue]
10:21:24.659 [main] INFO com.example.rabbitmq.ProducerService - done with sending message
10:21:24.675 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1) PC:Nack:(close):1
10:21:24.677 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - Sending confirm PendingConfirm [correlationData=null cause=channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'nonExistentExchange' in vhost '/', class-id=60, method-id=40)]
10:21:24.677 [AMQP Connection 127.0.0.1:5672] INFO com.example.rabbitmq.ProducerService - In confirm callback, ack=false, cause=channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'nonExistentExchange' in vhost '/', class-id=60, method-id=40), correlationData=null
10:21:24.677 [AMQP Connection 127.0.0.1:5672] INFO com.example.rabbitmq.ProducerService - sending `resend Message`
10:21:24.678 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1) PC:Nack:(close):1
10:21:24.679 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - AMQChannel(amqp://guest@127.0.0.1:5672/,1) No listener for seq:1
10:21:24.679 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Removed publisher confirm channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1) from map, size now 0
10:21:24.679 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Removed publisher confirm channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1) from map, size now 0
10:21:24.679 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - PendingConfirms cleared
ProducerService:
@Service
public class ProducerService {
static final String EXISTENT_EXCHANGE = "ExistentExchange";
private static final String NON_EXISTENT_EXCHANGE = "nonExistentExchange";
private static final String QUEUE_NAME = "nonExistentQueue";
private final Logger logger = LoggerFactory.getLogger(getClass());
private final RabbitTemplate rabbitTemplate;
private final Executor executor = Executors.newCachedThreadPool();
boolean runInSeparateThread = false;
public ProducerService(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
rabbitTemplate.setConfirmCallback(this::confirmCallback);
}
private void confirmCallback(CorrelationData correlationData, boolean ack, String cause) {
logger.info("In confirm callback, ack={}, cause={}, correlationData={}", ack, cause, correlationData);
if (!ack) {
if (runInSeparateThread) {
executor.execute(() -> sendMessage("resend Message", EXISTENT_EXCHANGE));
} else {
sendMessage("resend Message", EXISTENT_EXCHANGE);
}
} else {
logger.info("sending was acknowledged");
}
}
public void produceMessage() {
sendMessage("initial Message", NON_EXISTENT_EXCHANGE);
}
private void sendMessage(String messageBody, String exchangeName) {
logger.info("sending `{}`", messageBody);
rabbitTemplate.send(exchangeName, QUEUE_NAME, new Message(messageBody.getBytes(), new MessageProperties()));
logger.info("done with sending message");
}
}
ProducerServiceTest:
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = {RabbitAutoConfiguration.class, ProducerService.class})
@DirtiesContext
public class ProducerServiceTest {
@Autowired
private ProducerService service;
@SpyBean
private RabbitTemplate rabbitTemplate;
@Autowired
private AmqpAdmin amqpAdmin;
@Autowired
private CachingConnectionFactory cachingConnectionFactory;
@Before
public void setup() {
cachingConnectionFactory.setPublisherConfirms(true);
amqpAdmin.declareExchange(new DirectExchange(ProducerService.EXISTENT_EXCHANGE));
}
@After
public void cleanup() {
amqpAdmin.deleteExchange(ProducerService.EXISTENT_EXCHANGE);
}
@Test
public void sendMessageToNonexistentExchange() throws InterruptedException {
final CountDownLatch sentMessagesLatch = new CountDownLatch(2);
final List<Message> sentMessages = new ArrayList<>();
doAnswer(invocation -> {
invocation.callRealMethod();
sentMessages.add(invocation.getArgument(2));
sentMessagesLatch.countDown();
return null;
}).when(rabbitTemplate).send(anyString(), anyString(), any(Message.class));
// service.runInSeparateThread = true;
service.produceMessage();
sentMessagesLatch.await();
List<String> messageBodies = sentMessages.stream().map(message -> new String(message.getBody())).collect(toList());
assertThat(messageBodies, equalTo(Arrays.asList("initial Message", "resend Message")));
}
}
我想这可能被认为是一个错误，但它是我们缓存频道以提高性能的方式的产物。问题在于，尝试在为同一通道提供 ack 的同一线程上的通道上发布会导致客户端库出现死锁。
我们有 open issue 来研究解决方案（出于不同的原因）；我们只是还没有抽出时间来。据我所知，自从我们添加了对确认和 returns.
的支持以来，您是 6 年多来第二个点击此按钮的用户
编辑
其实这是另外一种情况；它不会重用通道，因为通道已关闭。它试图创建一个新的渠道，而这正是陷入僵局的地方。我看不出我们 (Spring AMQP) 能做什么；这是 java 客户端的限制；您不能在 ack 线程上执行操作。
我目前正在试验通过消息代理 RabbitMQ 进行通信时可能发生的故障场景。目标是评估如何使这种沟通更具弹性。
特别是，我想在 producer-commit mode 中发送消息时触发 nack（非确认）确认。
为此，我通过 Spring AMQP 的 RabbitTemplate.send 向不存在的交换器发送消息。在通过 RabbitTemplate.setConfirmCallback 提供的回调中，我然后通过将消息重新发送到现有交换来处理 ack=false 确认（模拟我处理了 nack 原因）。
下面提供了一个示例class和相关的测试，完整的示例项目可以在my github repository中找到。我使用 RabbitMQ 3.6 和 Spring Boot/AMQP 2.0.2.
当 运行 测试时，按预期使用 ack=false 调用回调。
但是，在重新创建频道时重新发送消息会挂起（10 分钟后超时异常）。下面提供了调用堆栈和日志的转储。
问题的解决方案似乎是按照提议 service.runInSeparateThread = true; 行，一切正常！
但是，除了上面提到的post，我既不真正理解为什么事情（不）有效，也没有在任何地方读到过这种做法。这是预期的行为还是错误？有人可以解释一下细节吗？
非常感谢您的建议！
调用堆栈快照:
"AMQP Connection 127.0.0.1:5672@3968" prio=5 tid=0xe nid=NA waiting
java.lang.Thread.State: WAITING
at java.lang.Object.wait(Object.java:-1)
at com.rabbitmq.utility.BlockingCell.get(BlockingCell.java:73)
at com.rabbitmq.utility.BlockingCell.uninterruptibleGet(BlockingCell.java:120)
at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:36)
at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:494)
at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:288)
at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:138)
at com.rabbitmq.client.impl.ChannelN.open(ChannelN.java:133)
at com.rabbitmq.client.impl.ChannelManager.createChannel(ChannelManager.java:176)
at com.rabbitmq.client.impl.AMQConnection.createChannel(AMQConnection.java:542)
at org.springframework.amqp.rabbit.connection.SimpleConnection.createChannel(SimpleConnection.java:57)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.createBareChannel(CachingConnectionFactory.java:1156)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.access0(CachingConnectionFactory.java:1144)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.doCreateBareChannel(CachingConnectionFactory.java:585)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createBareChannel(CachingConnectionFactory.java:568)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.getCachedChannelProxy(CachingConnectionFactory.java:538)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.getChannel(CachingConnectionFactory.java:520)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.access00(CachingConnectionFactory.java:94)
at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.createChannel(CachingConnectionFactory.java:1161)
at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:1803)
at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1771)
at org.springframework.amqp.rabbit.core.RabbitTemplate.send(RabbitTemplate.java:859)
...
日志:
...
10:21:24.613 [main] DEBUG org.springframework.amqp.rabbit.core.RabbitAdmin - declaring Exchange 'ExistentExchange'
10:21:24.630 [main] INFO com.example.rabbitmq.ProducerService - sending `initial Message`
10:21:24.648 [main] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - Added listener org.springframework.amqp.rabbit.core.RabbitTemplate$MockitoMock2329793@562c877a
10:21:24.648 [main] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Added publisher confirm channel: Cached Rabbit Channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1), conn: Proxy@3013909b Shared Rabbit Connection: SimpleConnection@12db3386 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 1341] to map, size now 1
10:21:24.649 [main] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Executing callback RabbitTemplate$$Lambda5/1694519286 on RabbitMQ Channel: Cached Rabbit Channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1), conn: Proxy@3013909b Shared Rabbit Connection: SimpleConnection@12db3386 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 1341]
10:21:24.649 [main] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Publishing message (Body:'[B@67001148(byte[15])' MessageProperties [headers={}, contentType=application/octet-stream, contentLength=0, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])on exchange [nonExistentExchange], routingKey = [nonExistentQueue]
10:21:24.659 [main] INFO com.example.rabbitmq.ProducerService - done with sending message
10:21:24.675 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1) PC:Nack:(close):1
10:21:24.677 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - Sending confirm PendingConfirm [correlationData=null cause=channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'nonExistentExchange' in vhost '/', class-id=60, method-id=40)]
10:21:24.677 [AMQP Connection 127.0.0.1:5672] INFO com.example.rabbitmq.ProducerService - In confirm callback, ack=false, cause=channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'nonExistentExchange' in vhost '/', class-id=60, method-id=40), correlationData=null
10:21:24.677 [AMQP Connection 127.0.0.1:5672] INFO com.example.rabbitmq.ProducerService - sending `resend Message`
10:21:24.678 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1) PC:Nack:(close):1
10:21:24.679 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - AMQChannel(amqp://guest@127.0.0.1:5672/,1) No listener for seq:1
10:21:24.679 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Removed publisher confirm channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1) from map, size now 0
10:21:24.679 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.core.RabbitTemplate - Removed publisher confirm channel: PublisherCallbackChannelImpl: AMQChannel(amqp://guest@127.0.0.1:5672/,1) from map, size now 0
10:21:24.679 [AMQP Connection 127.0.0.1:5672] DEBUG org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl - PendingConfirms cleared
ProducerService:
@Service
public class ProducerService {
static final String EXISTENT_EXCHANGE = "ExistentExchange";
private static final String NON_EXISTENT_EXCHANGE = "nonExistentExchange";
private static final String QUEUE_NAME = "nonExistentQueue";
private final Logger logger = LoggerFactory.getLogger(getClass());
private final RabbitTemplate rabbitTemplate;
private final Executor executor = Executors.newCachedThreadPool();
boolean runInSeparateThread = false;
public ProducerService(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
rabbitTemplate.setConfirmCallback(this::confirmCallback);
}
private void confirmCallback(CorrelationData correlationData, boolean ack, String cause) {
logger.info("In confirm callback, ack={}, cause={}, correlationData={}", ack, cause, correlationData);
if (!ack) {
if (runInSeparateThread) {
executor.execute(() -> sendMessage("resend Message", EXISTENT_EXCHANGE));
} else {
sendMessage("resend Message", EXISTENT_EXCHANGE);
}
} else {
logger.info("sending was acknowledged");
}
}
public void produceMessage() {
sendMessage("initial Message", NON_EXISTENT_EXCHANGE);
}
private void sendMessage(String messageBody, String exchangeName) {
logger.info("sending `{}`", messageBody);
rabbitTemplate.send(exchangeName, QUEUE_NAME, new Message(messageBody.getBytes(), new MessageProperties()));
logger.info("done with sending message");
}
}
ProducerServiceTest:
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = {RabbitAutoConfiguration.class, ProducerService.class})
@DirtiesContext
public class ProducerServiceTest {
@Autowired
private ProducerService service;
@SpyBean
private RabbitTemplate rabbitTemplate;
@Autowired
private AmqpAdmin amqpAdmin;
@Autowired
private CachingConnectionFactory cachingConnectionFactory;
@Before
public void setup() {
cachingConnectionFactory.setPublisherConfirms(true);
amqpAdmin.declareExchange(new DirectExchange(ProducerService.EXISTENT_EXCHANGE));
}
@After
public void cleanup() {
amqpAdmin.deleteExchange(ProducerService.EXISTENT_EXCHANGE);
}
@Test
public void sendMessageToNonexistentExchange() throws InterruptedException {
final CountDownLatch sentMessagesLatch = new CountDownLatch(2);
final List<Message> sentMessages = new ArrayList<>();
doAnswer(invocation -> {
invocation.callRealMethod();
sentMessages.add(invocation.getArgument(2));
sentMessagesLatch.countDown();
return null;
}).when(rabbitTemplate).send(anyString(), anyString(), any(Message.class));
// service.runInSeparateThread = true;
service.produceMessage();
sentMessagesLatch.await();
List<String> messageBodies = sentMessages.stream().map(message -> new String(message.getBody())).collect(toList());
assertThat(messageBodies, equalTo(Arrays.asList("initial Message", "resend Message")));
}
}
我想这可能被认为是一个错误，但它是我们缓存频道以提高性能的方式的产物。问题在于，尝试在为同一通道提供 ack 的同一线程上的通道上发布会导致客户端库出现死锁。
我们有 open issue 来研究解决方案（出于不同的原因）；我们只是还没有抽出时间来。据我所知，自从我们添加了对确认和 returns.
的支持以来，您是 6 年多来第二个点击此按钮的用户编辑
其实这是另外一种情况；它不会重用通道，因为通道已关闭。它试图创建一个新的渠道，而这正是陷入僵局的地方。我看不出我们 (Spring AMQP) 能做什么；这是 java 客户端的限制；您不能在 ack 线程上执行操作。