[Java知识库]RabbitMQ -- part I

主流消息中间件介绍- RabbitMQ

第一部分：介绍及架构

RabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。AMQP协议更多用在企业系统内，对数据一致
性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。

第二部分：核心概念

1.互联网大厂为什么选择RabbitMQ?

滴滴、美团、头条、去哪儿、艺龙.

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

2.RabbitMQ的高性能之道是如何做到的?

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

3.什么是AMQP高级协议?

AMQP全称: Advanced Message Queuing Protocol(AMQP翻译:高级消息队列协议)

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

4.AMQP核心概念是什么?

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

5.RabbitMQ整体架构模型是什么样子的?

6.RabbitMQ消息是如何流转的?（重点）

重点理解：（1）你这个消息发送到那个Exchange上（2）发消息的时候需要带上Routing key即路由规则，通过这个规则在Exchange和Message Queue之间建立绑定规则，使得消息发送到指定的队列上。（3）Consumer端直接去监听这个队列就好了。

7.RabbitMQ安装与使用

1.官网地址: http://www.rabbitmq.com/
2.提前准备:安装Linux必要依赖包
3.下载RabbitMQ必须安装包
4.配置文件修改

安装3个东西（先将rpm包上传到liunx系统，另外必须按照顺序安装）：

（1）安装erlang环境：rpm -ivh erlang-18.3-1.e17 . centos.x8664 . rpm

（2）依赖：rpm -ivh socat-1. 7.3.2-5.e17. lux.x8664 . rpm

（3）rpm -ivh rabbitmq-server-3.6.5-1. noarch. rpm

（4）修改如下配置：vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app

只留下[guest]

(5)如下图：启动以后先需要安装管理插件
rabbitMQ 安装与使用

（6）当配置完成以后访问ip就可以看到控制台信息了（用户名和密码都为guest）。

8.命令行与管控台

基础操作：

• 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 reset:移除所有数据，要在rabbitmqctl stop_ app之后使用
• rabbitmqctl join_ cluster [–ram] :组成集群命令
• rabbitmqctl cluster_ status:查看集群状态
• rabbitmqctl change_ cluster_ node_ type disc| ram 修改集群节点的存储形式
• rabbitmqctl forget_ cluster_ node [–offline] 忘记节点(摘除节点)
• rabbitmqctl rename cluster_ node oldnode1 newnode1 [oldnode2] [newnode2…] (修改 节点名称)

9.RabbitMQ消息生产与消费

急速入门：

1. ConnectionFactory:获取连接工厂
2. Connection:一个连接
3. Channel:数据通信信道，可发送和接收消息
4. Queue:具体的消息存储队列
5. Producer & Consumer生产和消费者

Demo实例：

1. 新建Springboot项目，引入mq的pom

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

2.生产者

public class Producer {
    public static void main(String[] args) throws Exception {
        //1.创建ConnectionFactory获取连接工厂,并进行一些参数的配置
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("47.98.155.175");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        //2 通过连接工厂创建连接
        Connection connection = connectionFactory.newConnection();

        //3 通过connection创建一个Channel
        Channel channel = connection.createChannel();

        //4.通过channel发送数据
        for (int i = 0; i < 5; i++) {
            String msg = "Hello RabbitMQ!";
            //1 exchange   2 routingKey
            channel.basicPublish("", "test001", null, msg.getBytes());
        }
        //5 记得要关闭相关的连接
        channel.close();
        connection.close();

    }

重点理解：channel.basicPublish("", “test001”, null, msg.getBytes());

一般来讲，我们在发送一条消息的时候需要指定一个exchange,但是这里我们指定，那么它就会根据你写的routingKey去队列中找有没有相同名字的队列，如果有同名的就将消息路由到指定的队列。

3.消费端代码

public class Consumer {
    public static void main(String[] args) throws  Exception{
        //1 创建一个ConnectionFactory, 并进行配置
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("47.98.155.175");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        //2 通过连接工厂创建连接
        Connection connection = connectionFactory.newConnection();

        //3 通过connection创建一个Channel
        Channel channel = connection.createChannel();

        //4 声明（创建）一个队列
        String queueName = "test001";
        channel.queueDeclare(queueName, true, false, false, null);

        //5.创建一个消费者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        //6.设置及启动channel
        channel.basicConsume(queueName,true,queueingConsumer);

        //7.获取消息
        while (true){
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg=new String(delivery.getBody());
            System.out.println("消费端: " + msg);

        }
    }

10.RabbitMQ交换机详解

Exchange:接收消息，并根据路由键转发消息所绑定的队列

交换机属性：

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

交换机分类：

（1）Direct Exchang(直连交换机)

所有发送到Direct Exchange的消息被转发到RouteKey中指定的Queue注意: Direct模式可以使用RabbitMQ自带的Exchange: defaultExchange,所以不需要将Exchange进行任何绑定(binding)操作,消息传递时，RouteKey必须完全匹配才会被队列接收，否则该消息会被抛弃.

实例程序：

生产者：

public class Producer4DirectExchange {

	
	public static void main(String[] args) throws Exception {
		
		//1 创建ConnectionFactory
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("47.98.155.175");
		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.direct";
		//5 发送
		String msg = "Hello World RabbitMQ 4  Direct Exchange Message 111 ... ";
		channel.basicPublish(exchangeName, routingKey , null , msg.getBytes()); 	
		
	}
	
}

消费端：

public class Consumer4DirectExchange {

