如何在Java中清空RabbitMQ Testcontainer的所有队列？

针对你在集成测试中复用RabbitMQ Testcontainer时的清理需求，这里提供两种可行方案，优先推荐Java内调用的方式，更贴合你的技术栈：

一、Java 单次调用实现（推荐） #

利用Spring AMQP配合RabbitMQ的管理API，一次性完成未确认消息回收和全队列清空，步骤是先关闭所有活跃连接（自动触发未确认消息的退回），再清空所有队列：

import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.web.client.RestTemplate;
import java.util.Base64;
import java.util.List;
import java.util.Map;

public void cleanRabbitMQTestcontainer(RabbitTemplate rabbitTemplate) {
    // 获取RabbitMQ连接配置
    var factory = rabbitTemplate.getConnectionFactory();
    String adminUrl = String.format("http://%s:%d/api", factory.getHost(), factory.getPort());
    String auth = Base64.getEncoder().encodeToString(
            (factory.getUsername() + ":" + factory.getPassword()).getBytes()
    );

    // 初始化RestTemplate调用管理API
    RestTemplate restTemplate = new RestTemplate();
    HttpHeaders headers = new HttpHeaders();
    headers.set("Authorization", "Basic " + auth);
    HttpEntity<Void> entity = new HttpEntity<>(headers);

    try {
        // 1. 关闭所有活跃连接：未确认消息会被自动退回原队列
        List<Map<String, Object>> connections = restTemplate.exchange(
                adminUrl + "/connections",
                HttpMethod.GET,
                entity,
                (org.springframework.core.ParameterizedTypeReference<List<Map<String, Object>>>) 
                    new org.springframework.core.ParameterizedTypeReference<>() {}
        ).getBody();

        if (connections != null) {
            for (Map<String, Object> conn : connections) {
                String connName = (String) conn.get("name");
                restTemplate.exchange(
                        adminUrl + "/connections/" + connName,
                        HttpMethod.DELETE,
                        entity,
                        Void.class
                );
            }
        }

        // 2. 清空所有队列：此时所有消息（包括刚退回的）都会被清除
        RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
        for (String queueName : rabbitAdmin.getQueueNames()) {
            rabbitAdmin.purgeQueue(queueName, false);
        }
    } catch (Exception e) {
        throw new RuntimeException("RabbitMQ Testcontainer清理失败", e);
    }
}

说明：

• 关闭连接会断开所有当前的消费者/生产者会话，适合测试间隙执行（测试结束后本就没有活跃业务连接）。
• RabbitMQ Testcontainer默认启用了Management插件，所以HTTP管理接口可直接使用。

二、Bash 脚本实现 #

如果需要通过容器命令执行清理，可以利用RabbitMQ自带的rabbitmqctl工具编写脚本，直接在Testcontainer内执行：

清理脚本内容 #

# 关闭所有连接，触发未确认消息退回
rabbitmqctl list_connections name | grep -v "name" | while read conn_id; do
    rabbitmqctl close_connection "$conn_id" "Test cleanup"
done

# 遍历并清空所有队列
rabbitmqctl list_queues name | grep -v "name" | while read queue_name; do
    rabbitmqctl purge_queue "$queue_name"
done

在Java中调用脚本（通过Testcontainer） #

你可以直接通过Testcontainer的execInContainer方法执行上述命令：

rabbitMQContainer.execInContainer("bash", "-c",
    "rabbitmqctl list_connections name | grep -v 'name' | while read conn_id; do rabbitmqctl close_connection \"$conn_id\" \"Test cleanup\"; done;" +
    "rabbitmqctl list_queues name | grep -v 'name' | while read queue_name; do rabbitmqctl purge_queue \"$queue_name\"; done;"
);

注意：执行前确保测试中的消费者已经停止，避免清理过程中产生新的未确认消息。

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