MQ（消息中间件）概述及 RabbitMQ 的基本介绍-阿里云开发者社区

概述

消息中间件（MQ）

消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。流量削锋：削减峰值压力（秒杀，抢购）
MQ（Message Queue，消息队列）是典型的生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。
生产者：向消息中间件发送消息的一方
消费者：从中间获取消息使用的一方
broker：消息中间件
常见的消息中间件：

RabbitMQ：基于 AMQP 协议，erlang 语言开发，稳定性好
一款开源的消息管理系统，支持主流的操作系统，Linux、Windows、MacOX 等，支持多种开发语言，Java、Python、Ruby、.NET、PHP、C/C++、node.js 等
RocketMQ：基于 JMS，阿里巴巴产品，目前交由 Apache 基金会
ActiveMQ：基于JMS
kafka：分布式消息系统，高吞吐量

AMQP 和 JMS 标准

实现 MQ 有两种主流方式：AMQP、JMS。

AMQP，即 Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。
基于此协议的客户端与消息中间件可传递消息，并不受客户端中间件不同产品，不同的开发语言等条件的限制。
JMS，通常而言提到 JMS（Java MessageService）实际上是指 JMS API。JMS 是由 Sun 公司早期提出的消息标准，旨在为 Java 应用提供统一的消息操作，包括 create、send、receive等。
JMS 已经成为 java Enterprise Edition 的一部分。从使用角度看，JMS 和 JDBC 担任差不多的角色，用户都是根据相应的接口可以和实现了 JMS 的服务进行通信，进行相关的操作。

两者间的区别和联系：

JMS 是定义了统一的接口，来对消息操作进行统一
AMQP 是通过规定协议来统一数据交互的格式
JMS 限定了必须使用 Java 语言
AMQP 只是协议，不规定实现方式，因此是跨语言的
JMS 规定了两种消息模型点对点、发布订阅
AMQP 的消息模型更加丰富（7种），工作中常用的有 5 种

RabbitMQ 简述

基本概念

Broker：表示消息队列服务器实体
Message：消息，消息是不具名的，它由消息头和消息体组成。
消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。
Publisher：消息的生产者，也是一个向交换器发布消息的客户端应用程序。
Virtual Host：虚拟主机，表示一批交换器、消息队列和相关对象。
虚拟主机是共享相同的身份认证和加密环境的独立服务器域。
每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。
vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 /
Exchange：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Binding：绑定，用于消息队列和交换器之间的关联。
一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
Queue：消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。
一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
Connection：网络连接，比如一个 TCP 连接。
Channel：信道，多路复用连接中的一条独立的双向数据流通道。
信道是建立在真实的 TCP 连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。
因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
Consumer：消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

消息模型

RabbitMQ 提供了 6 种消息模型，但是第 6 种其实是 RPC，并不是MQ。其中 3、4、5 这三种都属于订阅模型，只不过进行路由的方式不同。

基本消息模型（直连）

图示：

P：生产者，也就是要发送消息的程序
C：消费者，即消息的接受者，会一直等待消息到来。
图中红色部分：queue（消息队列），类似一个邮箱，可以缓存消息；生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。

示例代码

生产者：

// 连接工具类
public class ConnectionUtil {
    // 建立与RabbitMQ的连接
    public static Connection getConnection() throws Exception {
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("127.0.0.1");
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setVirtualHost("/saas");
        factory.setUsername("saas");
        factory.setPassword("saas");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }
}

生产者发送消息：

public class Send {
    private final static String QUEUE_NAME = "simple_queue";
    public static void main(String[] argv) throws Exception {
        // 获取到连接
        Connection connection = ConnectionUtil.getConnection();
        // 从连接中创建通道，使用通道才能完成消息相关的操作
        Channel channel = connection.createChannel();
        // 声明（创建）队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 消息内容
        String message = "Hello World!";
        // 向指定的队列中发送消息
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
        //关闭通道和连接
        channel.close();
        connection.close();
    }
}

工作（任务）模型

工作模型，Work queues，也被称为（Task queues，任务模型）。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多，无法及时处理。此时就可以使用 work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。

