我已经重试设置为true。我的理解是,信息应该不断地不断地传递给消费者。相反,它只是坐在那里不消耗重新排队的消息或任何新消息。我将 com.budjb 和 com.rabbitmq 和 org.springframework.amqp 的日志一直打开到 TRACE 并且没有看到任何断开连接... Heeelppp
应用程序.groovy
rabbitmq {
uri = new URI(System.env.CLOUDAMQP_URL ?: "amqp://test:test@localhost/test")
username = uri.userInfo.split(":")[0]
password = uri.userInfo.split(":")[1]
connections = [
[name : 'main',
host : uri.host,
port : 5672,
username : username,
requestedHeartbeat: 10,
automaticReconnect: true,
virtualHost : uri.path.substring(1), //remove leading slash
password : password]
]
queues = [[name: com.coco.jms.RabbitQueues.INDEX_TRANSACTION.destinationName, autoDelete: false, durable: true, exclusive: false]]
消费者:
class IndexTransactionConsumer implements MessageConsumerEventHandler {
static rabbitConfig = [
connection: 'main',
consumers : 1,
queue : Boolean.valueOf((String) System.getProperty("is_amqp_consumer")) ? RabbitQueues.INDEX_TRANSACTION.destinationName : null,
transacted: true,
autoAck : AutoAck.POST,
retry : true
]
def handleMessage(Map body, MessageContext messageContext) {
log.info("RABBITMQ - *CONSUME* Received event to index transaction (Map). " + body)
throw new Exception("Force fail")
}
....
}
更新 看来,当 transacted=true 和 autoAck = AutoAck.POST 在 AbstractConsumerContext.groovy 内触发的 txRollback() 正在阻止 basicReject nack 到达 RabbitMQ 服务器。
if (configuration.getTransacted()) {
context.getChannel().txRollback()
}
if (configuration.getAutoAck() == AutoAck.POST) {
context.getChannel().basicReject(context.getEnvelope().deliveryTag, configuration.getRetry())
}