RabbitMQ入门与核心概念-阿里云开发者社区

初识RabbitMQ

rabbitMQ是一个开源的消息代理和队列服务器，用于通过普通协议在完全不同的应用之间共享数据，RabbitMQ是使用Erlang语言来编写的，并且RabbitMQ是基于AMQP协议的。

哪些大厂在用RabbitMQ，为什么？

滴滴、美团、头条、去哪儿、ppmoney
开源、性能优秀，稳定性保障
提供可靠的消息投递模式（confirm）、返回模式（return）
与SpringAMQP完美的整合，API丰富
集群模式丰富，表达式配置，HA模式，镜像队列模型
保证消息数据不丢失的前提做到高可靠性、可用性

RabbitMQ高性能的原因

Erlang语言最初是用于交换机领域的架构模式，这样的话RabbitMQ在Broker之间进行数据交互的性能是优秀的。
Erlang语言的特点：Erlang有着和原生Socket一样的延迟

什么是AMQP高级消息队列协议？

AMQP定义：是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计，是一个规范，RabbitMQ就是这个规范的一种实现。

AMQP协议模型

AMQP核心概念

Server:又称Broker，接受客户端的连接，实现AMQP实体服务
Connection：连接，应用程序与Broker的网络连接
Channel：网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道。客户端可建立多个Channel，每个Channel代表一个会话任务。
Message：消息，服务器与应用程序之间传递的数据，由Properties和Body组成。Properties可以对消息进行修饰，比如消息的优先级，延迟等高级特性；Body则是消息体内容。
Virtual host：虚拟地址，用于进行逻辑隔离，最上层的消息路由。一个Virtual host里面可以有若干个Exchange和Queue，同一个Virtual host里面不能有相同名称的Exchange和Queue（例如：redis有16个数据库，也是逻辑隔离的概念）
Exchange：交换机，接受消息，根据路由键转发消息到绑定的队列
Binding：Exchange和Queue之间的虚拟连接，binding中可以包含routing key
Routing key：一个路由规则，虚拟机可以用它来确定如何路由一个特定消息
Queue：也称为Message Queue，消息队列，保存消息并将它们转发给消费者

RabbitMQ的整体架构

生产者只需要关注把消息投递到指定的exchange即可，消费者监听指定的队列就可以了。通过这个图可以发现，生产者不关心我把消息投递到哪个队列，消费者也不关系我这个队列的消息是从哪个exchange来的，这两者完全没有耦合的情况。那么它们是怎么流转的，是通过exchange和queue的绑定关系。

消息流转图

RabbitMQ的安装

rabbitMQ之安装和配置(一)

rabbitmq之配置文件详解（二）

安装完成之后我们可以看到有三个与rabbitmq相关的命令

[root@bhz76 software]# rabbitmq
rabbitmqctl rabbitmq-plugins  rabbitmq-server 
[root@bhz76 software]#

服务的启动：rabbitmq-server start & 
服务的停止：rabbitmqctl stop_app
管理插件： rabbitmq-plugins enable rabbitmq_management

Troot@bhz76 software]# rabbitmq-server start
[1] 3268  启动rabbitmq  
root@bhz76 software]#
RabbitMQ 3.6.5.Copyright (C)2007-2016 Pivotal Software, Inc.
## ## Licensed under the MPL. See http://www.rabbitmq.com/  
## ##
##########  Logs:/var/log/rabbitmq/rabbit@bhz76.log 
###### ## /var/log/rabbitmq/rabbit@bhz76-sasl.log 
##########  这是主机名 
Starting broker...
completed with 0 plugins.

验证是否启动成功，出现如下内容则为成功

[root@bhz76 software]# lsof -i:5672
COMMAND PID USER  FD  TYPE DEVICE SIZE O  FF NODE NAME  
beam  3382 rabbitmq 48u IPv6  22841 0 t0 TCP *:amqp (LISTEN)

使用rabbitmq-plugins list命令查看所有已经存在的插件

