RabbitMQ

MQ基本概念

概述

MQ全称 Message Queue(消息队列)，存储消息的中间件，是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。

分布式系统通信两种方式:直接远程调用和借助第三方完成间接通信；发送方称为生产者，接收方称为消费者

优势

• 应用解耦：提高系统容错性和可维护性（系统的耦合性越高，容错性就越低，可维护性就越低）
• 异步提速：提升用户体验和系统吞吐量(单位时间内处理请求的数目)
• 削峰填谷：提高系统稳定性

劣势

• 系统可用性降低：

系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用?

• 系统复杂度提高：

MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费?（幂等性问题）怎么处理消息丢失情况?那么保证消息传递的顺序性?

• 一致性问题：

A系统外理完业务，通过 MO给、C、D二个系统发消息数据，如果B 系统、C系统处理成功，D系统处理失败。如何保证消息数据处理的一致性?

使用

生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空这才让明明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。

容许短暂的不一致性。

确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本

RabbitMQ概述

AMQP

即 Advanced Message Queuing Protocol (高级消息队列协议)，是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。2006年，AMQP 规范发布。类比HTTP。

2007年，Rabbit 技术公司基于AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发，Erlang 语言由Ericson 设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。

RabbitMQ基础架构如下图:

• Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker
• Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建exchange /queue 等
• Connection：publisher /consumer和 broker 之间的 TCP 连接
• Channel: 如果每一次访问 RabbitMQ 都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel作为轻量级的 Connection极大减少了操作系统建立TCP connection的开销
• Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有: direct (poi-to-point), topic (publish-subscribe) and fanout (multicast)
• Queue：消息最终被送到这里等待 consumer 取走
• Binding：exchange和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到exchange中的查询表中，用于 message的分发依据

RabbitMQ 提供了 6 种工作模式: 简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing路由模式、Topics 主题模式、RPC 远程调用模式(远程调用，不太算 MQ;暂不作介绍)。官网对应模式介绍:https://www.rabbitmq.com/getstarted.html

JMS

JMS 即Java 消息服务 (JavaMessage Service) 应用程序接口，是一个Java 平台中关于面向消息中间件的API

JMS是JavaEE 规范中的一种，类比JDBC

很多消息中间件都实现了JMS规范，例如: ActiveMQ。RabbitMQ 官方没有提供JMS 的实现包，但是开源社区有

RabbitMQ的工作模式

AMQPClient方式实现

根据架构写

<dependencies>
    <!--rabbitmq java客户端-->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.6.0</version>
    </dependency>
</dependencies>

• 简单模式 HelloWorld
  • 一个生产者、一个消费者，不需要设置交换机 (使用默认的交换机)。
• 工作队列模式Work Queue
  • 一个生产者、多个消费者(竞争关系)，不需要设置交换机 (使用默认的交换机)。
• 发布订阅模式Publish/subscribe
  • 需要设置类型为 fanout 的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列。
• 路由模式Routing
  • 需要设置类型为 direct 的交换机，交换机和队列进行绑定，并且指定 routing key，当发送消息到交换机交换机会根据routing key 消息发送到对应的队列。
• 通配符模式 Topic
  • 需要设置类型为 topic 的交换机，交换机和队列进行绑定，并且指定通配符方式的 routing key，当发送消息到交换机后，交换机会根据routing key 将消息发送到对应的队列。

简单模式Hello World

使用简单模式完成消息传递

1. 创建工程（生产者、消费者）
2. 加依赖
3. 编写生产者发送消息，消费者接收消息

P：生产者，也就是要发送消息的程序
C：消费者:消息的接收者，会一直等待消息到来；消费者不用关闭资源
queue：消息队列。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息

