Java 针对rabbitmq队列消息积压 进行监听

要针对 RabbitMQ 队列中的消息积压进行监听，可以使用 RabbitMQ 的 Java 客户端库，例如 `amqp-client` 库。以下是一个简单的 Java 代码示例，可以使用 `amqp-client` 库对队列中的消息积压进行监听：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Delivery;

import java.io.IOException;

public class MessageListener {

    private static final String QUEUE_NAME = "my_queue";

    public static void main(String[] args) throws IOException {
        // 创建连接工厂并设置连接参数
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");

        // 创建连接
        Connection connection = factory.newConnection();

        // 创建通道
        Channel channel = connection.createChannel();

        // 声明要消费的队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 设置每次最多接收1个消息
        channel.basicQos(1);

        // 创建消息消费者
        channel.basicConsume(QUEUE_NAME, false, (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received message: " + message);
            try {
                Thread.sleep(1000); // 模拟处理消息的时间
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // 手动确认消息已被处理
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, consumerTag -> {});

        // 监听队列中未处理的消息数
        channel.basicConsume(QUEUE_NAME, true, (consumerTag, delivery) -> {
            long messageCount = channel.messageCount(QUEUE_NAME);
            System.out.println("Unprocessed message count: " + messageCount);
        }, consumerTag -> {});

        // 等待消息
        System.out.println("Waiting for messages...");
        while (true) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

在这个例子中，我们首先创建了一个连接工厂并设置了连接参数，然后创建了一个连接和一个通道。接着，我们声明了一个名叫 `my_queue` 的队列，并设置了每次最多接收1个消息。然后，我们创建了一个消息消费者，使用 `basicConsume` 方法来消费队列中的消息。在消费消息的回调函数中，我们模拟了处理消息的时间，然后手动确认消息已被处理。同时，我们也创建了另一个消息消费者，用来监听队列中未处理的消息数，并在控制台输出未处理的消息数。最后，我们使用一个死循环等待消息到达。当有消息到达时，第一个消息消费者的回调函数就会被调用，对消息进行处理，并手动确认消息已被处理。在处理消息的同时，第二个消息消费者也会被调用，监听队列中未处理的消息数，并在控制台输出未处理的消息数。这样，我们就可以实时监控队列中未处理的消息数，以及对消息积压进行处理。