RabbitMQ六种通信模式介绍——简单模式（Hello Word！）

落日映苍穹つ 2022-12-24 00:43 142阅读 0赞

### 文章目录 ###

*   *  一。简介
     *  二。代码实现
     *  三。程序运行效果
     *  四。RabbitMQ运转流程
     *   *  1.生产者发送消息
         *  2.消费者接收消息
         *  3.生产者流转过程说明
         *  4.生产者流转过程说明
     *  五。模式总结

--------------------

## 一。简介 ##

本篇博客所讲的为RabbitMQ六种通信模式之一的简单模式，官网给出的图如下所示：

![1555991074575][]

在上图的模型中，有以下概念：

**P**：生产者，也就是要发送消息的程序  
**C**：消费者：消息的接受者，会一直等待消息到来。  
**queue**：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

**在rabbitMQ中消费者是一定要到某个消息队列中去获取消息的。**

## 二。代码实现 ##

1.导入依赖：

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

2.创建连接工具类：

import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    /** * 创建连接工具类 */
    public class ConnectionUtil { 
    
        public static Connection getConnection() throws Exception { 
            //创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //主机地址;默认为 localhost
            connectionFactory.setHost("localhost");
            //连接端口;默认为 5672
            connectionFactory.setPort(5672);
            //虚拟主机名称;默认为 /
            connectionFactory.setVirtualHost("hello");
            //连接用户名；默认为guest
            connectionFactory.setUsername("guest");
            //连接密码；默认为guest
            connectionFactory.setPassword("guest");
    
            //创建连接
            return connectionFactory.newConnection();
        }
    
    }

3.消息生产者：

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    /** * 消息生产者 */
    public class Producer { 
        static final String QUEUE_NAME = "rabbitmq_queue";
    
        public static void main(String[] args) throws Exception { 
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 创建频道
            Channel channel = connection.createChannel();
    
            // 声明（创建）队列
            /** * 参数1：队列名称 * 参数2：是否定义持久化队列（重启后是否能够再次看见未销毁的队列） * 参数3：是否独占本次连接（是否一个channel占用一个队列） * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    
            // 要发送的信息
            String message = "Hello RabbitMQ！";
            /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("已发送消息：" + message);
    
            // 关闭资源
            channel.close();
            connection.close();
        }
    
    }

4.消息消费者：

import com.rabbitmq.client.*;
    import java.io.IOException;
    
    /** * 消息消费者 */
    public class Consumer { 
    
        public static void main(String[] args) throws Exception { 
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 创建频道
            Channel channel = connection.createChannel();
    
            // 声明（创建）队列
            /** * 参数1：队列名称（消费者与提供者传输的队列名称需要一致） * 参数2：是否定义持久化队列（重启后是否能够再次看见未销毁的队列） * 参数3：是否独占本次连接（是否一个channel占用一个队列） * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */
            channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
    
            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){ 
                @Override
                /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("接收到的消息为：" + new String(body, "utf-8"));
                    System.out.println("__________________________________________________________");
                }
            };
            //监听消息
            /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */
            channel.basicConsume(Producer.QUEUE_NAME, true, consumer);
    
            //不关闭资源，应该一直监听消息
            //channel.close();
            //connection.close();
        }
    }

**注意**：需要创建对应名称的Virtual Hosts，不然会出现找不到Virtual Hosts的错误：

## 三。程序运行效果 ##

1.启动消息生产者，程序后台输出：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 5]  
6.一般我们消息的消费者是不做关闭处理的，有消息就立马读消息，在本篇博客消费者的代码中，关闭连接的代码被注释了，所以当消息提供者不断的发送消息，消息消费者就会一直实时的消费消息。例如我们现在启动消息提供者，发送一条“今天天气真好！”的消息：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 7]

## 四。RabbitMQ运转流程 ##

在以上案例中：

### 1.生产者发送消息 ###

1.  生产者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker；
2.  声明队列并设置属性；如是否排它，是否持久化，是否自动删除；
3.  将路由键（空字符串）与队列绑定起来；
4.  发送消息至RabbitMQ Broker；
5.  关闭信道；
6.  关闭连接；

### 2.消费者接收消息 ###

1.  消费者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker
2.  向Broker 请求消费相应队列中的消息，设置相应的回调函数；
3.  等待Broker回应闭关投递响应队列中的消息，消费者接收消息；
4.  确认（ack，自动确认）接收到的消息；
5.  RabbitMQ从队列中删除相应已经被确认的消息；
6.  关闭信道；（这里没有做关闭，所以能一直读取队列中的消息）
7.  关闭连接；（这里没有做关闭，所以能一直读取队列中的消息）

![在这里插入图片描述][20201127112550928.png_pic_center]

### 3.生产者流转过程说明 ###

1.客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。

2.客户端调用connection.createChannel方法。此方法开启信道，其包装的channel.open命令发送给Broker,等待channel.basicPublish方法，对应的AMQP命令为Basic.Publish,这个命令包含了content Header 和content Body()。content Header 包含了消息体的属性，例如:投递模式，优先级等，content Body 包含了消息体本身。

3.客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 8]

### 4.生产者流转过程说明 ###

1.消费者客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。

2.消费者客户端调用connection.createChannel方法。和生产者客户端一样，协议涉及Channel . Open/Open-Ok命令。

3.在真正消费之前，消费者客户端需要向Broker 发送Basic.Consume 命令(即调channel.basicConsume 方法〉将Channel 置为接收模式，之后Broker 回执Basic . Consume - Ok 以告诉消费者客户端准备好消费消息。

4.Broker 向消费者客户端推送(Push) 消息，即Basic.Deliver 命令，这个命令和Basic.Publish 命令一样会携带Content Header 和Content Body。

5.消费者接收到消息并正确消费之后，向Broker 发送确认，即Basic.Ack 命令。

6.客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 9]

## 五。模式总结 ##

**1、简单模式 HelloWorld**

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

**2、工作队列模式 Work Queue**

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

**3、发布订阅模式 Publish/subscribe**

需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

**4、路由模式 Routing**

需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

**5、通配符模式 Topic**

需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

[1555991074575]: /images/20221120/22ab0080e9bb4b54923199cd22515577.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20201125221945851.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201127110503509.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[http_localhost_15672]: http://localhost:15672/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201127110702968.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20201127110823616.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 4]: https://img-blog.csdnimg.cn/20201127110953936.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 5]: https://img-blog.csdnimg.cn/20201127111041207.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 6]: https://img-blog.csdnimg.cn/20201127111447666.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 7]: https://img-blog.csdnimg.cn/20201127111545704.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[20201127112550928.png_pic_center]: https://img-blog.csdnimg.cn/20201127112550928.png#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 8]: https://img-blog.csdnimg.cn/20201127113146243.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw_size_16_color_FFFFFF_t_70_pic_center 9]: https://img-blog.csdnimg.cn/20201127113213552.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwOTQ0Nw==,size_16,color_FFFFFF,t_70#pic_center