package com.lmcode.producer;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Channel;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


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

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.设置参数
        factory.setHost("196.168.88.130");//ip 默认localhost
        factory.setPort(5672);// 端口 默认5672
        factory.setVirtualHost("/lm");// 虚拟机 默认/
        factory.setUsername("lm");//默认guest
        factory.setPassword("1234");//默认guest
        //3.创建连接connection
        // Connection connection = factory.newConnection();
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel channel = connection.createChannel();
        //5.创建队列queue
        //如果没有一个名字叫hello_world的队列会自动创建，没有则不会
        /*
           参数
           queue：队列名称
           durable：是否持久化，当mq重启后还在
           exclusive：
              - 是否独占，只能有一个消费者监听这个队列
              - 当Connection关闭时，是否删除队列
           autoDelete：是否自动删除，当没有Consumer时自动删除
           argument：参数

          */
        channel.queueDeclare("hello_world",true,false,false,null);
        //6.发送消息
        /*
            参数：
            exchange：交换机名称，简单模式下交换机使用默认的""
            routingkey:路由名称
            props：配置信息
            body：发送消息数据
        */
        String body = "hello rabbitmq~~~";
        channel.basicPublish("","hello_world",null,body.getBytes());
        //7.释放资源
        channel.close();
        connection.close();

    }
}

package com.lmcode.comsumer;


import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


//发送消息
public class c_HelloWorld {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.设置参数
        factory.setHost("196.168.88.130");//ip 默认localhost
        factory.setPort(5672);// 端口 默认5672
        factory.setVirtualHost("/lm");// 虚拟机 默认/
        factory.setUsername("lm");//默认guest
        factory.setPassword("1234");//默认guest
        //3.创建连接connection
        // Connection connection = factory.newConnection();
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel channel = connection.createChannel();
        //5.创建队列queue
        //如果没有一个名字叫hello_world的队列会自动创建，没有则不会

        //接收消息
        /*
            参数：
            queue:队列名称
            autoAck：是否自动确认
            callback：回调对象
        */
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回调方法，当收到消息后，会自动执行该方法

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

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

    }
}

工作队列模式Work queues

与简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。竞争关系

Work Queues 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。两个人分担压力

例如: 短信服务部署多个只需要有一个节点成功发送即可

//5.创建队列queue
channel.queueDeclare("work_queue",true,false,false,null);
//6.发送消息
for (int i = 0; i < 10; i++) {
    String body = i+"hello rabbitmq~~~";
    channel.basicPublish("","work_queue",null,body.getBytes());
}

channel.basicConsume("work_queue",true,consumer);
//两个消费者

订阅模式Pub/Sub

一条消息可以被多个消费者消费

P：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X(交换机)

C：消费者，消息的接收者，会一直等待消息到来

Queue：消息队列，接收消息、缓存消息

Exchange: 交换机X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchane有常见以下3种类型

• Fanout:广播，将消息交给所有绑定到交换机的队列
• Direct:定向，把消息交给符合指定routing key 的队列
• Topic:通配符，把消息交给符合routing pattern (路由模式) 的队列

交换机只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失!

案例：打印日志，c1打印出来，c2存储

先声明交换机，指定交换机类型为FANOUT，这个类型将来绑定队列不需要指定routingkey

package com.lmcode.producer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


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

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.设置参数
        factory.setHost("196.168.88.130");//ip 默认localhost
        factory.setPort(5672);// 端口 默认5672
        factory.setVirtualHost("/lm");// 虚拟机 默认/
        factory.setUsername("lm");//默认guest
        factory.setPassword("1234");//默认guest
        //3.创建连接connection
        // Connection connection = factory.newConnection();
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel channel = connection.createChannel();
        //5.创建交换机
        /*
            exchange:交换机名称
            type:交换机类型
            - DIRECT("direct"):定向
            - FANOUT("fanout"):扇形(广播)，发送消息所有队列
            - TOPIC("topic"):通配符方法
            - HEADERS("headers"):参数匹配
            durable：是否持久化
            autoDelete：自动删除
            internal：内部使用，一般false
            argument：参数
        */
        String exchangeName = "test_fanout";
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
        //6.创建队列queue
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);
        //7.绑定队列和交换机
        /*
            queue:队列名称
            exchange：交换机名称
            routingkey：路由键，绑定规则
            - 如果交换机的类型为fanout，routingkey设置为""
        */
        channel.queueBind(queue1Name,exchangeName,"");
        channel.queueBind(queue2Name,exchangeName,"");
        //8.发送消息
        String body = "日志级别:info...";
        channel.basicPublish(exchangeName,"",null,body.getBytes());
        //9.释放资源
        channel.close();
        connection.close();

    }
}

