RabbitMQ Java：如何将消费的消息存储到变量中？

你遇到的第一个核心问题是Java方法重写的规则：DefaultConsumer的handleDelivery方法本身就是void返回值，你不能随意修改它的返回类型——这违反了重写的方法签名匹配要求，所以编译器必然会报错。

那怎么把消息存到变量里呢？关键要明白RabbitMQ的消费是异步触发的：当有消息到达时，RabbitMQ客户端会在自己的线程里调用handleDelivery方法，所以我们需要用线程安全的方式来存储消息，避免多线程竞态问题。

下面给你两种常用的实现方式：

方式1：使用线程安全的成员变量 #

可以在外部类中定义一个线程安全的变量（比如AtomicReference），然后在handleDelivery里给它赋值：

import java.util.concurrent.atomic.AtomicReference;
import com.rabbitmq.client.*;

public class RabbitMQReceiver {
    private Channel channelRecv;
    private String queRecv = "your_queue_name";
    // 用AtomicReference保证多线程下的安全赋值
    private AtomicReference<String> receivedMessage = new AtomicReference<>();

    public void setupConnection() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        channelRecv = connection.createChannel();
        // 先声明队列，避免队列不存在的报错
        channelRecv.queueDeclare(queRecv, false, false, false, null);
    }

    public void startConsuming() throws Exception {
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        Consumer consumer = new DefaultConsumer(channelRecv) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
                // 将消息存入线程安全变量
                receivedMessage.set(message);
            }
        };
        channelRecv.basicConsume(queRecv, true, consumer);
    }

    // 对外提供获取消息的方法
    public String getReceivedMessage() {
        return receivedMessage.get();
    }

    public static void main(String[] args) throws Exception {
        RabbitMQReceiver receiver = new RabbitMQReceiver();
        receiver.setupConnection();
        receiver.startConsuming();
        // 这里可以根据业务逻辑等待消息，比如休眠等待
        Thread.sleep(5000);
        System.out.println("Stored message: " + receiver.getReceivedMessage());
    }
}

方式2：用BlockingQueue实现同步等待消息 #

如果你的场景需要等待消息到达后再继续执行（比如测试场景），可以用BlockingQueue来实现同步阻塞：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import com.rabbitmq.client.*;

public class RabbitMQReceiver {
    private Channel channelRecv;
    private String queRecv = "your_queue_name";
    // 用阻塞队列实现消息的同步等待
    private BlockingQueue<String> messageQueue = new ArrayBlockingQueue<>(1);

    public void setupConnection() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        channelRecv = connection.createChannel();
        channelRecv.queueDeclare(queRecv, false, false, false, null);
    }

    public String recv() throws Exception {
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        Consumer consumer = new DefaultConsumer(channelRecv) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
                try {
                    // 将消息放入阻塞队列
                    messageQueue.put(message);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        };
        channelRecv.basicConsume(queRecv, true, consumer);
        // 阻塞等待消息到达，拿到消息后返回
        return messageQueue.take();
    }

    public static void main(String[] args) throws Exception {
        RabbitMQReceiver receiver = new RabbitMQReceiver();
        receiver.setupConnection();
        String message = receiver.recv();
        System.out.println("Received and stored message: " + message);
    }
}

关于你遇到的IOException报错 #

这个报错大概率和消费的初始化有关，你可以优先排查这几个点：

• 确认RabbitMQ服务是否正常运行，连接参数（host、port、用户名密码）是否正确
• 确认队列queRecv是否已经存在，或者在消费前先调用queueDeclare声明队列
• 确认你的客户端账号是否有该队列的消费权限

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