RabbitMQ系列（二）深刻了解RabbitMQ工做原理及简单使用

时间  2019-11-12 标签 rabbitmq 系列 深刻 了解 原理 简单 使用

深刻了解RabbitMQ工做原理及简单使用css

RabbitMQ系列文章

  1. RabbitMQ在Ubuntu上的环境搭建
  2. 深刻了解RabbitMQ工做原理及简单使用
  3. RabbitMQ交换器Exchange介绍与实践
  4. RabbitMQ事务和Confirm发送方消息确认——深刻解读
  5. 使用Docker部署RabbitMQ集群
  6. 你不知道的RabbitMQ集群架构全解

RabbitMQ简介

在介绍RabbitMQ以前实现要介绍一下MQ，MQ是什么？html

MQ全称是Message Queue，能够理解为消息队列的意思，简单来讲就是消息以管道的方式进行传递。java

RabbitMQ是一个实现了AMQP（Advanced Message Queuing Protocol）高级消息队列协议的消息队列服务，用Erlang语言的。数据库

使用场景

在咱们秒杀抢购商品的时候，系统会提醒咱们稍等排队中，而不是像几年前同样页面卡死或报错给用户。服务器

像这种排队结算就用到了消息队列机制，放入通道里面一个一个结算处理，而不是某个时间断忽然涌入大批量的查询新增把数据库给搞宕机，因此RabbitMQ本质上起到的做用就是削峰填谷，为业务保驾护航。架构

为何选择RabbitMQ

如今的市面上有不少MQ能够选择，好比ActiveMQ、ZeroMQ、Appche Qpid，那问题来了为何要选择RabbitMQ？并发

  1. 除了Qpid，RabbitMQ是惟一一个实现了AMQP标准的消息服务器；
  2. 可靠性，RabbitMQ的持久化支持，保证了消息的稳定性；
  3. 高并发，RabbitMQ使用了Erlang开发语言，Erlang是为电话交换机开发的语言，天生自带高并发光环，和高可用特性；
  4. 集群部署简单，正是应为Erlang使得RabbitMQ集群部署变的超级简单；
  5. 社区活跃度高，根据网上资料来看，RabbitMQ也是首选；

工做机制

生产者、消费者和代理maven

在了解消息通信以前首先要了解3个概念：生产者、消费者和代理。tcp

生产者：消息的建立者，负责建立和推送数据到消息服务器；ide

消费者：消息的接收方，用于处理数据和确认消息；

代理：就是RabbitMQ自己，用于扮演“快递”的角色，自己不生产消息，只是扮演“快递”的角色。

消息发送原理

首先你必须链接到Rabbit才能发布和消费消息，那怎么链接和发送消息的呢？

你的应用程序和Rabbit Server之间会建立一个TCP链接，一旦TCP打开，并经过了认证，认证就是你试图链接Rabbit以前发送的Rabbit服务器链接信息和用户名和密码，有点像程序链接数据库，使用Java有两种链接认证的方式，后面代码会详细介绍，一旦认证经过你的应用程序和Rabbit就建立了一条AMQP信道（Channel）。

信道是建立在“真实”TCP上的虚拟链接，AMQP命令都是经过信道发送出去的，每一个信道都会有一个惟一的ID，不管是发布消息，订阅队列或者介绍消息都是经过信道完成的。

为何不经过TCP直接发送命令？

对于操做系统来讲建立和销毁TCP会话是很是昂贵的开销，假设高峰期每秒有成千上万条链接，每一个链接都要建立一条TCP会话，这就形成了TCP链接的巨大浪费，并且操做系统每秒能建立的TCP也是有限的，所以很快就会遇到系统瓶颈。

若是咱们每一个请求都使用一条TCP链接，既知足了性能的须要，又能确保每一个链接的私密性，这就是引入信道概念的缘由。

你必须知道的Rabbit

想要真正的了解Rabbit有些名词是你必须知道的。

包括：ConnectionFactory（链接管理器）、Channel（信道）、Exchange（交换器）、Queue（队列）、RoutingKey（路由键）、BindingKey（绑定键）。

ConnectionFactory（链接管理器）：应用程序与Rabbit之间创建链接的管理器，程序代码中使用；

Channel（信道）：消息推送使用的通道；

Exchange（交换器）：用于接受、分配消息；

Queue（队列）：用于存储生产者的消息；

RoutingKey（路由键）：用于把生成者的数据分配到交换器上；

BindingKey（绑定键）：用于把交换器的消息绑定到队列上；

看到上面的解释，最难理解的路由键和绑定键了，那么他们具体怎么发挥做用的，请看下图：

关于更多交换器的信息，咱们在后面再讲。

消息持久化

Rabbit队列和交换器有一个不可告人的秘密，就是默认状况下重启服务器会致使消息丢失，那么怎么保证Rabbit在重启的时候不丢失呢？答案就是消息持久化。

当你把消息发送到Rabbit服务器的时候，你须要选择你是否要进行持久化，但这并不能保证Rabbit能从崩溃中恢复，想要Rabbit消息能恢复必须知足3个条件：

  1. 投递消息的时候durable设置为true，消息持久化，代码：channel.queueDeclare(x, true, false, false, null)，参数2设置为true持久化；
  2. 设置投递模式deliveryMode设置为2（持久），代码：channel.basicPublish(x, x, MessageProperties.PERSISTENT_TEXT_PLAIN,x)，参数3设置为存储纯文本到磁盘；
  3. 消息已经到达持久化交换器上；
  4. 消息已经到达持久化的队列；

持久化工做原理

Rabbit会将你的持久化消息写入磁盘上的持久化日志文件，等消息被消费以后，Rabbit会把这条消息标识为等待垃圾回收。

持久化的缺点

