当前位置：首页 > news >正文

rabbitMQ 的安装和使用

news 2025/10/22 8:21:29

1、ubuntu24.04 安装 rabbitMQ

sudo apt update
sudo apt install erlang
sudo apt install rabbitmq-server

安装管理插件：sudo rabbitmq-plugins enable rabbitmq_management

启动服务 sudo systemctl start rabbitmq-server
设置开机自启 sudo systemctl enable rabbitmq-server
检查服务状态 sudo systemctl status rabbitmq-server

新建用户并授权：

# 创建新用户（将 your_username 和 your_password替换为你自己的）
       sudo rabbitmqctl add_user your_username your_password
       # 赋予管理员权限
       sudo rabbitmqctl set_user_tags your_username administrator
       # 授予所有权限
       sudo rabbitmqctl set_permissions -p / your_username ".*" ".*" ".*"

登录管理页面：

浏览器打开 http://ip:15672/ ，使用新建的用户和密码登录

2、使用demo

pom.xml中引入：

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

生产者：确认消息发送成功的几种级别，可靠级别依次升高

1、消息写入到缓冲区，也就是以下代码中的实现(不可靠)；

2、消息到达rabbitMQ服务器(较可靠)

               channel.txSelect();
               try {
                   channel.basicPublish("", "my_queue", null, message.getBytes());
                   channel.txCommit(); // 消息确认到达服务器
                   System.out.println("消息已到达服务器");
               } catch (Exception e) {
                   channel.txRollback();
                   System.out.println("消息发送失败");
               }

3、消息到达rabbitMQ服务器并持久化到磁盘 (高可靠)

           // 持久化队列 + 持久化消息 + 发布者确认
           channel.queueDeclare("my_queue", true, false, false, null);
           channel.confirmSelect();

           channel.basicPublish("", "my_queue",
               MessageProperties.PERSISTENT_TEXT_PLAIN,
               message.getBytes());

           if (channel.waitForConfirms()) {
               System.out.println("消息已持久化到磁盘");
           }

4、到达队列并准备好消费(最可靠)

               // 使用 mandatory 标志和 ReturnListener
               channel.addReturnListener((replyCode, replyText, exchange, routingKey, properties, body) -> {
                   System.out.println("消息无法路由到队列: " + replyText);
               });

               channel.basicPublish("", "my_queue",
                   true, // mandatory = true, 如果无法路由到队列会返回
                   MessageProperties.PERSISTENT_TEXT_PLAIN,
                   message.getBytes());

生产者demo:

public class Producer {public static void main(String[] args) throws Exception {// 1. 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.187.133"); // RabbitMQ服务器地址factory.setPort(5672);        // 默认客户端端口factory.setUsername("rabbit"); // 用户名factory.setPassword("rabbit"); //认密码// 2. 创建连接和信道// 使用try-with-resources确保资源自动关闭try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 3. 声明一个队列。参数：队列名，是否持久化，是否独占，是否自动删除，其他参数channel.queueDeclare("hello", true, false, false, null);// 4. 准备消息String message = "nihao";// 5. 发送消息到默认交换机，路由键为队列名channel.basicPublish("", "hello", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}// 6. 连接和通道会自动关闭}
}

消费者：消费者中需要考虑消费失败、重试、死信队列等场景，收到消息之后如果失败，则将消息的header中添加失败的次数，然后再将消息放入到队列中，再次消费这个消息时判断header中的次数是否超过指定值，如果超过放入到死信队列

消费者demo：

public class Consumer {public static void main(String[] args) throws Exception {// 1. 创建连接工厂（配置同生产者）ConnectionFactory factory = new ConnectionFactory();factory.setHost("192.168.187.133");factory.setUsername("rabbit");factory.setPassword("rabbit");// 2. 建立连接和信道Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.basicQos(10); // 消费者端最多存在未提交数据的条数，默认1条// 3. 声明队列（必须与生产者声明的队列一致）channel.queueDeclare("hello", true, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");// 在消息头中记录重试次数Map<String, Object> headers = new HashMap<>();headers.put("retry-count", 0);// 4. 创建回调函数，定义如何处理消息DeliverCallback deliverCallback = (consumerTag, delivery) -> {try {String message = new String(delivery.getBody(), "UTF-8");// 获取消息头中的重试次数Map<String, Object> headersData = delivery.getProperties().getHeaders();Integer retryCount = (Integer) headersData.getOrDefault("retry-count", 0);if (retryCount >= 3) {// 超过最大重试次数，进入死信队列channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);System.out.println("消息重试超过3次，进入死信队列");return;}// 处理消息System.out.println(" [x] Received '" + message + "'");// 手动确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);} catch (Exception e) {// 增加重试次数并重新入队AMQP.BasicProperties props = delivery.getProperties();Map<String, Object> newHeaders = new HashMap<>(props.getHeaders());Integer currentRetry = (Integer) newHeaders.getOrDefault("retry-count", 0);newHeaders.put("retry-count", currentRetry + 1);AMQP.BasicProperties newProps = new AMQP.BasicProperties.Builder().headers(newHeaders).build();// 重新发布消息（带有更新后的重试次数）channel.basicPublish("", "hello", newProps, delivery.getBody());channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}};}
}