RabbitMQ概述和运行机制

雨点打透心脏的1/2处 2022-04-04 06:50 204阅读 0赞

# RabbitMQ概述和运行机制 #

# RabbitMQ概述 #

**RabbitMQ概述**：RabbitMQ是使用最广泛的开源消息代理。RabbitMQ轻量级，易于在集群内部和云平台中部署。它支持多种消息传递协议。 它可以满足企业高规模，高可用性的要求。RabbitMQ使用Erlang语言开发的。  
**MQ概述**：全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。  
**MQ运行机制**： P表示生产者，C表示消费者，红色部分为消息队列

![89a7e7ca27fbdef99cd1ff4541998f26.png][]

## MQ实战场景： ##

1.我们在双11的时候，当我们凌晨大量的秒杀和抢购商品，然后去结算的时候，就会发现，界面会提醒我们，让我们稍等，以及一些友好的图片文字提醒。而不是像前几年的时代，动不动就页面卡死，报错等来呈现给用户。在这个业务场景中，我们就可以采用队列的机制来处理，因为同时结算就只能达到这么多。

![09c0bade4a049cd813c9c18a3e16f57a.png][]

![38edc1cb8949d2b782829981299fa1a0.png][]

2.在我们平时的超市中购物也是一样，当我们在结算的时候，并不会一窝蜂一样涌入收银台，而是排队结算。这也是队列机制。一个接着一个的处理，不能插队。

![eacdbc0be34dc7503f3aa20feac429bd.png][]

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

**RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。**

RabbitMQ是AMQP服务器的一种。  
**AMQP简介**：AMQP，即Advanced Message Queuing Protocol，**高级消息队列协议**，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。 AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。  
它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，**并不受产品、开发语言等条件的限制。**

AMQP 里主要要说两个组件：Exchange 和 Queue （在 AMQP 1.0 里还会有变动），如下图所示，绿色的 X 就是 Exchange ，红色的是 Queue ，这两者都在 Server 端，又称作 Broker ，这部分是 RabbitMQ 实现的，而蓝色的则是客户端，通常有 Producer（生产者） 和 Consumer（消费者） 两种类型：  
![b434078b89eb5c1cadbbc0a0f363adef.png][]  
Publisher  
消息的生产者，也是一个向交换器发布消息的客户端应用程序。  
Exchange  
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。  
Queue  
消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。  
Consumer  
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。  
Broker  
表示消息队列服务器实体。

RabbitMQ 最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。具体特点包括：

*  可靠性（Reliability）  
    RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。
 *  灵活的路由（Flexible Routing）  
    在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。
 *  消息集群（Clustering）  
    多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker 。
 *  高可用（Highly Available Queues）  
    队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。
 *  多种协议（Multi-protocol）  
    RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。
 *  多语言客户端（Many Clients）  
    RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。
 *  管理界面（Management UI）  
    RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。
 *  跟踪机制（Tracing）  
    如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。
 *  插件机制（Plugin System）  
    RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。

# RabbitMQ 安装 #

RabbitMQ是用erlang语言编写的，所以我们先安装erlang语言环境  
配置erlang语言环境

vim /etc/yum.repos.d/rabbitmq-erlang.repo
    # 在rabbitmq-erlang.repo 文件中加入下面的代码
    [rabbitmq-erlang]
    name=rabbitmq-erlang
    baseurl=https://dl.bintray.com/rabbitmq/rpm/erlang/20/el/7
    gpgcheck=1
    gpgkey=https://dl.bintray.com/rabbitmq/Keys/rabbitmq-release-signing-key.asc
    enabled=1
    # 执行导入key
    rpm --import  https://dl.bintray.com/rabbitmq/Keys/rabbitmq-release-signing-key.asc
    # 安装erlang
    yum install erlang -y  #安装erlang

**安装rabbitmq服务**

