如何在AMQP-Backed Channel中实现手动ACK？

问题原因 #

AMQP-Backed Channel 与 AMQP Inbound Channel Adapter 的设计定位不同，后者作为入站端点默认会将原生AMQP Channel、投递标签等元信息映射到消息头中，而前者默认的头映射规则不会暴露这类底层对象，所以你直接从消息头里拿不到AmqpHeaders.CHANNEL。

解决步骤 #

1. 修改AmqpChannelFactoryBean配置

你需要自定义AmqpHeaderMapper，明确指定需要映射AmqpHeaders.CHANNEL和AmqpHeaders.DELIVERY_TAG两个头，配置示例如下：

@Bean(name = "AMQP_BACKED_CHANNEL")
public AmqpChannelFactoryBean pubSub(ConnectionFactory connectionFactory) {
    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean();
    factoryBean.setConnectionFactory(connectionFactory);
    factoryBean.setQueueName(AMQP_BACKED_CHANNEL);
    factoryBean.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    factoryBean.setPubSub(false);
    factoryBean.setExtractPayload(true);

    // 新增自定义头映射配置
    DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
    headerMapper.setInboundHeaderNames(Arrays.asList(
        AmqpHeaders.CHANNEL,
        AmqpHeaders.DELIVERY_TAG,
        // 你需要的其他AMQP头都可以在这里加
        AmqpHeaders.CONTENT_TYPE,
        AmqpHeaders.REPLY_TO
    ));
    factoryBean.setHeaderMapper(headerMapper);

    return factoryBean;
}

2. 手动处理ACK

配置完成后，你就可以从消息头中获取所需属性完成手动确认，示例代码如下：

import com.rabbitmq.client.Channel;
import org.springframework.amqp.support.AmqpHeaders;

// 从消息头提取属性
Channel amqpChannel = message.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
Long deliveryTag = message.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class);

// 确认消息
if (amqpChannel != null && deliveryTag != null) {
    try {
        // 第二个参数为true时会批量确认当前通道所有小于等于deliveryTag的未确认消息
        amqpChannel.basicAck(deliveryTag, false);
    } catch (IOException e) {
        // 处理确认异常
    }
}

// 若处理失败需要拒收消息可以用以下代码
// amqpChannel.basicNack(deliveryTag, false, true); 第三个参数为true时消息会重新入队

注意事项 #

• AMQP Channel本身是非线程安全的，不要在多线程场景下并发调用同一个Channel的确认方法，避免通道被意外关闭。
• 如果你需要操作完整的原始AMQP消息，可以将setExtractPayload设置为false，此时消息的payload就是原生的org.springframework.amqp.core.Message对象，你可以直接从该对象中获取所有元信息。

内容的提问来源于stack exchange，提问作者DColella