RabbitMQ初相识、

阳光穿透心脏的1/2处 2022-04-11 14:28 206阅读 0赞

什么是消息队列？为什么使用RabbitMQ？其特点有哪些？RabbitMQ 中5种概念模型都有哪些？  
RabbitMQ和 activeMQ？

> 1、下载安装、  
> 　 Rabbit MQ 是建立在强大的Erlang OTP平台上，因此安装RabbitMQ之前要先安装Erlang。  
> erlang：[http://www.erlang.org/download.html][http_www.erlang.org_download.html]  
> rabbitmq：[http://www.rabbitmq.com/download.html][http_www.rabbitmq.com_download.html]

> Erlang环境变量配置完成后，使用erl查看下版本、看下是否配置生效；

![2][]

> rabbitmq\_managemen是管理后台的插件、我们要开启这个插件才能通过浏览器访问登录页面  
> 进入到sbin目录下：rabbitmq-plugins enable rabbitmq\_management  
> 然后开启服务： rabbitmq-server start

![3][]  
默认账号密码均是guest  
![5][]

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

3、RabbitMQ的特性：没什么卵用、  
可靠性（Reliability）  
RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。

（2）灵活的路由（Flexible Routing）  
在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。

（3）消息集群（Clustering）  
多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker 。

（4）高可用（Highly Available Queues）  
队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。

（5）多种协议（Multi-protocol）  
RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。

（6）多语言客户端（Many Clients）  
RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。

（7）管理界面（Management UI）  
RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。

（8）跟踪机制（Tracing）  
如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。

（9）插件机制（Plugin System）  
RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件  
原文：[https://blog.csdn.net/qq\_31634461/article/details/79377256][https_blog.csdn.net_qq_31634461_article_details_79377256]

> ###### 4、RabbitMQ 中概念模型 ######
> 
> 1、使用队列queue 模型、  
> 2、订阅发布模型pub/sub、

![1][]

1、队列queue 模型，是一种一对一的；
    生产者：消费者：如下

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

package com.mq.queue;
    /** * @auther SyntacticSugar * @data 2018/12/6 0006下午 8:09 */
    public class Send { 
        private static final String QUEUE_NAME = "hello.august";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            //创建链接工厂
            ConnectionFactory connf = new ConnectionFactory();
    // connf.setHost(Inet4Address.getLocalHost().getHostAddress());
            connf.setHost("localhost");
            //创建链接、创建channel
            Connection connection = connf.newConnection();
            Channel channel = connection.createChannel();
            //
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //设置发布的资源
            String message="hello world";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            //
            System.out.println("send message :"+message);
            //释放资源
            channel.close();
            connection.close();
        }
    }

package com.mq.queue;
    /** * @auther SyntacticSugar * @data 2018/12/6 0006下午 8:33 */
    public class Receive { 
        //接收端同样声明了一个Queue，这是为什么呢？因为完全会出现一种情况，就是发送端还没有启动，但是接收端已经启动了，这时候要监听消息需要确保Queue已经存在
        private static final String QUEUE_NAME = "hello.august";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //创建消费者
            // 定义回调获取message、重写handleDelivery
            DefaultConsumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    String message = new String(body, "UTF-8");
                    System.out.println("Received :" + message);
                }
            };
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

定义回调获取message、查看下DefaultConsumer(channel)源码；  
![1][1 1]

/** * Constructs a new instance and records its association to the passed-in channel. * @param channel the channel to which this consumer is attached */
        public DefaultConsumer(Channel channel) { 
            _channel = channel;
        }

消息接受者先启动、等待接收消息、然后消息发送者发送消息、；  
![1][1 2]

> 概览  
> ![34][]

2、发布订阅模型、
    是一种一对多的关系、发布者publish message 可以发送到多个sub 接受者、
    生产者：消费者：如下

package com.mq.pub_sub;
    /** * @auther SyntacticSugar * @data 2018/12/6 0006下午 9:07 */
    public class Pub { 
        private static final String EXCHANGE_NAME = "hello.logs";
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setPort(5672);
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            //交换机name、模式是什么：fanout发布订阅模式、是否持久化
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true);
            //定义message、发布
            for (int i = 0; i < 8; i++) { 
                String message="发布"+i+"message";
                channel.basicPublish(EXCHANGE_NAME, "", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes() );
                System.out.println(message);        }
            //释放资源
            channel.close();
            connection.close();
        }
    }

