当前位置： 代码迷 >> 综合 >> RabbitMQ -- part I
  详细解决方案

RabbitMQ -- part I

热度：72   发布时间：2023-10-13 11:53:32.0

主流消息中间件介绍- RabbitMQ

第一部分：介绍及架构

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

RabbitMQ -- part I

第二部分：核心概念

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

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

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

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

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

3.什么是AMQP高级协议?

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

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

RabbitMQ -- part I

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整体架构模型是什么样子的?

RabbitMQ -- part I

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

RabbitMQ -- part I

重点理解：（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]

RabbitMQ -- part I

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

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

RabbitMQ -- part I

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创建一个ChannelChannel channel = connection.createChannel();//4.通过channel发送数据for (int i = 0; i < 5; i++) {
    String msg = "Hello RabbitMQ!";//1 exchange 2 routingKeychannel.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创建一个ChannelChannel channel = connection.createChannel();//4 声明（创建）一个队列String queueName = "test001";channel.queueDeclare(queueName, true, false, false, null);//5.创建一个消费者QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//6.设置及启动channelchannel.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:接收消息，并根据路由键转发消息所绑定的队列

RabbitMQ -- part I

交换机属性：

  • 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必须完全匹配才会被队列接收，否则该消息会被抛弃.

RabbitMQ -- part I

实例程序：

生产者：

public class Producer4DirectExchange {
    public static void main(String[] args) throws Exception {
    //1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("47.98.155.175");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel 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、Consumerchannel.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

RabbitMQ -- part I

重点理解这个图：

RabbitMQ -- part I

实例程序：

生产者代码：

public class Producer4TopicExchange {
    public static void main(String[] args) throws Exception {
    //1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.11.76");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel 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、Consumerchannel.basicConsume(queueName, true, consumer);  //循环获取消息 while(true){
      //获取消息，如果没有消息，这一步将会一直阻塞 Delivery delivery = consumer.nextDelivery();  String msg = new String(delivery.getBody());    System.out.println("收到消息：" + msg);  } }
}

(3)Fanout Exchange

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

RabbitMQ -- part I

实例程序：

生产者：

public class Producer4FanoutExchange {
    public static void main(String[] args) throws Exception {
    //1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.11.76");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel 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、Consumerchannel.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创建一个ChannelChannel 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 routingKeychannel.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创建一个ChannelChannel channel = connection.createChannel();//4 声明（创建）一个队列String queueName = "test001";channel.queueDeclare(queueName, true, false, false, null);//5 创建消费者QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//6 设置Channelchannel.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 设置Channelchannel.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
  相关解决方案

代码迷 - 您身边的软件异常，代码异常，编程异常助手（发现，解决，分享）
Copyright © 2009-2013 DAIMAMI.COM. All rights reserved.