Java RabbitMQ案例代码

本文目录导读：

1. 环境准备
2. 简单队列模式（点对点）
3. 工作队列模式（任务分配）
4. 发布/订阅模式（广播）
5. 路由模式（定向分发）
6. 主题模式（通配符匹配）
7. 使用Spring Boot集成RabbitMQ
8. RabbitMQ配置优化
9. 使用注意事项

我来为您提供几个Java RabbitMQ的完整案例代码，涵盖基本的生产者-消费者模式、工作队列、发布订阅等常见场景。

环境准备

Maven依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.18.0</version>
</dependency>

连接工具类

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitMQConnection {
    private static final String HOST = "localhost";
    private static final int PORT = 5672;
    private static final String USERNAME = "guest";
    private static final String PASSWORD = "guest";
    public static Connection getConnection() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername(USERNAME);
        factory.setPassword(PASSWORD);
        // 自动重连
        factory.setAutomaticRecoveryEnabled(true);
        // 网络异常重连间隔
        factory.setNetworkRecoveryInterval(5000);
        return factory.newConnection();
    }
}

简单队列模式（点对点）

生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class SimpleProducer {
    private static final String QUEUE_NAME = "simple_queue";
    public static void main(String[] args) throws Exception {
        try (Connection connection = RabbitMQConnection.getConnection();
             Channel channel = connection.createChannel()) {
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 发送消息
            String message = "Hello RabbitMQ!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

消费者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
public class SimpleConsumer {
    private static final String QUEUE_NAME = "simple_queue";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages...");
        // 消费消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
    }
}

工作队列模式（任务分配）

生产者（任务发送方）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
public class TaskProducer {
    private static final String QUEUE_NAME = "task_queue";
    public static void main(String[] args) throws Exception {
        try (Connection connection = RabbitMQConnection.getConnection();
             Channel channel = connection.createChannel()) {
            // 持久化队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            for (int i = 0; i < 10; i++) {
                String message = "Task " + i + "...";
                // 持久化消息
                channel.basicPublish("", QUEUE_NAME, 
                    MessageProperties.PERSISTENT_TEXT_PLAIN, 
                    message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
                Thread.sleep(1000);
            }
        }
    }
}

消费者（工作任务处理）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
public class Worker {
    private static final String QUEUE_NAME = "task_queue";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();
        // 持久化队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 预取1条消息，实现公平分发
        channel.basicQos(1);
        System.out.println(" [*] Waiting for tasks. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
            try {
                doWork(message);
            } finally {
                // 手动确认消息
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                System.out.println(" [x] Done");
            }
        };
        // 手动确认模式
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {});
    }
    private static void doWork(String task) {
        // 模拟工作耗时
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

发布/订阅模式（广播）

生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class PubSubProducer {
    public static void main(String[] args) throws Exception {
        try (Connection connection = RabbitMQConnection.getConnection();
             Channel channel = connection.createChannel()) {
            // 声明交换机（fanout类型）
            String EXCHANGE_NAME = "logs";
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            String message = "log message";
            // 发送到交换机
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

消费者1（从队列1接收）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
public class PubSubConsumer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();
        String EXCHANGE_NAME = "logs";
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        // 创建临时队列（随机名称，断开时删除）
        String queueName = channel.queueDeclare().getQueue();
        // 绑定交换机
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println(" [*] Consumer1 Waiting for logs...");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [Consumer1] Received '" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

消费者2（从队列2接收）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
public class PubSubConsumer2 {
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();
        String EXCHANGE_NAME = "logs";
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        // 创建临时队列
        String queueName = channel.queueDeclare().getQueue();
        // 绑定交换机
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println(" [*] Consumer2 Waiting for logs...");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [Consumer2] Received '" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

路由模式（定向分发）

生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class RoutingProducer {
    public static void main(String[] args) throws Exception {
        try (Connection connection = RabbitMQConnection.getConnection();
             Channel channel = connection.createChannel()) {
            String EXCHANGE_NAME = "direct_logs";
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
            String[] severities = {"info", "warning", "error"};
            for (String severity : severities) {
                String message = severity + " message";
                // 发送带路由键的消息
                channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "' with routing key: " + severity);
            }
        }
    }
}

消费者（只接收error级别）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
public class RoutingConsumerError {
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();
        String EXCHANGE_NAME = "direct_logs";
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        String queueName = channel.queueDeclare().getQueue();
        // 只绑定error路由键
        channel.queueBind(queueName, EXCHANGE_NAME, "error");
        System.out.println(" [*] Waiting for ERROR messages...");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [ErrorConsumer] Received '" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

主题模式（通配符匹配）

生产者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class TopicProducer {
    public static void main(String[] args) throws Exception {
        try (Connection connection = RabbitMQConnection.getConnection();
             Channel channel = connection.createChannel()) {
            String EXCHANGE_NAME = "topic_logs";
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
            // 发送不同主题的消息
            String[][] messages = {
                {"kern.error", "kernel error message"},
                {"auth.info", "auth info message"},
                {"user.warning", "user warning message"}
            };
            for (String[] msg : messages) {
                channel.basicPublish(EXCHANGE_NAME, msg[0], null, msg[1].getBytes());
                System.out.println(" [x] Sent '" + msg[1] + "' with topic: " + msg[0]);
            }
        }
    }
}

消费者（匹配所有error和info）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
public class TopicConsumer {
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQConnection.getConnection();
        Channel channel = connection.createChannel();
        String EXCHANGE_NAME = "topic_logs";
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();
        // 绑定多个主题模式
        channel.queueBind(queueName, EXCHANGE_NAME, "*.error");
        channel.queueBind(queueName, EXCHANGE_NAME, "kern.*");
        System.out.println(" [*] Waiting for messages with patterns *.error and kern.*...");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            String routingKey = delivery.getEnvelope().getRoutingKey();
            System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

使用Spring Boot集成RabbitMQ

Maven依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置文件 application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    # 生产者确认
    publisher-confirm-type: correlated
    publisher-returns: true

配置类

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    // 声明队列
    @Bean
    public Queue helloQueue() {
        return new Queue("hello", true);
    }
    // 声明交换机
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("direct.exchange");
    }
    // 声明绑定
    @Bean
    public Binding bindingDirect() {
        return BindingBuilder.bind(helloQueue()).to(directExchange()).with("hello");
    }
}

Spring Boot 生产者

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class MessageProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String message) {
        // 发送到默认交换机，使用队列名作为路由键
        rabbitTemplate.convertAndSend("hello", message);
    }
    public void sendToExchange(String exchange, String routingKey, Object message) {
        rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }
    // 带消息属性的发送
    public void sendWithProperties(String message) {
        MessageProperties properties = new MessageProperties();
        properties.setContentType("application/json");
        properties.setHeader("source", "producer");
        Message msg = new Message(message.getBytes(), properties);
        rabbitTemplate.send("hello", msg);
    }
}

Spring Boot 消费者

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class MessageConsumer {
    @RabbitListener(queues = "hello")
    public void receive(String message) {
        System.out.println("Received message: " + message);
    }
    // 批量消费配置
    @RabbitListener(queues = "hello", containerFactory = "simpleRabbitListenerContainerFactory")
    public void receiveBatch(Message message) {
        System.out.println("Received with properties: " + message.getMessageProperties());
        System.out.println("Message: " + new String(message.getBody()));
    }
}

RabbitMQ配置优化

import com.rabbitmq.client.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQOptimizeConfig {
    @Bean
    public CachingConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setPort(5672);
        // 连接池配置
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        factory.setPublisherReturns(true);
        // 设置缓存
        factory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        factory.setChannelCacheSize(10);
        factory.setConnectionCacheSize(5);
        return factory;
    }
}

使用注意事项

1. 连接管理：确保连接和Channel在使用后关闭
2. 消息持久化：重要消息设置持久化，防止丢失
3. 手动ack：处理消息失败时进行重试或进入死信队列
4. 合理配置prefetch：根据系统性能设置合理的预取数量
5. 监控告警：监控队列堆积情况，及时处理异常
6. 异常处理：添加重试机制和死信队列处理失败消息

这些案例涵盖了RabbitMQ的主要使用场景,您可以根据实际需求进行调整和扩展。