【消息中间件】1小时快速上手RabbitMQ

1.MQ的介绍1.1 MQ的基本概念在这里插入图片描述在这里插入图片描述

1.2 MQ的优缺点下面是MQ的优缺点在这里插入图片描述

1.2.1 优点1：应用解耦考虑由订单系统与库存系统、支付系统、物流系统直接通过远程调用方式通信，模块耦合度较高的情况。在这里插入图片描述

会存在问题：

(1)容错性较低。

一个库存系统出问题，通信链路断了，订单下单就失败了。在这里插入图片描述

(2)可维护性低增加x系统，需要修改订单系统的代码。再加再改，撤回也改。在这里插入图片描述

引入消息中间件MQ后。订单系统作为消息生产者生产订单消息生产消息到MQ，库存、支付、物流作为消息消费者消费消息。即使库存系统出现问题，也可以在恢复以后再从MQ中消费消息，而不会导致订单失败。

在这里插入图片描述

扩展新的x系统，只需要x系统把MQ中的消息再消费一次即可，订单系统无需改代码。在这里插入图片描述

1.2.2 优点2：异步提速引入中间件之前，需要同步完成订单系统对其它模块的调用，即调用库存返回后调用支付，调用支付返回后调用物流。在这里插入图片描述引入中间件之后，25ms返回请求。在这里插入图片描述

1.2.3 优点3：削峰填谷引入消息中间件前，假如A系统来个秒杀活动，会寄。在这里插入图片描述引入之后，MQ就像一个大仓库，5000个请求，对MQ小意思。A系统慢慢从仓库里运货吧。在这里插入图片描述

在这里插入图片描述

1.2.4 缺点在这里插入图片描述

在这里插入图片描述

1.3常见的消息中间件在这里插入图片描述

2.RabbitMQ的介绍与安装2.1 RabbitMQ简介在这里插入图片描述在这里插入图片描述在这里插入图片描述在这里插入图片描述

2.2 JMS概念在这里插入图片描述

2.3 RabbitMQ安装官网地址：https://www.rabbitmq.com/

在这里插入图片描述往下翻一翻，下载软件和文档。在这里插入图片描述由于只是方便学习，本文基于windows安装rabbitmq，如果使用linux推荐使用docker进行部署。

2.3.1 安装依赖环境由于rabbitmq是基于erlang语言开发的，因此需要有erlang语言环境。

先看看官方文档中，对于erlang的版本要求说明。

https://www.rabbitmq.com/which-erlang.html

可以看到，不同版本的rabbitmq需要的erlang版本不同。这里我们使用3.11.5版的rabbitmq。因此需要的erlang版本是25.0-25.2在这里插入图片描述通过下列网址可以下载erlang25.2：https://www.erlang.org/patches/otp-25.2

选择下图中的windows installer，下载exe安装包。

在这里插入图片描述点击exe文件安装即可。

配置环境变量。在这里插入图片描述在这里插入图片描述

cmd键入erl -version判断是否配置成功

C:\Users\24724>erl -versionErlang (SMP,ASYNC_THREADS) (BEAM) emulator version 13.1.3122.3.2 安装rabbitmq下载exe安装包。https://www.rabbitmq.com/install-windows.html

版本选用：3.11.5在这里插入图片描述双击安装。从安装目录打开cmd，在这里插入图片描述使用如下命令安装管理页面的插件

rabbitmq-plugins enable rabbitmq_management1在这里插入图片描述

双击bat脚本启动在这里插入图片描述这时，打开浏览器输入http://localhost:15672，账号密码默认是：guest/guest

大功告成咯！

在这里插入图片描述

2.4 rabbitmq管控台的使用rabbitmq的管控台还是比较友好的，可以随便点点。这里带大家简单熟悉下。

添加用户。在这里插入图片描述在这里插入图片描述添加虚拟机。在这里插入图片描述点击添加的虚拟机ittest，进入详情页面为其授权。在这里插入图片描述在这里插入图片描述授权成功。在这里插入图片描述

其它的您自己点点点吧，很简单。

3.RabbitMQ快速入门3.1 需求描述RabbitMQ有六种工作模式，先来介绍最简单的一种，实现一个helloworld。

需求如下：在这里插入图片描述图示如下。在这里插入图片描述

3.2 准备工作idea新建项目rabbitqmdemo，maven模块rabbitmq-producer，rabbitmq-consumer在这里插入图片描述在两个模块中引入依赖。

<dependencies><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>5.1.7.RELEASE</version></dependency>

    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>2.1.8.RELEASE</version>
    </dependency>

    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>

    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-test</artifactId>
        <version>5.1.7.RELEASE</version>
    </dependency>
</dependencies>

123456789101112131415161718192021222324253.3 生产者实现producer下新建Producer_HelloWorld.java。

其要实现的代码逻辑其实看着下图就会了。

在这里插入图片描述代码如下。

/***

发送消息*/public class Producer_HelloWorld {public static void main(String[] args) throws IOException, TimeoutException {}}

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556代码运行前，guest用户的管控台是这样的，没有queue

在这里插入图片描述管控台变成了这个样子在这里插入图片描述

3.4 消费者实现consumer模块新建Consumer_HelloWorld.java。

public class Consumer_HelloWorld {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory = new ConnectionFactory();//2. 设置参数factory.setHost("localhost");//ip 默认值 localhostfactory.setPort(5672); //端口默认值 5672factory.setVirtualHost("/ittest");//虚拟机默认值/factory.setUsername("guest");//用户名默认 guestfactory.setPassword("guest");//密码默认值 guest//3. 创建连接 ConnectionConnection connection = factory.newConnection();//4. 创建ChannelChannel channel = connection.createChannel();//5. 创建队列Queue/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1. queue：队列名称2. durable:是否持久化，当mq重启之后，还在3. exclusive：* 是否独占。只能有一个消费者监听这队列* 当Connection关闭时，是否删除队列*4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉5. arguments：参数。

     */
    //如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建
    channel.queueDeclare("hello_world",true,false,false,null);

    /*
    basicConsume(String queue, boolean autoAck, Consumer callback)
    参数：
        1. queue：队列名称
        2. autoAck：是否自动确认
        3. callback：回调对象

     */
    // 接收消息
    Consumer consumer = new DefaultConsumer(channel){
        /*
            回调方法，当收到消息后，会自动执行该方法

            1. consumerTag：标识
            2. envelope：获取一些信息，交换机，路由key...
            3. properties:配置信息
            4. body：数据

         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            System.out.println("consumerTag："+consumerTag);
            System.out.println("Exchange："+envelope.getExchange());
            System.out.println("RoutingKey："+envelope.getRoutingKey());
            System.out.println("properties："+properties);
            System.out.println("body："+new String(body));
        }
    };
    channel.basicConsume("hello_world",true,consumer);


    //关闭资源？不要,因为要一直监听消息
}

}

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566执行结果如下，消费者成功的拿到了消息。在这里插入图片描述