不用声明队列

//4.创建Channel
Channel channel = connection.createChannel();
String queue1Name = "test_fanout_queue1";

//接收消息
Consumer consumer = new DefaultConsumer(channel){
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body){
        System.out.println("body: "+new String(body));
        System.out.println("打印");
    }
};
channel.basicConsume(queue1Name,true,consumer);

//4.创建Channel
Channel channel = connection.createChannel();
String queue2Name = "test_fanout_queue2";

//接收消息
Consumer consumer = new DefaultConsumer(channel){
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body){
        System.out.println("body: "+new String(body));
        System.out.println("保存");
    }
};
channel.basicConsume(queue2Name,true,consumer);

路由模式Routing

队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey (路由key)

消息的发送方在向Exchange发送消息时，也必须指定消息的 RoutingKey

Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的Routingkey 与消息的 Routing key 完全一致，才会接收到消息

Routing 模式要求队列在绑定交换机时要指定 routing key，消息会转发到符合 routing key 的队列。

package com.lmcode.producer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


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

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.设置参数
        factory.setHost("196.168.88.130");//ip 默认localhost
        factory.setPort(5672);// 端口 默认5672
        factory.setVirtualHost("/lm");// 虚拟机 默认/
        factory.setUsername("lm");//默认guest
        factory.setPassword("1234");//默认guest
        //3.创建连接connection
        // Connection connection = factory.newConnection();
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel channel = connection.createChannel();
        //5.创建交换机
        String exchangeName = "test_direct";
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);
        //6.创建队列queue
        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";


        channel.queueDeclare("hello_world",true,false,false,null);
        //7.绑定队列和交换机
        channel.queueBind(queue1Name,exchangeName,"error");
        channel.queueBind(queue2Name,exchangeName,"info");
        channel.queueBind(queue2Name,exchangeName,"error");
        channel.queueBind(queue2Name,exchangeName,"warning");
        //8.发送消息
        //- 发送消息时除了指定交换机，还要指定消息对应的routingkey
        String body = "日志级别：info";
        channel.basicPublish(exchangeName,"info",null,body.getBytes());//此时只有队列2能收到消息
        //9.释放资源
        channel.close();
        connection.close();

    }
}

package com.lmcode.producer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


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

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.设置参数
        factory.setHost("196.168.88.130");//ip 默认localhost
        factory.setPort(5672);// 端口 默认5672
        factory.setVirtualHost("/lm");// 虚拟机 默认/
        factory.setUsername("lm");//默认guest
        factory.setPassword("1234");//默认guest
        //3.创建连接connection
        // Connection connection = factory.newConnection();
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel channel = connection.createChannel();
        //5.创建交换机
        String exchangeName = "test_direct";
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);
        //6.创建队列queue
        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";


        channel.queueDeclare("hello_world",true,false,false,null);
        //7.绑定队列和交换机
        channel.queueBind(queue1Name,exchangeName,"error");
        channel.queueBind(queue2Name,exchangeName,"info");
        channel.queueBind(queue2Name,exchangeName,"error");
        channel.queueBind(queue2Name,exchangeName,"warning");
        //8.发送消息
        //- 发送消息时除了指定交换机，还要指定消息对应的routingkey
        String body = "日志级别：info";
        //测试，修改为error和warning
        channel.basicPublish(exchangeName,"info",null,body.getBytes());//此时只有队列2能收到消息
        //9.释放资源
        channel.close();
        connection.close();

    }
}

通配符模式Topics

Topic 主题模式可以实现 Pub/Sub 发布与订阅模式和 Routing 路由模式的功能，只是 Topic 在配置routing key的时候可以使用通配符，显得更加灵活。