下载rabbitmq 地址：[http://www.rabbitmq.com/download.html][http_www.rabbitmq.com_download.html]

wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.8/rabbitmq-server-3.7.8-1.el7.noarch.rpm
    
    yum install  rabbitmq-server-3.7.8-1.el7.noarch.rpm

启用RabbitMQ的web插件 ，方便后期管理界面

rabbitmq-plugins enable rabbitmq_management

The following plugins have been configured:
      rabbitmq_management
      rabbitmq_management_agent
      rabbitmq_web_dispatch
    Applying plugin configuration to rabbit@localhost...
    The following plugins have been enabled:
      rabbitmq_management
      rabbitmq_management_agent
      rabbitmq_web_dispatch
    
    set 3 plugins.
    Offline change; changes will take effect at broker restart.

设置开机启动

systemctl enable rabbitmq-server.service

启动服务

systemctl start rabbitmq-server

rabbitmq配置文件位置

ls /var/lib/rabbitmq/mnesia

访问控制台  
默认用户名和密码： guest/guest 。guest用户仅允许从在服务器以localhost或127.0.0.1作为ip登录

如果远程登录，如：[http://192.168.1.63:15672/][http_192.168.1.63_15672], 则会提示错误，登录不了。

为RabbitMQ创建用户并赋权。

rabbitmqctl add_user root 123456  #添加用户
    rabbitmqctl set_user_tags root administrator #设置用户权限为administrator

到此，已经搭建成功。

# RabbitMQ使用方法 #

**RabbitMQ查看相关的命令**

rabbitmqctl list_connections   #用于查看当前的连接
    rabbitmqctl list_queues   #会列出所有队列名称，后边可能还会带着这个队列当前消息数
     rabbitmqctl status        #查看当前队列信息

## RabbitMQ的vhost管理 ##

当我们在创建用户时，会指定用户能访问一个虚拟机，并且该用户只能访问该虚拟机下的队列和交换机，如果没有指定，默认的是”/”;一个rabbitmq服务器上可以运行多个vhost，以便于适用不同的业务需要，这样做既可以满足权限配置的要求，也可以避免不同业务之间队列、交换机的命名冲突问题，因为不同vhost之间是隔离的。

添加yang-web和yang-bbs两个虚拟机来管理网站和论坛的队列

rabbitmqctl add_vhost yang-web
    rabbitmqctl add_vhost yang-bbs

查看创建的虚拟主机 网页查看

删除bbs虚拟机

rabbitmqctl delete_vhost yang-bbs

查看虚拟机列表

rabbitmqctl list_vhosts

**“/”是rabbitmq默认的虚拟机，之前默认连接的都是它**

# RabbitMQ管理用户、角色和权限管理 #

1、用户管理语法  
添加用户：rabbitmqctl add\_user \{username\} \{password\}  
删除用户：rabbitmqctl delete\_user \{username\}  
修改密码：rabbitmqctl change\_password \{username\} \{newpassword\}

2、角色权限分配  
设置用户角色语法：rabbitmqctl set\_user\_tags \{username\} \{tag\}  
RabbitMQ的tag用户角色分类：none、management、policymaker、monitoring、administrator  
tag常用角色为：administrator, monitoring, management

### RabbitMQ各类角色描述： ###

（1）、none角色权限 ：不能访问 management plugin  
（2）、management角色权限：  
列出自己可以通过AMQP登入的virtual hosts  
查看自己的virtual hosts中的queues, exchanges 和 bindings  
查看和关闭自己的channels 和 connections  
查看有关自己的virtual hosts的“全局”的统计信息，包含其他用户在这些virtual hosts中的活动。

（3）、policymaker角色权限 \#policymaker \['pɒləsɪmeɪkə(r)\] 决策者  
拥有management的所有权限，还拥有查看、创建和删除自己的virtual hosts所属的policies（策略）和parameters（\[pəˈræmɪtə(r)\] 参数 ）  
（4）、monitoring 角色权限  
拥有management的所有权限，还拥有：  
列出所有virtual hosts，包括他们不能登录的virtual hosts  
查看其他用户的connections和channels  
查看节点级别的数据如clustering和memory使用情况  
查看真正的关于所有virtual hosts的全局的统计信息  
（5）、administrator角色权限  
拥有policymaker和monitoring的所有权限，还拥有：  
创建和删除virtual hosts  
查看、创建和删除users  
查看创建和删除permissions  
关闭其他用户的connections

TODO

# 使用python调用rabbitmq服务器 #

pip install pika \#安装pika模块。python用pika模块调用rabbitmq。  
**注：** rabbitmq本质是一个生产者和消费者的模型结构。生产者->rabbitmq->消费者，即生产者产生消息，给到rabbitmq存储，消费者从rabbitmq中读取数据。

[创建生产者代码send.py][send.py]

import pika
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='hello')
    
    channel.basic_publish(exchange='',
                          routing_key='hello',
                          body='Hello World!')
    print(" [x] Sent 'Hello World!'")
    connection.close()
    
    注：declare  [dɪˈkleə(r)]  声明  ； consuming [kənˈsju:mɪŋ]  消费；publish  [ˈpʌblɪʃ]  颁布

[创建消耗者代码receive.py][receive.py]

import pika
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))
    channel = connection.channel()
    
    channel.queue_declare(queue='hello')
    
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
    
    channel.basic_consume(callback,
                          queue='hello',
                          no_ack=True)
    
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

## 开始测试队列 ##

python [send.py][send.py 1] \#此命令执行两次，产生两个名字叫hello的消息  
rabbitmqctl list\_queues \#查看消息队列为2

在web界面查看消息队列，发现在2个消息队列等待处理：  
python [receive.py][receive.py 1] \#消费或处理这2个消息  
rabbitmqctl list\_queues \#查看队列，已经为0  
在web界面，查看队列，也为0了

[89a7e7ca27fbdef99cd1ff4541998f26.png]: /images/20220404/74440e1c08944f67867055e6377f0bf4.png
[09c0bade4a049cd813c9c18a3e16f57a.png]: /images/20220404/b3fffcace72747159ece9565532365fa.png
[38edc1cb8949d2b782829981299fa1a0.png]: /images/20220404/21ddcdb9b1ff411ebc6dfb94148bc9c3.png
[eacdbc0be34dc7503f3aa20feac429bd.png]: /images/20220404/be56931cd57f444c8659cb994532ecbd.png
[b434078b89eb5c1cadbbc0a0f363adef.png]: /images/20220404/810a4cf38d894223b8b0876f9c8ea014.png
[http_www.rabbitmq.com_download.html]: http://www.rabbitmq.com/download.html
[http_192.168.1.63_15672]: http://192.168.1.63:15672/
[send.py]: http://xn--send-fx5fmmg6qf23a0f8b69f4xx.py/
[receive.py]: http://xn--receive-0b5kg3vpx8abq7aij0ae1vgha.py/
[send.py 1]: http://send.py/
[receive.py 1]: http://receive.py/