package com.mq.pub_sub;
    /** * @auther SyntacticSugar * @data 2018/12/6 0006下午 9:07 */
    public class Sub { 
        private static final String EXCHANGE_NAME = "hello.logs";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setPort(5672);
            Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel();
            // 产生一个随机队列从交换机获取信息
            final String queue = channel.queueDeclare().getQueue();
            channel.queueBind(queue, EXCHANGE_NAME, "");
            //定义回调获取message
            final DefaultConsumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    String message = new String(body, "UTF-8");
                    System.out.println("receive :" + message);
                }
            };
            // 手动确认
            channel.basicConsume(queue, true, consumer);
        }
    }

![在这里插5入图片描述][5 1]

pub

发布0message
    发布1message
    发布2message
    发布3message
    发布4message
    发布5message
    发布6message
    发布7message

sub

receive :发布0message
    receive :发布1message
    receive :发布2message
    receive :发布3message
    receive :发布4message
    receive :发布5message
    receive :发布6message
    receive :发布7message

> 源码分享：

12.25更新

##### 1、消息确认机制： #####

手动确认签收，用来防止消息的重复消费：  
消费者消费但不确认，会导致消息被消费，但仍停留在队列中不被删除，消费者停止消费以后message会重新回到要被消费状态：  
![1][1 3]

##### 2、work模型： #####

2.1、  
一共发50条消息，consumer1 消费者 1条/s 手动签收，但是ready状态却从50直接到0了，这个消费者直接把所有待消费的message都占为已有了，自己以1条/s 的龟速慢慢消费、  
![1][1 4]

2.2、  
先开启consumer1 、consumer2等待接收，发送50条消息供它们消费，consumer1消费慢，consumer2消费快，但是接收到的数目都是一样的，它们各自抢到的资源是一样多的；  
![3][3 1]

2.3、  
但是要快速消费掉 待消费消息，应当设置

// 设置每个消费者同时只能处理一条消息
    //        channel.basicQos(1);

让消费速度快的 consumer 多消费，慢的少消费，这样可以最短时间内消费完消息；

##### 3、pub/sub订阅模型： #####

>  *  P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）  
>     C：消费者，消息的接受者，会一直等待消息到来。  
>     Queue：消息队列，接收消息、缓存消息。  
>     Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：

*  Fanout：广播，将消息交给所有绑定到交换机的队列 - Direct：定向，把消息交给符合指定routing key 的队列 - Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列  
    Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

###### 3.1、订阅模型-Fanout： ######

Exchange把消息发送给每一个绑定到它的queue;

> 1、在交换机未绑定队列的情况下，生产者向交换机发送消息，message丢失，交换机不能储存message，message失效；  
> 2、绑定队列时候，生产者向交换机发送消息，交换机会发消息储存到队列queue中，等待consumer消费：

![1][1 5]

###### 3.1、订阅模型-Direct： ######

Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息；

> 1/ 队列与交换机的绑定要指定一个RoutingKey（路由key）  
> 2、发送方 往Exchange发送消息时，也必须指定消息的 RoutingKey。

发送方指定Routingkey ，接收方的队列queue绑定Routingkey ，一致就能收到消息，不一致收不到；  
![4][]

###### 3.1、订阅模型-Topic： ######

Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符，通配符 \# 和 \* 。

使用 统配\*.\* 的话，就是匹配多个路径、  
![45][]

[http_www.erlang.org_download.html]: http://www.erlang.org/download.html
[http_www.rabbitmq.com_download.html]: http://www.rabbitmq.com/download.html
[2]: /images/20220411/45f5d1cdeccc4c57af0db74be594bdb4.png
[3]: /images/20220411/86d48677bd6541f6a9cb5494cca5f4a2.png
[5]: /images/20220411/27d2356e919c479cbef7653989bd0afc.png
[https_blog.csdn.net_qq_31634461_article_details_79377256]: https://blog.csdn.net/qq_31634461/article/details/79377256
[1]: /images/20220411/b08a26cef2bc41d0a35cadd24b99a521.png
[1 1]: /images/20220411/44a0b2a01e944eb4be8aa1135429fe5e.png
[1 2]: /images/20220411/fd594814cfc448e8a302184985ea9e58.png
[34]: /images/20220411/af7949964f4f459891b365d36d81cb2e.png
[5 1]: /images/20220411/2b48656a5e8741918b410bb5d705a5f0.png
[1 3]: /images/20220411/0ead6cacecb44413a651d1c74634f9e2.png
[1 4]: /images/20220411/6290e8843bb54472a95c0c6b0ff47f90.png
[3 1]: /images/20220411/fa97a5098e5b4ecfb128f774be93861a.png
[1 5]: /images/20220411/04e3efb53ff34d9abe2c3ae53e7ef5ed.png
[4]: /images/20220411/28b3915377c1419891584fb95cff995a.png
[45]: /images/20220411/5277133f1cf247a2a32c983b257a4659.png