RabbitMQ 简单队列

超、凢脫俗 2024-03-25 20:16 83阅读 0赞

##### 一、简单队列 #####

官方内容参考：[http://www.rabbitmq.com/tutorials/tutorial-one-java.html][http_www.rabbitmq.com_tutorials_tutorial-one-java.html]

即一个生产者对应一个消费者，一对一的关系，不允许多个消费者消费。

![fd1f6f1d1caf4a5d9d97df49381895e8.png][]

##### 二、代码演示 #####

连接RabbitMQ工具类

虚拟机【virtualHost】概念是RabbitMQ的核心,在用户未自定义虚拟机前已经内置有虚拟机,在使用RabbitMQ中,可以进行自定义配置虚拟机.一个虚拟机中可以含有多个队列信息

虚拟机最大的好处在于可以根据不同的用户分配不同的操作空间

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

如果不希望日志报红，可以引入如下jar包

<dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.26</version>
    </dependency>
    
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.26</version>
    </dependency>

package com.rabbitmq.util;
    
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * RabbitMQ连接工具类
     */
    public class ConnectionUtil {
    
        private static final String host = "192.168.70.139";
        private static final int port = 5672;
        private static final String userName = "admin";
        private static final String password = "admin";
        private static final String virtualHost = "testhost";
    
        /**
         * 获取RabbitMQ Connection连接
         * @return
         * @throws IOException
         * @throws TimeoutException
         */
        public static Connection getConnection() throws IOException, TimeoutException {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(host);
            connectionFactory.setPort(5672);
    
            connectionFactory.setUsername(userName);
            connectionFactory.setPassword(password);
            connectionFactory.setVirtualHost(virtualHost);
    
            return connectionFactory.newConnection();
        }
    }

如上所示，如果配置有用户名密码以及vhost，则配置即可。

###### 生产者 ######

package com.rabbitmq.simple;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 生产者
     */
    public class Send {
    
        private static final String QUEUE_NAME = "test_simple_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtil.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 申明一个队列，没有就会创建
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            /*
             Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;
    
             queue: 队列名称
             durable: 是否持久化，true表示RabbitMQ重启后，队列仍然存在
             exclusive: true表示当前连接的专用队列，在连接断开后，会自动删除该队列
             autoDelete: true 表示当没有任何消费者使用时，自动删除该队列
             arguments: 该队列其他配置参数
             */
    
            String message = "Hello RabbitMQ !";
            // 发送消息
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            /*
            void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;
            exchange: 指定交换机，这里为简单队列，不需要使用，默认""即可，不能为null
            routingKey: 路由key，这里为队列名称
            props: 其他消息属性，路由头信息等等
            body: 消息byte内容
             */
            System.out.println(" [x] Sent '" + message + "'");
    
            // 关闭通道和连接
            channel.close();
            connection.close();
        }
    
    }

API说明：

Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;
    
    queue: 队列名称
    durable: 是否持久化，true表示RabbitMQ重启后，队列仍然存在
    exclusive: true表示当前连接的专用队列，在连接断开后，会自动删除该队列
    autoDelete: true 表示当没有任何消费者使用时，自动删除该队列
    arguments: 该队列其他配置参数

void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;
    exchange: 指定交换机，这里为简单队列，不需要使用，默认""即可，不能为null
    routingKey: 路由key，这里为队列名称
    props: 其他消息属性，路由头信息等等
    body: 消息byte内容

###### 消费者 ######

package com.rabbitmq.simple;
    
    import com.rabbitmq.client.*;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 消费者
     */
    public class Recv {
    
        private static final String QUEUE_NAME = "test_simple_queue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
    
            // 获取连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 打开通道
            Channel channel = connection.createChannel();
    
            // 申明要消费的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 创建一个回调的消费者处理类
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    // 接收到的消息
                    String message = new String(body);
                    System.out.println(" [x] Received '" + message + "'");
                }
            };
    
            // 消费消息
            channel.basicConsume(QUEUE_NAME, true, consumer);
    
        }
    
    }

API说明：

String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;
    queue: 队列名称
    autoAck: 是否自动应答，即生产者发送消息即认为该消息被消费
    callback: 回调处理类，即消息被消费时进行回调处理

##### 三、测试 #####

执行生产者main方法。

[x] Sent 'Hello RabbitMQ !'

执行消费者main方法

[x] Received 'Hello RabbitMQ !'

表示生产者消费者都没有问题。

额外测试：

可以先启动两个消费者，然后再执行一个生产者，会发现只能其中一个消费者处理，这就是简单队列模型。

[http_www.rabbitmq.com_tutorials_tutorial-one-java.html]: http://www.rabbitmq.com/tutorials/tutorial-one-java.html
[fd1f6f1d1caf4a5d9d97df49381895e8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/a205b0c701e646c784af827066dcf42f.png