*是一个单词，#是0到多个单词

package com.lmcode.producer;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


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

        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.设置参数
        factory.setHost("196.168.88.130");//ip 默认localhost
        factory.setPort(5672);// 端口 默认5672
        factory.setVirtualHost("/lm");// 虚拟机 默认/
        factory.setUsername("lm");//默认guest
        factory.setPassword("1234");//默认guest
        //3.创建连接connection
        // Connection connection = factory.newConnection();
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel channel = connection.createChannel();
        //5.创建交换机
        String exchangeName = "test_topic";
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,null);
        //6.创建队列queue
        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";


        channel.queueDeclare("hello_world",true,false,false,null);
        //7.绑定队列和交换机
        // - routing key两部分组成，系统名称.日志级别
        // - error级别存入DB、order系统的日志存入DB
        channel.queueBind(queue1Name,exchangeName,"#.error");
        channel.queueBind(queue1Name,exchangeName,"order.*");
        channel.queueBind(queue2Name,exchangeName,"*.*");

        //8.发送消息
        //- 发送消息时除了指定交换机，还要指定消息对应的routingkey
        String body = "日志级别：info";
        //测试，修改为error和warning
        channel.basicPublish(exchangeName,"order.info",null,body.getBytes());//此时队列1、2都能收到消息
        //9.释放资源
        channel.close();
        connection.close();

    }
}

package com.lmcode.comsumer;


import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


//发送消息
public class c_Topic1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.设置参数
        factory.setHost("196.168.88.130");//ip 默认localhost
        factory.setPort(5672);// 端口 默认5672
        factory.setVirtualHost("/lm");// 虚拟机 默认/
        factory.setUsername("lm");//默认guest
        factory.setPassword("1234");//默认guest
        //3.创建连接connection
        // Connection connection = factory.newConnection();
        Connection connection = factory.newConnection();
        //4.创建Channel
        Channel channel = connection.createChannel();

        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";
        //5.创建队列queue
        //如果没有一个名字叫hello_world的队列会自动创建，没有则不会

        //接收消息
        /*
            参数：
            queue:队列名称
            autoAck：是否自动确认
            callback：回调对象
        */
        Consumer consumer = new DefaultConsumer(channel){
            /*
                回调方法，当收到消息后，会自动执行该方法

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

            */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body){
                System.out.println("body: "+new String(body));
                System.out.println("存DB");
            }
        };
        channel.basicConsume(queue1Name,true,consumer);

    }
}

SpringBoot整合RabbitMQ

SpringBoot提供了快速整合RabbitMQ的方式

基本信息再yml中配置，队列交互机以及绑定关系在配置类中使用Bean的方式配置

生产端直接注入RabbitTemplate完成消息发送

消费端直接使用@RabbitListener完成消息接收

生产者

1. 引依赖写配置
2. 定义交换机，队列及绑定关系的配置类
3. 注入RabbitTemplate，调用方法，完成消息发送

<!--父工程依赖-->
<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.4.RELEASE</version>
</parent>
<dependencies>
    <!--RabbitMQ依赖-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <!--Junit单元测试-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
</dependencies>

spring:
  rabbitmq:
    host: 192.168.88.130
    port: 5672
    username: guest
    password: guest
    virtual-host: /

本栏目推荐文章
• 动力节点RabbitMQ教程|12小时学会rabbitmq消息中间件-01
• RabbitMQ（十二）
• RabbitMQ队列和消息的优先级
• RabbitMQ发布确认及备份交换机
• Rabbitmq
• RabbitMQ延迟队列
• RabbitMQ消息的生存时间TTL（Time To Live）
• PHP RabbitMQ 异步ACK持久化
• 在ASP.NET Core微服务架构下使用RabbitMQ如何实现CQRS模式
• RabbitMQ安装

RabbitMQrabbitmq消息 rabbitmq3 rabbitmq模式 rabbitmq个人 rabbitmq课程 rabbitmq docker docker-rabbitmq rabbitmq linux rabbitmq kafka rabbitmq模式07