	public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory() ;
        
        connectionFactory.setHost("47.98.183.175");
        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);
		
        //创建消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //参数：队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, true, consumer);  
        //循环获取消息  
        while(true){  
            //获取消息，如果没有消息，这一步将会一直阻塞  
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String msg = new String(delivery.getBody());    
            System.out.println("收到消息：" + msg);  
        } 
	}
}

（2）Topic Exchange

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

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

重点理解这个图：

实例程序：

生产者代码：

public class Producer4TopicExchange {

	
	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();  
	}
	
}

消费端代码：

public class Consumer4TopicExchange {
	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);  
        } 
	}
}

(3)Fanout Exchange

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

实例程序：

生产者：

public class Producer4FanoutExchange {

	
	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();  
	}
	
}

消费端:

public class Consumer4FanoutExchange {

	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);  
        } 
	}
}

11.RabbitMQ队列、绑定、虚拟主机、消息

Binding-绑定

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

Queue-消息队列

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

Message-消息

• 服务器和应用程序之间传送的数据
• 本质上就是一段数据，由Properties和Payload ( Body )组成
• 常用属性: delivery mode、headers (自定义属性)

Message-其他属性

• content_ type、content encoding、priority
• correlation id、reply to、expiration、message_ id
• timestamp、type、user_ id、app_ id、cluster_ id

传递消息及定义属性实例

生产者：

public class Procuder {

	
	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 = connectionFactory.newConnection();
		
		//3 通过connection创建一个Channel
		Channel channel = connection.createChannel();
		
		Map<String, Object> headers = new HashMap<>();
		headers.put("my1", "111");
		headers.put("my2", "222");
		
		
		AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
				.deliveryMode(2)
				.contentEncoding("UTF-8")
				.expiration("10000")
				.headers(headers)
				.build();
		
		//4 通过Channel发送数据
		for(int i=0; i < 5; i++){
			String msg = "Hello RabbitMQ!";
			//1 exchange   2 routingKey
			channel.basicPublish("", "test001", properties, msg.getBytes());
		}

		//5 记得要关闭相关的连接
		channel.close();
		connection.close();
	}
}

消费者：

public class Consumer {

	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 = connectionFactory.newConnection();
		
		//3 通过connection创建一个Channel
		Channel channel = connection.createChannel();
		
		//4 声明（创建）一个队列
		String queueName = "test001";
		channel.queueDeclare(queueName, true, false, false, null);
		
		//5 创建消费者
		QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
		
		//6 设置Channel
		channel.basicConsume(queueName, true, queueingConsumer);
		
		while(true){
			//7 获取消息
			Delivery delivery = queueingConsumer.nextDelivery();
			String msg = new String(delivery.getBody());
			System.err.println("消费端: " + msg);
			Map<String, Object> headers = delivery.getProperties().getHeaders();
			System.err.println("headers get my1 value: " + headers.get("my1"));
			
			//Envelope envelope = delivery.getEnvelope();
		}
		
	}
}

Virtual host-虚拟主机

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

;

	//6 设置Channel
	channel.basicConsume(queueName, true, queueingConsumer);
	
	while(true){
		//7 获取消息
		Delivery delivery = queueingConsumer.nextDelivery();
		String msg = new String(delivery.getBody());
		System.err.println("消费端: " + msg);
		Map<String, Object> headers = delivery.getProperties().getHeaders();
		System.err.println("headers get my1 value: " + headers.get("my1"));
		
		//Envelope envelope = delivery.getEnvelope();
	}
	
}

}

### Virtual host-虚拟主机

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