RabbitMQ 研究

1 、介绍

1.1 RabbitMQ

MQ全称为Message Queue，即消息队列， RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message
Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。RabbitMQ官方地址：http://www.rabbitmq.com/

开发中消息队列通常有如下应用场景：
1、任务异步处理。
将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。提高了应用程序的响应时间。
2、应用程序解耦合
MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合

市场上还有哪些消息队列？
ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ、Redis

为什么使用RabbitMQ呢？
1、使得简单，功能强大。
2、基于AMQP协议。
3、社区活跃，文档完善。
4、高并发性能好，这主要得益于Erlang语言。
5、Spring Boot默认已集成RabbitMQ

1.2 其它相关知识

AMQP是什么 ？

image.png
总结：AMQP是一套公开的消息队列协议，最早在2003年被提出，它旨在从协议层定义消息通信数据的标准格式，为的就是解决MQ市场上协议不统一的问题。RabbitMQ就是遵循AMQP标准协议开发的MQ服务。
官方：http://www.amqp.org/

JMS是什么 ？

image.png

总结：
JMS是java提供的一套消息服务API标准，其目的是为所有的java应用程序提供统一的消息通信的标准，类似java的jdbc，只要遵循jms标准的应用程序之间都可以进行消息通信。它和AMQP有什么 不同，jms是java语言专属的消息服务标准，它是在api层定义标准，并且只能用于java应用；而AMQP是在协议层定义的标准，是跨语言的。

2 快速入门

2.1 RabbitMQ的工作原理

下图是RabbitMQ的基本结构：

image.png

组成部分说明如下：
Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue。
Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
Producer：消息生产者，即生产方客户端，生产方客户端将消息发送到MQ。
Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。

消息发布接收流程：
-----发送消息-----
1、生产者和Broker建立TCP连接。
2、生产者和Broker建立通道。
3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。
4、Exchange将消息转发到指定的Queue（队列）

----接收消息-----
1、消费者和Broker建立TCP连接
2、消费者和Broker建立通道
3、消费者监听指定的Queue（队列）
4、当有消息到达Queue时Broker默认将消息推送给消费者。
5、消费者接收到消息。

2.2 下载安装

2.2.1 下载安装

RabbitMQ由Erlang语言开发，Erlang语言用于并发及分布式系统的开发，在电信领域应用广泛，OTP（Open
Telecom Platform）作为Erlang语言的一部分，包含了很多基于Erlang开发的中间件及工具库，安装RabbitMQ需要安装Erlang/OTP，并保持版本匹配，如下图：

RabbitMQ的下载地址：http://www.rabbitmq.com/download.html

image.png
1）下载erlang
地址如下：
http://erlang.org/download/otp_win64_20.3.exe
erlang安装完成需要配置erlang环境变量： ERLANG_HOME=D:\Program Files\erl9.3 在path中添
加%ERLANG_HOME%\bin;

2）安装RabbitMQ
https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.7.3

2.2.2启动

安装成功后会自动创建RabbitMQ服务并且启动
1）从开始菜单启动RabbitMQ
完成在开始菜单找到RabbitMQ的菜单：

image.png

1）安装并运行服务
rabbitmq-service.bat install 安装服务 rabbitmq-service.bat stop 停止服务 rabbitmq-service.bat start 启动服务

2）安装管理插件
安装rabbitMQ的管理插件，方便在浏览器端管理RabbitMQ

image.png

管理员身份运行 rabbitmq-plugins.bat enable rabbitmq_management

3、启动成功 登录RabbitMQ
进入浏览器，输入：http://localhost:15672
初始账号和密码：guest/guest

image.png

2.3 Hello World

按照官方教程(http://www.rabbitmq.com/getstarted.html)测试hello world:

image.png

2.3.1搭建环境

2）创建maven工程
创建生产者工程和消费者工程，分别加入RabbitMQ java client的依赖。
test-rabbitmq-producer：生产者工程
test-rabbitmq-consumer：消费者工程

 <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>4.0.3</version><!--此版本与spring boot 1.5.9版本匹配-->
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-logging</artifactId>
        </dependency>
    </dependencies>

2.3.2 生产者

在生产者工程下的test中创建测试类如下：

public class Producer01 {

    private static final String QUEUE = "helloworld";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = null;
        Channel channel = null;
        try {

        //创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务,一个mq有多个虚拟机

        //创建与RabbitMQ服务的TCP连接
         connection = connectionFactory.newConnection();
        //创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
         channel = connection.createChannel();
        /**
         * 声明队列，如果Rabbit中没有此队列将自动创建
         * QUEUE：队列名称
         * durable：是否持久化
         * exclusive：队列是否独占此链接
         * autoDelete：队列不再使用时是否自动删除此
         * arguments：队列参数
         */
        channel.queueDeclare(QUEUE, true, false, false, null);

        String message = "helloworld小明"+System.currentTimeMillis();

        /**
         * 消息发布方法
         * exchange: Exchange的名称，如果没有指定，则使用Default Exchange
         * routingKey:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
         * props: 消息包含的属性
         * body: 消息体
         * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显示绑定或解除绑定
         * 默认的交换机，routingKey等于队列名称
         */
        channel.basicPublish("", QUEUE, null, message.getBytes());
        System.out.println("Send Message is:'" + message + "'");

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

            if(channel != null)
            {
                channel.close();
            }
            if(connection != null)
            {
                connection.close();
            }
        }

    }

}

1.2.3 消费者

在消费者工程下的test中创建测试类如下

public class Consumer01 {


    private static final String QUEUE = "helloworld";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = null;
        Channel channel = null;

            //创建连接工程
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connectionFactory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务,一个mq有多个虚拟机

            //创建与RabbitMQ服务的TCP连接
            connection = connectionFactory.newConnection();
            //创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
            channel = connection.createChannel();
            /**
             * 声明队列，如果Rabbit中没有此队列将自动创建
             * QUEUE：队列名称
             * durable：是否持久化
             * exclusive：队列是否独占此链接
             * autoDelete：队列不再使用时是否自动删除此
             * arguments：队列参数
             */
            channel.queueDeclare(QUEUE, true, false, false, null);

            /**
            * 定义消费方法
            */
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
                /**
                 *
                 * @param consumerTag 消费者的标签，在channel.basicConsume()去指定
                 * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
                 * @param properties  properties
                 * @param body
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    //交换机
                    String exchange = envelope.getExchange();
                    //路由key
                    String routingKey = envelope.getRoutingKey();
                    //消息id
                    long deliveryTag = envelope.getDeliveryTag();
                    //消息内容
                    String msg = new String(body,"UTF-8");
                    System.out.println("receive message.." + msg);


                }
            };




        /**
             *监听队列String queue, boolean autoAck, Consumer callback
             * queue 队列名称
             * autoAck 是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动回复
             * callback 消费消息的方法，消费者接收到消息后调用此方法
             */
            channel.basicConsume(QUEUE,true,defaultConsumer);
    }
}

２.3.4 总结

1、发送端操作流程
1）创建连接
2）创建通道
3）声明队列
4）发送消息
2、接收端
1）创建连接
2）创建通道
3）声明队列
4）监听队列
5）接收消息