RabbitMQ的安装和使用

目录

什么是MQ

RabbitMQ概述

单机部署

简单队列模型

代码调试

publisher代码：

Consumer代码：

---

同步调用的优点:

1、时效性较强，可以立即得到结果

同步调用的问题:

1、耦合度高

2、性能和吞吐能力下降

3、有额外的资源消耗

4、有级联失败问题

---

异步通信的优点:

1、耦合度低

2、吞吐量提升

3、故障隔离

4、流量削峰

异步通信的缺点:

1、依赖于Broker的可靠性、安全性、吞吐能力

2、架构复杂了，业务没有明显的流程线，不好追踪管理

---

RabbitMQ中的几个概念:

1、channel:操作MQ的工具

2、exchange:路由消息到队列中

3、queue:缓存消息

4、virtual host:虚拟主机，是对queue、exchange等资源的逻辑分组

---

基本消息队列的消息发送流程:

1、建立connection

2、创建channel

3、利用channel声明队列

4、利用channel向队列发送消息

基本消息队列的消息接收流程:

1、建立connection

2、创建channel

3、利用channel声明队列

4、定义consumer的消费行为handleDelivery()

5、利用channel将消费者与队列绑定

 

什么是MQ

MQ (MessageQueue)，中文是消息队列，字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

RabbitMQ概述

RabbitMQ是基于Erlang语言开发的开源消息通信中间件，官网地址: https://www.rabbitmq.com/

单机部署

方式一：在线拉取

docker pull rabbitmq:3-management

方式二：从本地下载

下载该mq.tar的文件，把它拖到linux的tmp目录下

mq.tarhttps://pan.baidu.com/s/165iUwUhlPvkVxxMhFZTkiw?pwd=6utq

导入mq.tar :

docker load -i mq.tar

执行下面的命令来运行MQ容器：

docker run \
 -e RABBITMQ_DEFAULT_USER=itcast \
 -e RABBITMQ_DEFAULT_PASS=123321 \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

在本机的浏览器中，输入虚拟机的ip和端口15672，即可访问RabbitMQ的官网。

这里username为itcast，password为123321

注册新的用户，username为lisi，密码为123，下面是重复输入一次密码，tags选择admin，最后点击add user

创建新的虚拟主机，name为/itcast，最后点击add virtual host

 此时就出现了新的

创建一个新的虚拟主机

此时发现itcast有两个虚拟主机的权限，而lisi还是一个都没有

 点击lisi

选择itcast，让lisi有itcast的访问权限，点击set permission

此时lisi有itcast的访问权限

点击itcast，修改itcast的权限 

点击取消该权限，再点击确定

 此时这两个用户的业务就隔离了

简单队列模型

官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色:

1、 publisher:消息发布者，将消息发送到队列queue

2、queue:消息队列，负责接受并缓存消息

3、consumer:订阅队列，处理队列中的消息

把该mq.demo代码用idea运行

mq.demohttps://pan.baidu.com/s/1tzmzSUBu7xli33Oooa4oRA?pwd=h472运行publisher服务中的测试类PublisherTest中的测试方法testSendMessage()

记得把这里的17行ip地址改成虚拟机的

启动consumer服务，查看是否能接收消息

这里14行的ip记得改成虚拟机的

代码调试

publisher代码：

当这行代码完成时

Connection connection = factory.newConnection();

可以发现成功连上 

当这行代码完成时

Channel channel = connection.createChannel();

可以看到创建了一个通道 

 当这行代码完成时

channel.queueDeclare(queueName, false, false, false, null);

可以看到创建了一个队列

 当这行代码完成时

hannel.basicPublish("", queueName, null, message.getBytes());

可以看到队列里有一条消息 

按顺序点开这些按钮，可以看到发送的内容

Consumer代码：

当这行代码完成时

Connection connection = factory.newConnection();

可以看到连接出现

 当这行代码完成时

channel.queueDeclare(queueName, false, false, false, null);

我们可以看到，因为在publisher已经创建后，并不会重新创建。 

那为何consumer也要写这行代码呢？

因为有可能consumer会在publisher之前运行，没有这行会报错

当这些代码执行时。注意“等待接受消息”会在“接受到消息”之前打印。因为channel.basicConsume只是绑定消息到queueName，并未执行 handleDelivery。所以要等到消息被执行才会打印出“接收到消息”，但是代码并不会等待消息执行后才运行下去，所以会先打印“等待接收消息”之后打印“接收到消息”。这证明了是异步通信

channel.basicConsume(queueName, true, new DefaultConsumer(channel){
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope,
                               AMQP.BasicProperties properties, byte[] body) throws IOException {
        // 5.处理消息
        String message = new String(body);
        System.out.println("接收到消息：【" + message + "】");
    }
});
System.out.println("等待接收消息。。。。");

当消息被处理后，会被删除

 代码文件点击下载https://pan.baidu.com/s/1w3UhyAjXvUh0EoWlCDr38A?pwd=jtl0 

上一篇：Docker的安装和使用

下一篇：SpringAMQP的配置和使用