消息持久化的优势显而易见，但缺点也很明显，那就是性能，由于要写入硬盘要比写入内存性能较低不少，从而下降了服务器的吞吐量，尽管使用SSD硬盘可使事情获得缓解，但他仍然吸干了Rabbit的性能，当消息成千上万条要写入磁盘的时候，性能是很低的。

因此使用者要根据本身的状况，选择适合本身的方式。

虚拟主机

每一个Rabbit都能建立不少vhost，咱们称之为虚拟主机，每一个虚拟主机其实都是mini版的RabbitMQ，拥有本身的队列，交换器和绑定，拥有本身的权限机制。

vhost特性

  1. RabbitMQ默认的vhost是“/”开箱即用；

  2. 多个vhost是隔离的，多个vhost没法通信，而且不用担忧命名冲突（队列和交换器和绑定），实现了多层分离；

  3. 建立用户的时候必须指定vhost；

vhost操做

能够经过rabbitmqctl工具命令建立：

rabbitmqctl add_vhost[vhost_name]

删除vhost：

rabbitmqctl delete_vhost[vhost_name]

查看全部的vhost：

rabbitmqctl list_vhosts

环境搭建

前文咱们已经介绍了Ubuntu搭建RabbitMQ的步骤：RabbitMQ在Ubuntu上的环境搭建

若是你是在Windows10上去安装那就更简单了，先放下载地址：

Erlang/Rabbit Server百度网盘连接：https://pan.baidu.com/s/1TnKDV-ZuXLiIgyK8c8f9dg 密码：wct9

固然也可去Erlang和Rabbit官网去下，就是速度比较慢。个人百度云Rabbit最新版本：3.7.6，Erlang版本：20.2，注意：不要下载最新的Erlang，在Windows10上打开扩展插件有问题，打不开。

  1. 安装Erlang；

  2. 安装Rabbit Server；

  3. 进入安装目录\sbin下，使用命令“rabbitmq-plugins enable rabbitmq_management”启动网页管理插件；

  4. 重启Rabbit服务；

使用：http://localhost:15672进行测试，默认的登录帐号为：guest，密码为：guest

重复安装Rabbit Server的坑

若是不是第一次在Windows上安装Rabbit Server必定要把Rabbit和Erlang卸载干净以后，找到注册表：HKEY_LOCAL_MACHINE\SOFTWARE\Ericsson\Erlang\ErlSrv 删除其下的全部项。

否则会出现Rabbit安装以后启动不了的状况，理论上卸载的顺序也是先Rabbit在Erlang。

代码实现

java版实现，使用maven项目，建立能够查看：MyEclipse2017破解设置与maven项目搭建

项目建立成功以后，添加Rabbit Client jar包，只须要在pom.xml里面配置，以下信息：

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

java实现代码分为两个类，第一个是建立Rabbit链接，第二是应用类使用最简单的方式发布和消费消息。

Rabbit的链接，两种方式：

方式一：

public static Connection GetRabbitConnection() {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUsername(Config.UserName);
    factory.setPassword(Config.Password);
    factory.setVirtualHost(Config.VHost);
    factory.setHost(Config.Host);
    factory.setPort(Config.Port);
    Connection conn = null;
    try {
        conn = factory.newConnection();
    } catch (Exception e) {
        e.printStackTrace();
    }
    return conn;
}

方式二：

public static Connection GetRabbitConnection2() {
    ConnectionFactory factory = new ConnectionFactory();
    // 链接格式：amqp://userName:password@hostName:portNumber/virtualHost
    String uri = String.format("amqp://%s:%s@%s:%d%s", Config.UserName, Config.Password, Config.Host, Config.Port,
            Config.VHost);
    Connection conn = null;
    try {
        factory.setUri(uri);
        factory.setVirtualHost(Config.VHost);
        conn = factory.newConnection();
    } catch (Exception e) {
        e.printStackTrace();
    }
    return conn;
}

第二部分：应用类，使用最简单的方式发布和消费消息

public static void main(String[] args) {
    Publisher(); // 推送消息

    Consumer(); // 消费消息
}

/**
 * 推送消息
 */
public static void Publisher() {
    // 建立一个链接
    Connection conn = ConnectionFactoryUtil.GetRabbitConnection();
    if (conn != null) {
        try {
            // 建立通道
            Channel channel = conn.createChannel();
            // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开链接时是否删除队列；参数五：消息其余参数】
            channel.queueDeclare(Config.QueueName, false, false, false, null);
            String content = String.format("当前时间：%s", new Date().getTime());
            // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其余属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
            channel.basicPublish("", Config.QueueName, null, content.getBytes("UTF-8"));
            System.out.println("已发送消息：" + content);
            // 关闭链接
            channel.close();
            conn.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

/**
 * 消费消息
 */
public static void Consumer() {
    // 建立一个链接
    Connection conn = ConnectionFactoryUtil.GetRabbitConnection();
    if (conn != null) {
        try {
            // 建立通道
            Channel channel = conn.createChannel();
            // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开链接时是否删除队列；参数五：消息其余参数】
            channel.queueDeclare(Config.QueueName, false, false, false, null);

            // 建立订阅器，并接受消息
            channel.basicConsume(Config.QueueName, false, "", new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                        byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey(); // 队列名称
                    String contentType = properties.getContentType(); // 内容类型
                    String content = new String(body, "utf-8"); // 消息正文
                    System.out.println("消息正文：" + content);
                    channel.basicAck(envelope.getDeliveryTag(), false); // 手动确认消息【参数说明：参数一：该消息的index；参数二：是否批量应答，true批量确认小于index的消息】
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

代码里面已经写了很详细的注释，在这里也不过多的介绍了。

执行效果，如图：

联系我们 沪ICP备13005482号-10