缺点：当其中一个消费比较慢，比如一个消费者已经消费完。但是另一个消费者还在一个个的接收。这样会造成消息的积累。

优化方案：

消费者：要关闭自动确认消息
当前通道每一次只能消费一个消息

// 每次只能消费一个消息|一次只接收一条未确认的消息
channel.basicQos(1);

消息确认：手动确认消息

示例代码：

生产者

public class Provider {
    private static final String TASK_QUEUE_NAME = "task_queue";
    public static void main(String[] argv) throws Exception {
        // 获取连接对象
        Connection connection = RabbitMQUtils.getConnection();
        // 获取连接中的通道对象
        Channel channel = connection.createChannel();
        // 通过通道声明队列
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
        // 发送10条消息
        for (int i = 0; i < 20; i++) {
            channel.basicPublish("", TASK_QUEUE_NAME, null, (i+"你好啊").getBytes());
        }
    }
}

消费者（创建2个，区别是是否带下面这个线程休眠代码）

public class Customer2 {
    private static final String TASK_QUEUE_NAME = "task_queue";
    public static void main(String[] args) throws IOException {
        // 获取连接对象
        Connection connection = RabbitMQUtils.getConnection();
        // 获取连接中的通道对象
        Channel channel = connection.createChannel();
        // 每次只能消费一个消息
        channel.basicQos(1);
        // 通道绑定消息队列
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
        // 获取消息  参数2：消息自动确认，消费者自动向rabbitmq确认消息消费（只要消息队列有消息，就分配给消费者）
        channel.basicConsume(TASK_QUEUE_NAME, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者-1：" + new String(body));
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 参数1：确认队列中那个具体消息 参数2：是否开启多个消息同时确认
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        });
    }
}

发布订阅模型

1个生产者，多个消费者
每一个消费者都有自己的一个队列
每个队列都要绑定到交换机
生产者没有将消息直接发送到队列，而是发送到了交换机
生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

X（Exchanges）：交换机。一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于 Exchange 的类型。Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！

Exchange 类型有以下几种：

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

订阅模型-Fanout（广播）类型

广播模式的消息发送流程：

可以有多个消费者
每个消费者有自己的 queue（队列）
每个队列都要绑定到 Exchange（交换机）
生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
交换机把消息发送给绑定过的所有队列
队列的消费者都能拿到消息。实现一条消息被多个消费者消费

示例代码：

生产者

public class Provider {
    public static void main(String[] args) throws IOException {
        // 获取连接对象
        Connection connection = RabbitMQUtils.getConnection();
        // 获取连接中的通道对象
        Channel channel = connection.createChannel();
        // 将通道声明指定交换机 参数1：交换机名称 参数2：交换机类型，fanout是广播类型
        channel.exchangeDeclare("logs","fanout");
        // 发送消息
        channel.basicPublish("logs","",null,"fanout type message".getBytes());
    }
}

消费者

public class Customer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        channel.exchangeDeclare("logs","fanout");
        // 创建临时队列
        String queue = channel.queueDeclare().getQueue();
        // 绑定交换机和队列
        channel.queueBind(queue,"logs","");
        // 消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1"+new String(body));
            }
        });
    }
}

public class Customer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        channel.exchangeDeclare("logs","fanout");
        // 创建临时队列
        String queue = channel.queueDeclare().getQueue();
        // 绑定交换机和队列
        channel.queueBind(queue,"logs","");
        // 消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2"+new String(body));
            }
        });
    }
}

订阅模型-Direct 类型（routing）

在 Fanout 模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，可能希望不同的消息被不同的队列消费。这时就要用到 Direct 类型的 Exchange。

在 Direct 模型下：

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由 key）
消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的 Routingkey 与消息的 Routing key 完全一致，才会接收到消息

场景举例：出现 error 时，既需要在控制台打印，又需要在日志中持久化

P：生产者，向 Exchange 发送消息，发送消息时，会指定一个 routing key
x：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列
C1：消费者，其所在队列指定了需要 routing key 为 error 的消息
C2：消费者，其所在队列指定了需要 routing key 为 info、 error、 warning 的消息

示例代码：

生产者：