[root@bhz76 software]# rabbitmq-plugins enable  rabbitmq management 
The following plugins have been enabled:  安装可视化rabbitmq管控台  
mochiweb webmachine
rabbitmq web dispatch amqp client
rabbitmq management agent rabbitmq management
Applying plugin configuration to rabbit@bhz76.. started 6 plugenis

查看管理控制台页面

命令行与管控台的关系

基本操作

rabbitmqctl stop_app  #关闭应用
rabbitmqctl start_app  #启动应用
rabbitmqctl status   #节点状态
rabbitmqctl add_user username password  #增加用户
rabbitmqctl list_users   #列出所有用户
rabbitmqctl delete_user username  #删除用户
rabbitmqctl clear_permissions -p vhostpath username  #清除用户权限
rabbitmqctl list_user_permissions username  #列出用户权限
rabbitmqctl change_password username newpassword  #修改密码
# 设置用户权限
rabbitmqctl set_permissions -p vhostpath username ".*"".*"".*"
rabbitmqctl add_vhost vhostpath  #创建虚拟主机
rabbitmqctl list_vhosts   #列出所有虚拟主机
rabbitmqctl list_permissions -p vhostpath  #列出虚拟主机上所有权限
rabbitmqctl delete_vhost vhostpath  #删除虚拟主机
rabbitmqctl list_queues   #查看所有队列信息
rabbitmqctl -p vhostpath purge_queue blue   #清除队列里的消息

高级操作

#移除所有数据，要在rabbitmqctl stop_app 之后使用
rabbitmqctl reset 
#[--ram] 指定节点的存储模式，ram是内存基本存储
rabbitmqctl join_cluster <clusternode> [--ram]  #组成集群命令
rabbitmqctl cluster_status   #查看集群状态
# 修改集群节点的存储形式   disc磁盘 ram内存
rabbitmqctl change_cluster_node_type disc|ram 
# 忘记节点（摘除节点）
rabbitmqctl forget_cluster_node [--offline]  
#修改节点名称
rabbitmqctl rename_cluster_node oldnode1 newnode1 [oldnode2] [newnode2...]

管控台有的操作，ctl命令行也都有

实例代码操作

maven 依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>3.6.5</version>
</dependency>
复制代码

生产者发布消息

package com.example.rabbitmq.quickstart;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Procuder {
    public static void main(String[] args) throws Exception {
        // 1 创建一个ConnectionFactory,并进行配置
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        //2 通过连接工程创建连接
        Connection connection = connectionFactory.newConnection();
        // 3 通过connection创建一个channel
        Channel channel = connection.createChannel();
        for (int i = 0; i < 5; i++) {
            String msg = "hello world!";
            // 1 exchange   2 routing key
            channel.basicPublish("", "test01", null, msg.getBytes());
        }
        channel.close();
        connection.close();
    }
}

消费者订阅处理消息

package com.example.rabbitmq.quickstart;
import com.rabbitmq.client.*;
import com.rabbitmq.client.QueueingConsumer.Delivery;
public class Consumer {
    public static void main(String[] args) throws Exception {
        // 1 创建一个ConnectionFactory,并进行配置
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        //2 通过连接工程创建连接
        Connection connection = connectionFactory.newConnection();
        // 3 通过connection创建一个channel
        Channel channel = connection.createChannel();
        //4 声明（创建）队列
        String queueName = "test01";
        channel.queueDeclare(queueName, true, false, false, null);
        //5 创建消费者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        // 6 设置channel
        channel.basicConsume(queueName, true, queueingConsumer);
        while (true) {
            Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.out.println("消费端消息：" + msg);
            Envelope envelope = delivery.getEnvelope();
        }
    }
}

exchange交换机

交换机属性

Name: 交换机名称，
Type: 交换机类型direct、topic、fanout、headers
Durability: 是否需要持久化，true为持久化
Auto Delete：当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
Internal：当前Exchange是否用于RabbitMQ内部使用，默认为false
ArgArgumens：扩展参数，用于扩展AMQP协议自制定化使用

Direct Exchange

所有发送到Direct Exchange的消息被转发到RouteKey中指定到Queue

⚠️注意：Direct模式可以使用RabbitMQ自带的Exchange：default Exchange，所以不需要将Exchange进行任何绑定（binding）操作，消息传递时，RouteKey必须完全匹配才会被队列接收，否则该消息会被抛弃。

示例代码

package com.bfxy.rabbitmq.api.exchange.direct;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
public class ConsumerDirectExchange {
  public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory() ;  
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(3000);
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();  
    //4 声明
    String exchangeName = "test_direct_exchange";
    String exchangeType = "direct";
    String queueName = "test_direct_queue";
    String routingKey = "test.direct";
    //表示声明了一个交换机
    channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
    //表示声明了一个队列
    channel.queueDeclare(queueName, false, false, false, null);
    //建立一个绑定关系:
    channel.queueBind(queueName, exchangeName, routingKey);
        //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //参数：队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, true, consumer);  
        //循环获取消息  
        while(true){  
            //获取消息，如果没有消息，这一步将会一直阻塞  
            Delivery delivery = consumer.nextDelivery();  
            String msg = new String(delivery.getBody());    
            System.out.println("收到消息：" + msg);  
        } 
  }
}

package com.bfxy.rabbitmq.api.exchange.direct;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class ProducerDirectExchange {
  public static void main(String[] args) throws Exception {
    //1 创建ConnectionFactory
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.11.76");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    //2 创建Connection
    Connection connection = connectionFactory.newConnection();
    //3 创建Channel
    Channel channel = connection.createChannel();  
    //4 声明
    String exchangeName = "test_direct_exchange";
    String routingKey = "test.direct111";
    //5 发送
    String msg = "Hello World RabbitMQ 4  Direct Exchange Message 111 ... ";
    channel.basicPublish(exchangeName, routingKey , null , msg.getBytes());     
  }
}

Topic Exchange

所有发送到Topic Exchange的消息会被转发到所有关心该RouteKey中指定Topic的Queue上

Exchange将RouteKey和某Topic进行模糊匹配，此时队列需要绑定一个Topic

⚠️注意：可以使用通配符进行模糊匹配

符号 "#" 匹配一个或多个词   
符号 "*" 匹配不多不少一个词
例如："log.#" 能够匹配到 "log.info.oa"
      "log.*" 只会匹配到 "log.error"

示例代码

package com.bfxy.rabbitmq.api.exchange.topic;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
public class ConsumerTopicExchange {
  public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory() ;  
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(3000);
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();  
    //4 声明
    String exchangeName = "test_topic_exchange";
    String exchangeType = "topic";
    String queueName = "test_topic_queue";
    //String routingKey = "user.*";
    String routingKey = "user.*";
    // 1 声明交换机 
    channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
    // 2 声明队列
    channel.queueDeclare(queueName, false, false, false, null);
    // 3 建立交换机和队列的绑定关系:
    channel.queueBind(queueName, exchangeName, routingKey);
        //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //参数：队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, true, consumer);  
        //循环获取消息  
        while(true){  
            //获取消息，如果没有消息，这一步将会一直阻塞  
            Delivery delivery = consumer.nextDelivery();  
            String msg = new String(delivery.getBody());    
            System.out.println("收到消息：" + msg);  
        } 
  }
}

package com.bfxy.rabbitmq.api.exchange.topic;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class ProducerTopicExchange {
  public static void main(String[] args) throws Exception {
    //1 创建ConnectionFactory
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.11.76");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    //2 创建Connection
    Connection connection = connectionFactory.newConnection();
    //3 创建Channel
    Channel channel = connection.createChannel();  
    //4 声明
    String exchangeName = "test_topic_exchange";
    String routingKey1 = "user.save";
    String routingKey2 = "user.update";
    String routingKey3 = "user.delete.abc";
    //5 发送
    String msg = "Hello World RabbitMQ 4 Topic Exchange Message ...";
    channel.basicPublish(exchangeName, routingKey1 , null , msg.getBytes()); 
    channel.basicPublish(exchangeName, routingKey2 , null , msg.getBytes());  
    channel.basicPublish(exchangeName, routingKey3 , null , msg.getBytes()); 
    channel.close();  
        connection.close();  
  }
}

