5_RabbitMQ入门

wbingyang

2020-05-26

关注关注

消息中间件概述

什么是消息中间件
- MQ全称为Message Queue: 消息队列是应用程序之间的通信.
为什么要使用MQ
- 在项目中, 可将一些无需即时返回且耗时的操作提取出来, 进行异步处理, 而这种异步处理的方式能大大节省服务器的请求响应时间, 从而提高了系统的吞吐量.
开发中消息队列的应用场景
- 任务异步处理: 将不需要同步处理并且耗时长的操作由消息队列通知消息接收方进行异步处理, 加快了应用程序的响应时间.
- 应用程序解耦合: MQ相当于一个中介, 生产方通过MQ与消费方交互, 它将应用程序解耦合.
- 削峰填谷
  - 比如订单系统, 高峰期的时候写入数据库的量很大, 可能超出数据库支撑的并发写入, 容易宕机, 而低峰期远低于其并发写入上限.
  - 使用消息队列后, 消息被MQ保存起来, 然后系统可以按照自己的消费能力来消费. 比如每秒1000个数据, 这样慢慢写入, 就不会卡死数据库了.
  - 但是用了MQ后, 限制了消费消息的速度, 这样高峰期产生的数据会被积压在MQ中, 高峰就被削了, 但因为消息挤压, 在高峰期过去后, 消费消息速度会维持不变, 直到消费完挤压的消息, 这就叫填谷.
MQ是消息通信的模型, 实现MQ有两大方式: AMQP, JMS.
- AMQP
  - 是一种链接协议, 它不从API层面进行限定, 而直接定义网络交换的数据格式.
- JMS
  - 即Java消息服务应用程序接口, 是一个java平台中关于面向消息中间件的API, 用在两个应用之间, 或分布式系统中发送消息, 进行异步通信.
- 区别
  - JMS定义了统一的接口, 来对消息操作统一, 而AMQP通过规定协议来统一数据交互格式.
  - JMS必须使用Java, 而AMQP跨语言的.
  - JMS只有两种消息模式, 而AMQP有多种.
消息队列的产品
- ActiveMQ: 基于JMS
- ZeroMQ: 基于c语言
- RabbitMQ: 基于AMQP, erlang语言开发, 稳定性好.
- RocketMQ: JMS, 阿里产品
- Kafka: 类似MQ产品, 分布式消息系统, 高吞吐量, 大数据中多用.

RabbitMQ简介与安装

RabbitMQ由erlang语言开发, 基于AMQP(高级消息队列协议)实现的队列, 是一种应用程序之间的通信方式.
6种模式: 简单模式, work模式, Publish/Subscribe发布与订阅模式, Routing路由模式, Topics主题模式, RPC远程调用模式.
安装
- 安装Socat
  - yum install gcc
  - yum install socat
- 安装erlang
  - mkdir /opt/rabbitmq && cd /opt/rabbitmq
  - 下载好erlang的rpm文件后, 用 rpm -ivh 命令安装
- 安装RabbitMQ
  - rpm -ivh命令安装
- 安装完成后用 systemctl status rabbitmq-server 命令检验是否安装成功.
开启管理界面及配置
- 开启管理界面: rabbitmq-plugins enable rabbitmq_management
- 配置远程可使用guest登录mq
  - cd /usr/share/doc/rabbit-server-3.7.19
  - cp rabbitmq.config.example /etc/rabbitmq/rabbitmq.config
- 修改配置文件　　
  - vi /etc/rabbitmq/rabbitmq.config
- 启动: systemctl start rabbitmq-server

配置虚拟主机及用户

用户角色
- RabbitMQ安装好后, 访问http://ip地址:15672, 自带了guest/guest用户名密码登录, 如需自定义用户, 也可进行如下管理
- 角色说明
  - 超级管理员(administrator): 可登陆管理控制台, 可查看所有的信息, 并且可以对用户, 策略(policy)进行操作.
  - 监控者(monitoring): 可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
  - 策略制定者(policymaker): 可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分).
  - 普通管理者(management): 仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理.
  - 其他: 无法登陆管理控制台，通常就是普通的生产者和消费者.
Virtual Hosts配置
- RabbitMQ中有虚拟消息服务器Virtual Host, 每个Virtual Hosts相当于一个独立的RabbitMQ服务器, 每个VirtualHost之间是隔离的, exchange, queue, message不互通. 一般Virtual Name以"/"开头
- 创建Virtual Hosts
- 设置VirtualHost权限

RabbitMQ入门-简单模式

搭建示例

通信过程图解

加入依赖

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

编写生产者

public class Producer {

    static final String QUEUE_NAME = "simple_queue";

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

        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();

        //主机地址
        connectionFactory.setHost("192.168.47.132");
        //连接端口: 默认5672
        connectionFactory.setPort(5672);
        //虚拟主机名: 默认 /
        connectionFactory.setVirtualHost("/yellowstreak");
        //连接用户名: 默认guest
        connectionFactory.setUsername("yellowstreak");
        //连接密码: 默认guest
        connectionFactory.setPassword("123456");

        //创建连接
        Connection connection = connectionFactory.newConnection();
        //创建频道 - 根据频道通信
        Channel channel = connection.createChannel();

        /**
         * 参数1: 队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        //要发送的消息
        String message = "Hello World";
        /**
         * 参数1: 交换机exchange名称: 没有则使用默认Default Exchange
         * 参数2: 路由key, 简单模式下可以传递队列名称
         * 参数3: 消息的其他属性
         * 参数4: 参数内容, 要转换成字节数组
         */
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("已发送消息");
        // 关闭资源
        channel.close();
        connection.close();
    }
}

我们可以看到队列中有一条消息(队列在代码中设置了持久化, 而Connection, Channel资源释放了)

编写消费者

public class ConnectionUtil {

    public static Connection getConnection() throws IOException, TimeoutException {

        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setPort(5672);
        connectionFactory.setHost("192.168.47.132");
        connectionFactory.setVirtualHost("/yellowstreak");
        connectionFactory.setUsername("yellowstreak");
        connectionFactory.setPassword("123456");

        return connectionFactory.newConnection();
    }
}


public class Consumer {

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

        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //去连通生产者的队列.
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        //创建消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * @param consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * @param envelope 消息包的内容，可从中获取消息id，路由Key，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * @param properties 属性信息
             * @param body 消息
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key (这里是队列名)
                System.out.println(envelope.getRoutingKey());
                //消息id
                System.out.println(envelope.getDeliveryTag());
                //收到的消息
                System.out.println(new String(body, "utf-8"));
            }
        };//监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了, mq接收到回复会删除消息, 设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(Producer.QUEUE_NAME, true, consumer);

        //这里不关闭资源, 应该一直监听着..
        //这样只要生产者发出消息, 消费者就能收到并消费.

    }
}