public class Provider {
    public static void main(String[] args) throws IOException {
        // 创建connection
        Connection connection = RabbitMQUtils.getConnection();
        // 创建channel
        Channel channel = connection.createChannel();
        String exchangeName = "logs_direct";
        // 通道声明指定交换机 参数1：交换机名称 参数2：交换机类型，direct是路由模式
        channel.exchangeDeclare(exchangeName,"direct");
        // 发送消息
        channel.basicPublish(exchangeName,"error",null,("这是direct模型发布的基于route" +
                " kye: ["+"info"+"]发送的消息").getBytes());
    }
}

消费者：

public class Customer1 {
    public static void main(String[] args) throws IOException {
        // 1.建立连接
        Connection connection = RabbitMQUtils.getConnection();
        // 建立channel
        Channel channel = connection.createChannel();
        String exchangeName = "logs_direct";
        // 绑定交换机和通道
        channel.exchangeDeclare(exchangeName,"direct");
        // 创建队列
        String queue = channel.queueDeclare().getQueue();
        // 基于路由key ,绑定队列和交换机
        channel.queueBind(queue,exchangeName,"error");
        // 消费
        channel.basicConsume(queue,true,new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1："+ new String(body));
            }
        });
    }
}

public class Customer2 {
    public static void main(String[] args) throws IOException {
        // 1.建立连接
        Connection connection = RabbitMQUtils.getConnection();
        // 建立channel
        Channel channel = connection.createChannel();
        String exchangeName = "logs_direct";
        // 绑定交换机和通道
        channel.exchangeDeclare(exchangeName,"direct");
        // 创建队列
        String queue = channel.queueDeclare().getQueue();
        // 基于路由key ,绑定队列和交换机
        channel.queueBind(queue,exchangeName,"info");
        channel.queueBind(queue,exchangeName,"error");
        channel.queueBind(queue,exchangeName,"warning");
        // 消费
        channel.basicConsume(queue,true,new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2："+ new String(body));
            }
        });
    }
}

订阅模型-Topic 类型

Topic 类型的 Exchange 与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型 Exchange 可以让队列在绑定 Routing key 的时候使用通配符。

这种模型 Routingkey 一般都是由一个或多个单词组成，多个单词之间以"."分割，例如: item.insert

通配符：
  *   匹配不多不少恰好1个词
  #   匹配0个，1个或多个词
如:
  audit.#    匹配 audit.irs.corporate 或者 audit.irs 等
  audit.*    只能匹配 audit.irs

示例代码：

生产者

public class Provider {
    public static void main(String[] args) throws IOException {
        // 建立连接工厂
        Connection connection = RabbitMQUtils.getConnection();
        // 建立channel
        Channel channel = connection.createChannel();
        String exchangeName = "topics";
        // 交换机
        channel.exchangeDeclare(exchangeName,"topic");
        // 发消息
        String routekey = "user.save";
        channel.basicPublish(exchangeName,routekey,null,("这是topic动态路由模型发布的基于route" +
                " key :["+routekey+"]").getBytes());
    }
}

消费者

public class Customer1 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        channel.exchangeDeclare("topics","topic");
        // 创建临时队列
        String queue = channel.queueDeclare().getQueue();
        // 绑定交换机和队列
        channel.queueBind(queue,"topics","user.*");
        // 消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1"+new String(body));
            }
        });
    }
}

public class Customer2 {
    public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //通道绑定交换机
        channel.exchangeDeclare("topics","topic");
        // 创建临时队列
        String queue = channel.queueDeclare().getQueue();
        // 绑定交换机和队列
        channel.queueBind(queue,"topics","user.#");
        // 消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2"+new String(body));
            }
        });
    }
}

MQ（消息中间件）概述及 RabbitMQ 的基本介绍

概述

消息中间件（MQ）

AMQP 和 JMS 标准

RabbitMQ 简述

基本概念

消息模型

基本消息模型（直连）

工作（任务）模型

发布订阅模型

订阅模型-Fanout（广播）类型

订阅模型-Direct 类型（routing）

订阅模型-Topic 类型

RabbitMQ 控制台

管理界面

添加用户

创建 Virtual Hosts 并且分配权限

热门文章

最新文章

相关课程

相关电子书

相关实验场景