目录:
(1)什么是消息队列
(2)为什么要使用消息队列
(3)RabbitMQ特点
(4)RabbitMQ的安装
(5)RabbitMQ常用命令
(6)RabbitMQ角色介绍
(7)RabbitMQ的消息发送和接收
(8)Java发送和接收Queue的消息
(9)Direct交换机介绍
(10)发送消息到Direct交换机中
(11)消费者从消息队列中接收监听消息
(1)什么是消息队列
消息(Message)是指在应用间传送的数据。消息可以非常简单,比如只包含文本字符串,也可以更复杂,可能包含嵌入对象。
消息队列(Message Queue)是一种应用间的通信方式,消息发送后可以立即返回,由消息系统来确保消息的可靠传递。消息发布者只管把消息发布到 MQ 中而不用管谁来取,消息使用者只管从 MQ 中取消息而不管是谁发布的。这样发布者和使用者都不用知道对方的存在。
(2)为什么要使用消息队列
从上面的描述中可以看出消息队列是一种应用间的异步协作机制,那什么时候需要使用 MQ 呢?
以常见的订单系统为例,用户点击【下单】按钮之后的业务逻辑可能包括:扣减库存、生成相应单据、发红包、发短信通知。在业务发展初期这些逻辑可能放在一起同步执行,随着业务的发展订单量增长,需要提升系统服务的性能,这时可以将一些不需要立即生效的操作拆分出来异步执行,比如发放红包、发短信通知等。这种场景下就可以用 MQ ,在下单的主流程(比如扣减库存、生成相应单据)完成之后发送一条消息到 MQ 让主流程快速完结,而由另外的单独线程拉取MQ的消息(或者由 MQ 推送消息),当发现 MQ 中有发红包或发短信之类的消息时,执行相应的业务逻辑。
以上是用于业务解耦的情况,其它常见场景包括最终一致性、广播、错峰流控等等。
(3)RabbitMQ特点
RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。
AMQP :Advanced Message Queue,高级消息队列协议。它是应用层协议的一个开放标准,为面向消息的中间件设计,基于此协议的客户端与消息中间件可传递消息,并不受产品、开发语言等条件的限制。
RabbitMQ 最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。具体特点包括:
1、可靠性(Reliability)
RabbitMQ 使用一些机制来保证可靠性,如持久化、传输确认、发布确认。
2、灵活的路由(Flexible Routing)
在消息进入队列之前,通过 Exchange 来路由消息的。对于典型的路由功能,RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能,可以将多个 Exchange 绑定在一起,也通过插件机制实现自己的 Exchange 。
3、 消息集群(Clustering)
多个 RabbitMQ 服务器可以组成一个集群,形成一个逻辑 Broker 。
3、 高可用(Highly Available Queues)
队列可以在集群中的机器上进行镜像,使得在部分节点出问题的情况下队列仍然可用。
4、多种协议(Multi-protocol)
RabbitMQ 支持多种消息队列协议,比如 STOMP、MQTT 等等。
5、多语言客户端(Many Clients)
RabbitMQ 几乎支持所有常用语言,比如 Java、.NET、Ruby 等等。
6、 管理界面(Management UI)
RabbitMQ 提供了一个易用的用户界面,使得用户可以监控和管理消息 Broker 的许多方面。
7、 跟踪机制(Tracing)
如果消息异常,RabbitMQ 提供了消息跟踪机制,使用者可以找出发生了什么。
发送消息存在耦合
(4)RabbitMQ的安装
一般来说安装 RabbitMQ 之前要安装 Erlang ,可以去Erlang官网下载。接着去RabbitMQ官网下载安装包,之后解压缩即可。
Erlang官方下载地址:Downloads - Erlang/OTP
RabbitMQ官方下载地址:Downloading and Installing RabbitMQ — RabbitMQ
依赖包安装
安装RabbitMQ之前必须要先安装所需要的依赖包可以使用下面的一次性安装命令
yum install gcc glibc-devel make ncurses-devel openssl-devel xmlto -y
把需要的两个文件上传到Linux上面
解压:
进入解压的命令:
...
安装Erlang:
- 将Erlang源代码包otp_src_19.3.tar.gz上传到Linux的/home目录下
2、解压erlang 源码包
tar -zxvf otp_src_19.3.tar.gz
3、手动创建erlang 的安装目录
mkdir /usr/local/erlang
4、进入erlang的解压目录
cd otp_src_19.3
5、配置erlang的安装信息
./configure --prefix=/usr/local/erlang --without-javac
6、编译并安装
make && make install
7、配置环境变量
vim /etc/profile
8、将这些配置填写到profile文件的最后
ERL_HOME=/usr/local/erlang
PATH=$ERL_HOME/bin:$PATH
export ERL_HOME PATH
9、启动环境变量配置文件
source /etc/profile
安装RabbitMQ:
- 将RabbitMQ安装包rabbitmq-server-3.7.2-1.el7.noarch.rpm上传到/home目录
2、安装RabbitMQ
rpm -ivh --nodeps rabbitmq-server-3.7.2-1.el7.noarch.rpm
(5)RabbitMQ常用命令
启动和关闭
1、启动RabbitMQ
rabbitmq-server start &
注意:这里可能会出现错误,错误原因是/var/lib/rabbitmq/.erlang.cookie文件权限不够。
解决方案对这个文件授权
chown rabbitmq:rabbitmq /var/lib/rabbitmq/.erlang.cookie
chmod 400 /var/lib/rabbitmq/.erlang.cookie
2、停止服务
rabbitmqctl stop
插件管理
1、添加插件
rabbitmq-plugins enable {插件名}
2、删除插件
rabbitmq-plugins disable {插件名}
注意:RabbitMQ启动以后可以使用浏览器进入管控台但是默认情况RabbitMQ不允许直接使用浏览器浏览器进行访问因此必须添加插件
rabbitmq-plugins enable rabbitmq_management
3、使用浏览器访问管控台http://RabbitMQ服务器IP:15672
http://192.168.71.128:15672
按装插件成功之后可以直接通过Linux访问:重新启动,访问端口
也可以通过浏览器访问:
用户管理
RabbitMQ安装成功后使用默认用户名guest登录
账号:guest
密码:guest
注意:这里guest只允许本机登录访问需要创建用户并授权远程访问命令如下
- 添加用户:rabbitmqctl add_user {username} {password}
rabbitmqctl add_user root root
- 删除用户:rabbitmqctl delete_user {username}
- 修改密码:rabbitmqctl change_password {username} {newpassword}
rabbitmqctl change_password root 123456
- 设置用户角色:rabbitmqctl set_user_tags {username} {tag}
rabbitmqctl set_user_tags root administrator
tag参数表示用户角色取值为:management ,monitoring ,policymaker administrator
各角色详解:
management
用户可以通过AMQP做的任何事外加:
列出自己可以通过AMQP登入的virtual hosts
查看自己的virtual hosts中的queues, exchanges 和 bindings
查看和关闭自己的channels 和 connections
查看有关自己的virtual hosts的“全局”的统计信息,包含其他用户在这些virtual hosts中的活动。
policymaker
management可以做的任何事外加:
查看、创建和删除自己的virtual hosts所属的policies和parameters
monitoring
management可以做的任何事外加:
列出所有virtual hosts,包括他们不能登录的virtual hosts
查看其他用户的connections和channels
查看节点级别的数据如clustering和memory使用情况
查看真正的关于所有virtual hosts的全局的统计信息
administrator
policymaker和monitoring可以做的任何事外加:
创建和删除virtual hosts
查看、创建和删除users
查看创建和删除permissions
关闭其他用户的connections
权限管理
- 授权命令:rabbitmqctl set_permissions [-p vhostpath] {user} {conf} {write} {read}
-p vhostpath :用于指定一个资源的命名空间,例如 –p / 表示根路径命名空间
user:用于指定要为哪个用户授权填写用户名
conf:一个正则表达式match哪些配置资源能够被该用户配置。
write:一个正则表达式match哪些配置资源能够被该用户读。
read:一个正则表达式match哪些配置资源能够被该用户访问。
例如:
rabbitmqctl set_permissions -p / root '.*' '.*' '.*'
设置完权限之后就可以远程登录了:使用root root
用于设置root用户拥有对所有资源的 读写配置权限
2、查看用户权限 rabbitmqctl list_permissions [vhostpath]
例如
查看根径经下的所有用户权限
rabbitmqctl list_permissions
查看指定命名空间下的所有用户权限
rabbitmqctl list_permissions /abc
3、查看指定用户下的权限rabbitmqctl list_user_permissions {username}
例如
查看root用户下的权限
rabbitmqctl list_user_permissions root
4、清除用户权限rabbitmqctl clear_permissions {username}
例如:
清除root用户的权限
rabbitmqctl clear_permissions root
vhost管理
vhost是RabbitMQ中的一个命名空间,可以限制消息的存放位置利用这个命名空间可以进行权限的控制有点类似Windows中的文件夹一样,在不同的文件夹中存放不同的文件。
1、添加vhost: rabbitmqctl add vhost {name}
例如
rabbitmqctl add vhost bjpowernode
2、删除vhost:rabbitmqctl delete vhost {name}
例如
rabbitmqctl delete vhost bjpowernode
(6)RabbitMQ角色介绍
RabbitMQ的内部接收如下:
1、Message
消息,消息是不具体的,它由消息头和消息体组成。消息体是不透明的,而消息头则
由一系列的可选属性组成,这些属性包括routing-key(路由键)、priority(相对于其他消息
的优先权)、delivery-mode(指出该消息可能需要持久性存储)等。
2、Publisher
消息的生产者,也是一个向交换器发布消息的客户端应用程序。
3、Exchange
交换器,用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
4、Binding
绑定,用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换
器和消息队列连接起来的路由规则,所以可以将交换器理解成一个由绑定构成的
路由表。
5、Queue
消息队列,用来保存消息直到发送给消费者。它是消息的容器,也是消息的
终点。一个消息可投入一个或多个队列。消息一直在队列里面,等待消费者连接
到这个队列将其取走。
6、Connection
网络连接,比如一个TCP连接。
7、Channel
信道,多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的
TCP连接内地虚拟连接,AMQP 命令都是通过信道发出去的,不管是发布消息、
订阅队列还是接收消息,这些动作都是通过信道完成。因为对于操作系统来说建
立和销毁 TCP 都是非常昂贵的开销,所以引入了信道的概念,以复用一条 TCP
连接。
8、Consumer
消息的消费者,表示一个从消息队列中取得消息的客户端应用程序。
9、Virtual Host
虚拟主机,表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的
身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的
RabbitMQ 服务器,拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP
概念的基础,必须在连接时指定,RabbitMQ 默认的 vhost 是 / 。
10、Broker
表示消息队列服务器实体。
(7)RabbitMQ的消息发送和接收
RabbitMQ的消息发送和接收机制
所有 MQ 产品从模型抽象上来说都是一样的过程:
消费者(consumer)订阅某个队列。生产者(producer)创建消息,然后发布到队列(queue)中,最后将消息发送到监听的消费者。
上面是MQ的基本抽象模型,但是不同的MQ产品有有者不同的机制,RabbitMQ实际基于AMQP协议的一个开源实现,因此RabbitMQ内部也是AMQP的基本概念。
AMQP 中的消息路由
AMQP 中消息的路由过程和 Java 开发者熟悉的 JMS 存在一些差别,AMQP 中增加了 Exchange 和 Binding 的角色。生产者把消息发布到 Exchange 上,消息最终到达队列并被消费者接收,而 Binding 决定交换器的消息应该发送到那个队列
Exchange 交换机类型
Exchange分发消息时根据类型的不同分发策略有区别,目前共四种类型:direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键,此外 headers 交换器和 direct 交换器完全一致,但性能差很多,目前几乎用不到了,所以直接看另外三种类型
1、direct
消息中的路由键(routing key)如果和 Binding 中的 binding key 一致, 交换器就将消息发到对应的队列中。路由键与队列名完全匹配,如果一个队列绑定到交换机要求路由键为“dog”,则只转发 routing key 标记为“dog”的消息,不会转发“dog.puppy”,也不会转发“dog.guard”等等。它是完全匹配、单播的模式。
2、fanout
每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键,只是简单的将队列绑定到交换器上,每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播,每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。
3、topic
topic 交换器通过模式匹配分配消息的路由键属性,将路由键和某个模式进行匹配,此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词,这些单词之间用点隔开。它同样也会识别两个通配符:符号“#”和符号“*”。#匹配0个或多个单词,“*”匹配不多不少一个单词。
(8)Java发送和接收Queue的消息
新建一个Module:创建消息的发送者
创建Module消息的接受者:
配置发送者的配置:
创建发送者的类:SendMessage.java
主启动类:创建这个类:发送消息
需要开放端口:
在发送消息之前需要声明交换机、消息对列、绑定关系
创建一个配置类声明:
启动启动类发送消息:
这里就有了一条queue
绑定的是默认的:
获取到的消息:
(9)Direct交换机介绍
(10)发送消息到Direct交换机中
Java绑定Exchange发送和接收消息
AMQP 协议中的核心思想就是生产者和消费者的解耦,生产者从不直接将消息发送给队列。生产者通常不知道是否一个消息会被发送到队列中,只是将消息发送到一个交换机。先由 Exchange 来接收,然后 Exchange 按照特定的策略转发到 Queue 进行存储。Exchange 就类似于一个交换机,将各个消息分发到相应的队列中。
在实际应用中我们只需要定义好 Exchange 的路由策略,而生产者则不需要关心消息会发送到哪个 Queue 或被哪些 Consumer 消费。在这种模式下生产者只面向 Exchange 发布消息,消费者只面向 Queue 消费消息,Exchange 定义了消息路由到 Queue 的规则,将各个层面的消息传递隔离开,使每一层只需要关心自己面向的下一层,降低了整体的耦合度。
在配置类中创建交换机:
创建队列:
创建绑定:
启动项目:出现自定义交换机
出现队列:这里面目前没有发送消息为0
在SendMessage类中创建发送消息:
在主启动类中调用方法发送消息:
现在directQueue中就有消息了
(11)消费者从消息队列中接收监听消息
首先配置消息消费者模块的配置文件:
创建监听消息的ReceiveMessage类:
运行消息的启动类: