IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Java RabbitMQ案例代码

wen java案例 167

本文目录导读：

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的主要使用场景,您可以根据实际需求进行调整和扩展。

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 足球直播 足球直播 足球直播 篮球直播 篮球直播 篮球直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 亚冠直播 亚冠直播 亚冠直播 欧冠直播 欧冠直播 欧冠直播 英超直播 英超直播 英超直播 西甲直播 西甲直播 西甲直播 中超直播 中超直播 中超直播 足球直播 足球直播 足球直播 篮球直播 篮球直播 篮球直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 亚冠直播 亚冠直播 亚冠直播 欧冠直播 欧冠直播 欧冠直播 英超直播 英超直播 英超直播 西甲直播 西甲直播 西甲直播 中超直播 中超直播 中超直播 足球直播 足球直播 足球直播 篮球直播 篮球直播 篮球直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 亚冠直播 亚冠直播 亚冠直播 欧冠直播 欧冠直播 欧冠直播 英超直播 英超直播 英超直播 西甲直播 西甲直播 西甲直播 中超直播 中超直播 中超直播 足球直播 足球直播 足球直播 篮球直播 篮球直播 篮球直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 亚冠直播 亚冠直播 亚冠直播 欧冠直播 欧冠直播 欧冠直播 英超直播 英超直播 英超直播 西甲直播 西甲直播 西甲直播 中超直播 中超直播 中超直播 曼联直播 曼联直播 曼联直播 日职联直播 日职联直播 日职联直播 澳超直播 澳超直播 澳超直播 阿森纳直播 阿森纳直播 阿森纳直播 足球直播 足球直播 足球直播 NBA直播 NBA直播 NBA直播 曼联直播 曼联直播 曼联直播 日职联直播 日职联直播 日职联直播 澳超直播 澳超直播 澳超直播 阿森纳直播 阿森纳直播 阿森纳直播 足球直播 足球直播 足球直播 NBA直播 NBA直播 NBA直播 曼联直播 曼联直播 曼联直播 日职联直播 日职联直播 日职联直播 澳超直播 澳超直播 澳超直播 阿森纳直播 阿森纳直播 阿森纳直播 足球直播 足球直播 足球直播 NBA直播 NBA直播 NBA直播 曼联直播 曼联直播 曼联直播 日职联直播 日职联直播 日职联直播 澳超直播 澳超直播 澳超直播 阿森纳直播 阿森纳直播 阿森纳直播 足球直播 足球直播 足球直播 NBA直播 NBA直播 NBA直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 篮球直播 篮球直播 篮球直播 篮球直播 篮球直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 中超直播 中超直播 中超直播 中超直播 中超直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 篮球直播 篮球直播 篮球直播 篮球直播 篮球直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 中超直播 中超直播 中超直播 中超直播 中超直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播

上一篇Java实现延迟队列案例

下一篇Java Kafka生产者消费者案例

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