Fanout Exchange

不处理路由键，只需要简单到将队列绑定到交换机上
发送到交换机的消息都会被转发到与该交换机绑定到所有队列上
Fanout交换机转发消息上最快的

示例代码

package com.bfxy.rabbitmq.api.exchange.fanout;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
public class ConsumerFanoutExchange {
  public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory() ;  
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(3000);
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();  
    //4 声明
    String exchangeName = "test_fanout_exchange";
    String exchangeType = "fanout";
    String queueName = "test_fanout_queue";
    String routingKey = ""; //不设置路由键
    channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
    channel.queueDeclare(queueName, false, false, false, null);
    channel.queueBind(queueName, exchangeName, routingKey);
        //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //参数：队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, true, consumer); 
        //循环获取消息  
        while(true){  
            //获取消息，如果没有消息，这一步将会一直阻塞  
            Delivery delivery = consumer.nextDelivery();  
            String msg = new String(delivery.getBody());    
            System.out.println("收到消息：" + msg);  
        } 
  }
}

package com.bfxy.rabbitmq.api.exchange.fanout;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class ProducerFanoutExchange {
  public static void main(String[] args) throws Exception {
    //1 创建ConnectionFactory
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.11.76");
    connectionFactory.setPort(5672);
    connectionFactory.setVirtualHost("/");
    //2 创建Connection
    Connection connection = connectionFactory.newConnection();
    //3 创建Channel
    Channel channel = connection.createChannel();  
    //4 声明
    String exchangeName = "test_fanout_exchange";
    //5 发送
    for(int i = 0; i < 10; i ++) {
      String msg = "Hello World RabbitMQ 4 FANOUT Exchange Message ...";
      channel.basicPublish(exchangeName, "", null , msg.getBytes());      
    }
    channel.close();  
        connection.close();  
  }
}

基础概念讲解

Binding-绑定

Exchange和Exchange、Queue之间的连接关系
Binding可以包含RoutingKey或者参数

Queue-消息队列

消息队列，实际存储消息数据
Durability：是否持久化，Durable：是，Transient：否
Auto Delete：如选yes，代表当最后一个监听被移除之后，该Queue会自动被删除

Message-消息

服务器与应用程序之间传递当数据
本质上就是一段数据，由propertie和Payload（Body）组成
常用属性：delivery mode、headers（自定义属性）、content_type、content_enconding、priority、correlation_id、reply_to、expiration、message_id、timestamp、type、user_id、app_id、cluster_id

Virtual host-虚拟主机

虚拟地址，用于进行逻辑隔离，最上层当消息路由
一个Virtual host里面可以由若干个Exchange和Queue
同一个Virtual host里面不能有相同名称的Exchange或Queue

更多内容阅读：

rabbitmq核心概念总结

中文文档

RabbitMQ入门与核心概念

初识RabbitMQ

哪些大厂在用RabbitMQ，为什么？

RabbitMQ高性能的原因

什么是AMQP高级消息队列协议？

AMQP协议模型

AMQP核心概念

RabbitMQ的整体架构

RabbitMQ的安装

命令行与管控台的关系

实例代码操作

exchange交换机

Direct Exchange

Topic Exchange

Fanout Exchange

基础概念讲解

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

RabbitMQ入门与核心概念

初识RabbitMQ

哪些大厂在用RabbitMQ，为什么？

RabbitMQ高性能的原因

什么是AMQP高级消息队列协议？

AMQP协议模型

AMQP核心概念

RabbitMQ的整体架构

RabbitMQ的安装

命令行与管控台的关系

实例代码操作

exchange交换机

Direct Exchange

Topic Exchange

Fanout Exchange

基础概念讲解

热门文章

最新文章

相关课程

相关电子书

相关实验场景