欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页

RabbitMQ笔记

程序员文章站 2022-07-15 08:59:43
...

这是目前的几个主流消息中间件,最终选择的是RabbitMQ,留个笔记

相关示例代码: https://download.csdn.net/download/qq877507054/11965512

RabbitMQ笔记

 

 

 

一.消息中间件

服务器之间的中转站

消息中间件是在消息的传输过程中保存消息的容器,消息中间件再将消息从它的源中继到它的目标时充当中间人的作用,队列的主要目的是提供路由并保证消息的传递,如果发送消息时,接收者不可用,消息队列会保留消息,知道可以成功传递它为止,当然,消息队列保存消息也是有期限的

1.采用异步处理模式

消息发送者可以发送一个消息而无需等待响应,消息发送者将消息发送到一条虚拟的通道(主题或队列)上,消息接收者则订阅或监听该通道,一条消息可能最终转发给一个或多个消息接收者,这些接收者都无需对消息发送者做出同步回应,整个过程是异步的

比如用户信息注册,注册完毕后过段时间发送邮件或者短信

2.应用程序和应用程序调用关系为松耦合关系

发送者和接收者不必了解对方,只需要确认消息发送者和接收者 不必同时在线,比如在线交易系统为了保证数据的最终一致,在支付系统支付完成后会把支付结果放到消息中间件里通知订单系统修改订单支付状态,两个系统通过消息中间件解耦

 RabbitMQ笔记

3.消息中间传递模型

  1. 点对点模型(PTP)

点对点模型用于消息生产者和消息消费者之间点到点的通信,消息生产者将消息发送到某个名字标识的特定消费者,这个名字实际上对应消息服务中的一个队列(Queue),在消息传递给消费者之前,它被存储在这个队列中,队列消息可以放在内存中也可以是持久的,以保证在服务出现故障时仍然能够传递消息

特性

  1. )每个消息只用一个消费者
  2. )发送者和接受者没有时间依赖
  3. )接受者确认消息接受和处理成功

RabbitMQ笔记

2.发布-订阅模型(pub/sub)

 

发布者/订阅者模型支持向一个特定的消息主题生产消息,0或多个订阅者可能对接收来自特定消息主题的消息感兴趣,在这种模型下,发布者和订阅者彼此不知道对方,这种模型好比是匿名公告板,这种模型被概括为:多个消费者可以获得消息,在发布者和订阅者之间存在时间依赖性,发布者需要建立一个订阅(subscription),以便能够消费者订阅,订阅者必须保持持续的活动状态以接收消息,除非订阅者建立了持久的订阅,在这种情况下,在订阅者未连接时发布的消息将在订阅者重新连接时重新发布

特性

1.)每个消息可以有多个订阅者

2.)客户端只有订阅后才能接受到消息,发布者和订阅者有时间依赖

接收者和发布者只有建立订阅关系才能收到消息

3.)持久订阅和非持久订阅

  持久订阅:订阅关系建立后,消息就不会消失,不管订阅者是否都在线,(常用,订阅关系存起来)

  非持久化订阅:订阅者为了接受消息,必须一直在线,当只有一个订阅者时约等于点对点模型(长连接)

RabbitMQ笔记

4.互联网消息中间件的应用场景

  1. 网站用户注册,注册成功后发送邮件确认或者短信通知
  2. 把日志进行集中收集,用于计算pv,用户行为分析
  3. 延迟消息发送和暂存,把消息中间件当成可靠的消息暂存地
  4. 消息广播,比如缓存数据的同步更新,往应用推送数据,更新本地缓存

 

消息中间件的分类

  1. (push)推消息模型:消息生产者将消息发送给消息传递服务,消息传递服务又将消息推给消息消费者
  2. (pull) 拉消息模型:消费者请求消息服务接收消息,消息生产者从消息中间件拉该消息

RabbitMQ笔记

服务端(消息中间件),客户端(消息的消费者)

 

二.RabbitMQ

  1. rabbitMQ是什么

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的

RabbitMQ支持以下操作系统：

Windows

Linux/Unix

MAC OS X

MQ全称为Message Queue, 是一种分布式应用程序的的通信方法，它是消费-生产者模型的一个典型的代表，producer往消息队列中不断写入消息，而另一端consumer则可以读取或者订阅队列中的消息。RabbitMQ是MQ产品的典型代表，是一款基于AMQP协议可复用的企业消息系统。业务上，可以实现服务提供者和消费者之间的数据解耦，提供高可用性的消息传输机制，在实际生产中应用相当广泛。本文意在介绍Rabbitmq的基本原理，包括rabbitmq基本框架，概念，通信过程等

RabbitMQ,遵循AMQP协议，由内在高并发的erlang语言开发，用在实时的对可靠性要求比较高的消息传递上。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。
AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

　　学过websocket的来理解rabbitMQ应该是非常简单的了，websocket是基于服务器和页面之间的通信协议，一次握手，多次通信。 而rabbitMQ就像是服务器之间的socket，一个服务器连上MQ监听，而另一个服务器只要通过MQ发送消息就能被监听服务器所接收。

　　但是MQ和socket还是有区别的，socket相当于是页面直接监听服务器。而MQ就是服务器之间的中转站，例如邮箱，一个人投递信件给邮箱，另一个人去邮箱取，他们中间没有直接的关系，所以耦合度相比socket小了很多。

RabbitMQ笔记

扩展

 

1、AMQP

AMQP，即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有 RabbitMQ等。

 

2、Erlang语言

Erlang是一种通用的并发程序设计语言，它由乔·阿姆斯特朗（Joe Armstrong）在瑞典电信设备制造商爱立信所辖的计算机科学研究室开发，目的是创造一种可以应付大规模开发活动的程序设计语言和运行环境。Erlang于1987年发布正式版本，最早是爱立信拥有的私有软件，经过十年的发展，于1998年发表开放源代码版本。

 

Erlang是运作于虚拟机的解释型语言，但是现在也包含有乌普萨拉大学高性能Erlang计划（HiPE）[3]开发的原生代码编译器，自R11B-4版本开始，Erlang也支持脚本方式执行。在编程范型上，Erlang属于多重典范编程语言，涵盖函数式、并行及分布式。循序运行的Erlang是一个及早求值, 单次赋值和动态类型的函数式编程语言。

 

3.消息队列(Message Queue)

在计算机科学中，消息队列（英语：Message queue）是一种进程间通信或同一进程的不同线程间的通信方式，软件的贮列用来处理一系列的输入，通常是来自用户。消息队列提供了异步的通信协议，每一个贮列中的纪录包含详细说明的数据，包含发生的时间，输入设备的种类，以及特定的输入参数，也就是说：消息的发送者和接收者不需要同时与消息队列交互。消息会保存在队列中，直到接收者取回它。

 

实际上，消息队列常常保存在链表结构中。拥有权限的进程可以向消息队列中写入或读取消息。

目前，有很多消息队列有很多开源的实现，包括JBoss Messaging、JORAM、Apache ActiveMQ、Sun Open Message Queue、RabbitMQ、IBM MQ、Apache Qpid和HTTPSQS。

 

 

 

 

2.MQ使用场景

别啥固定式使用场景了，说的透彻一点，他就是服务器之间通信的，前面博文中提到的Httpclient也可以做到，但是这个相对于其他通信在中间做了一个中间仓库。

　　好处1：降低了两台服务器之间的耦合，哪怕是一台服务器挂了，另外一台服务器也不会报错或者休克，反正他监听的是MQ，只要服务器恢复再重新连上MQ发送消息，监听服务器就能再次接收。

　　好处2：MQ作为一个仓库，本身就提供了非常强大的功能，例如不再是简单的一对一功能，还能一对多，多对一，自己脑补保险箱场景，只要有特定的密码，谁都能存，谁都能取。也就是说能实现群发消息和以此衍生的功能。

RabbitMQ笔记

好处3：现在普遍化的持久化功能，当MQ挂掉可以存储在磁盘等下重启恢复。（需要设置）

你是否遇到过两个（多个）系统间需要通过定时任务来同步某些数据？你是否在为异构系统的不同进程间相互调用、通讯的问题而苦恼、挣扎？如果是，那么恭喜你，消息服务让你可以很轻松地解决这些问题。
消息服务擅长于解决多系统、异构系统间的数据交换（消息通知/通讯）问题，你也可以把它用于系统间服务的相互调用（RPC）。

3.专业术语介绍

1. 生产者： 在现实生活中就好比制造商品的工厂，他们是商品的生产者。生产者只意味着发送。发送消息的程序称之为一个生产者。

2. 队列：rabbitMQ就像一个仓库，一个仓库里面可以 有很多队列，每个队列才是服务器之间消息通信的载体。

3.消费者：消费者就好比是从商店购买或从仓库取走商品的人，消费的意思就是接收。消费者是一个程序，主要是等待接收消息。

4.交换器：在生产者和消息队列之间的交换器，功能类似于网络宽带的交换机，可以根据不同的关键字，将信息发送到不同的队列。

RabbitMQ笔记

    上图的E就是交换器，通过关键字绑定，如果生产者给的消息中指定类型是ERROR，就给队列1，如果是INFO或者WARN就给队列2。当然也可以一个关键字绑定两个队列。（INFO等字段自己可以定义，也可以用*，#来匹配。*（星号）表示一个单词#（井号）表示零个或者多个单词。 比如ok.yes可以被ok.*匹配到）

5.临时队列：根据需求临时创建的一条队列，在断开连接后自动删除。

4.流程介绍

　　生产者发送一条消息给交换机——交换机根据关键字匹配到对应的队列——将消息存入队列——消费者从队列中取出消息使用。

 

5.一些细节说明

       MQ不能进行批量的消息处理，你看到的传输再多也只是消息被一条一条的存入队列，消费者从队列中一条一条的取出。这kafKa有区别，所以在效率上比不上kafKa，但是MQ主打的是稳定。

每次消费者取出消息时会通知队列，我拿到了，当队列接收到这条消息，就会把消息删除，这是默认的ACK机制。如果在接收消息之后，消费者挂掉，或者任何情况没有返回ack，队列中这条消息将不会删除，可以一直存着，等待其他消费者来取。    

注意，但是如果设置不返回ack，在不断的发送消息到队列又不删除，会导致MQ仓库瘫痪

 

 

6.系统架构

Rabbitmq系统最核心的组件是Exchange和Queue，下图是系统简单的示意图。Exchange和Queue是在rabbitmq server（又叫做broker）端，producer和consumer在应用端。

RabbitMQ笔记

producer&Consumer

producer指的是消息生产者，consumer消息的消费者。

Queue

消息队列，提供了FIFO的处理机制，具有缓存消息的能力。rabbitmq中，队列消息可以设置为持久化，临时或者自动删除。

  1. 设置为持久化的队列，queue中的消息会在server本地硬盘存储一份，防止系统crash，数据丢失
  2. 设置为临时队列，queue中的数据在系统重启之后就会丢失
  3. 设置为自动删除的队列，当不存在用户连接到server，队列中的数据会被自动删除

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

RabbitMQ笔记

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

RabbitMQ笔记

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

RabbitMQ笔记

Exchange

自己的理解:队列中的中转站

Exchange类似于数据通信网络中的交换机，提供消息路由策略。rabbitmq中，producer不是通过信道直接将消息发送给queue，而是先发送给Exchange。一个Exchange可以和多个Queue进行绑定，producer在传递消息的时候，会传递一个ROUTING_KEY(生产者提供路由key)，Exchange会根据这个ROUTING_KEY按照特定的路由算法，将消息路由给指定的queue。和Queue一样，Exchange也可设置为持久化，临时或者自动删除。

Exchange Types

RabbitMQ常用的Exchange Type有4种类型：direct(默认)，fanout, topic, 和headers,（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述）,不同类型的Exchange转发(路由)消息的策略有所区别,下面分别进行介绍。

1.Direct
直接交换器，工作方式类似于单播，Exchange会将消息发送完全匹配ROUTING_KEY的Queue

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。

RabbitMQ笔记

以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

 

2.fanout
广播式交换器，不管消息的ROUTING_KEY设置为什么，Exchange都会将消息转发给所有绑定的Queue。fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

RabbitMQ笔记

上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

 

 

3.topic
主题交换器，工作方式类似于组播，Exchange会将消息转发和ROUTING_KEY匹配模式相同的所有队列，比如，ROUTING_KEY为user.stock的Message会转发给绑定匹配模式为 * .stock,user.stock， * . * 和#.user.stock.#的队列。（ * 表是匹配一个任意词组，#表示匹配0个或多个词组）

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”

binding key与routing key一样也是句点号“. ”分隔的字符串

binding key中可以存在两种特殊字符“*”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

RabbitMQ笔记

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

 

4.headers
消息体的header匹配（ignore）

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。
在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。
该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

 

 

我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。(需要一个中转站)

RabbitMQ笔记

相关代码

MQConfig

package com.mq.rabbitmq.rabbitmq;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * 定义交换机,队列,banding
 */
@Configuration
public class MQConfig {

	public static final String MIAOSHA_QUEUE = "miaosha.queue";

	public static final String FANOUT_EXCHANGE = "fanoutxchage";//发送消息到所有与它绑定的Queue中

	public static final String TOPIC_EXCHANGE = "topicExchage";//主题交换器
	public static final String TOPIC_QUEUE1 = "topic.queue1";
	public static final String TOPIC_QUEUE2 = "topic.queue2";


	public static final String HEADERS_EXCHANGE = "headersExchage";//消息头交换器
	public static final String HEADER_QUEUE = "header.queue";

	/*********************************************Direct模式******************************************************************************/
	//直接交换机
	@Bean
	/**
	 * 队列名称
	 * 是否持久化
	 */
	public Queue miaoShaQueue() {
		return new Queue(MIAOSHA_QUEUE, true);
	}



	/*****************************************************Topic模式 ************************************************************************ */
	//主题交换机Exchange
	@Bean
	public TopicExchange topicExchage(){
		return new TopicExchange(TOPIC_EXCHANGE);
	}
	//主题队列1
	@Bean
	public Queue topicQueue1() {
		return new Queue(TOPIC_QUEUE1, true);
	}
	//主题队列2
	@Bean
	public Queue topicQueue2() {
		return new Queue(TOPIC_QUEUE2, true);
	}

	//主题交换机绑定主题队列1
	@Bean
	public Binding topicBinding1() {
		return BindingBuilder.bind(topicQueue1()).to(topicExchage()).with("topic.key1");
	}
	//主题交换机绑定主题队列2
	@Bean
	public Binding topicBinding2() {
		return BindingBuilder.bind(topicQueue2()).to(topicExchage()).with("topic.#");
	}



	/******************************************************Fanout模式*****************************************************************/
	//广播交换机Exchange,fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。
	@Bean
	public FanoutExchange fanoutExchage(){
		return new FanoutExchange(FANOUT_EXCHANGE);
	}


	//广播交换机绑定队列1
	@Bean
	public Binding FanoutBinding1() {
		return BindingBuilder.bind(topicQueue1()).to(fanoutExchage());
	}
	//广播交换机绑定队列2
	@Bean
	public Binding FanoutBinding2() {
		return BindingBuilder.bind(topicQueue2()).to(fanoutExchage());
	}





	/***************************************************************Header模式************************************************************/
	//headers类型的Exchange不依赖于routing key与binding key
	//消息体交换机Exchange
	@Bean
	public HeadersExchange headersExchage(){
		return new HeadersExchange(HEADERS_EXCHANGE);
	}
	//消息体队列1
	@Bean
	public Queue headerQueue1() {
		return new Queue(HEADER_QUEUE, true);
	}
	//消息体交换机绑定消息体队列1
	@Bean
	public Binding headerBinding() {
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("header1", "value1");
		map.put("header2", "value2");
		return BindingBuilder.bind(headerQueue1()).to(headersExchage()).whereAll(map).match();
	}


}

MQSender
package com.mq.rabbitmq.rabbitmq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;


@Service
public class MQSender {

	private static Logger log = LoggerFactory.getLogger(MQSender.class);
	
	@Autowired
    AmqpTemplate amqpTemplate ;

	public void send(Object msg) {
		log.info("send message:"+msg);
		amqpTemplate.convertAndSend(MQConfig.MIAOSHA_QUEUE, msg);//第一个参数是路由key

	}

	public void sendTopic(Object msg) {
		log.info("send topic message:"+msg);
		amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE, "topic.key1", msg+"1");
		amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE, "topic.key2", msg+"2");
	}

	public void sendFanout(Object msg) {
		log.info("send fanout message:"+msg);
		amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE, "", msg);
	}
	public void sendHeader(Object msg) {
		log.info("send Header message:"+msg);
		MessageProperties properties = new MessageProperties();
		properties.setHeader("header1", "value1");
		properties.setHeader("header2", "value2");
		Message obj = new Message(msg.toString().getBytes(), properties);
		amqpTemplate.convertAndSend(MQConfig.HEADERS_EXCHANGE, "", obj);
	}

	
	
}

MQReceiver
package com.mq.rabbitmq.rabbitmq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;


@Service
public class MQReceiver {

		private static Logger log = LoggerFactory.getLogger(MQReceiver.class);

	
		@RabbitListener(queues=MQConfig.MIAOSHA_QUEUE) //binding key
		public void receive(String message) {
			log.info("receive message:"+message);
		}
//
		@RabbitListener(queues=MQConfig.TOPIC_QUEUE1)
		public void receiveTopic1(String message) {
			log.info(" topic  queue1 message:"+message);
		}

		@RabbitListener(queues=MQConfig.TOPIC_QUEUE2)
		public void receiveTopic2(String message) {
			log.info(" topic  queue2 message:"+message);
		}

		@RabbitListener(queues=MQConfig.HEADER_QUEUE)
		public void receiveHeaderQueue(byte[] message) {
			log.info(" header  queue message:"+new String(message));
		}

		
}

controller 

package com.mq.rabbitmq.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.mq.rabbitmq.bean.QueueBean;
import com.mq.rabbitmq.rabbitmq.MQSender;
import com.mq.rabbitmq.utils.AjaxResponse;
import com.mq.rabbitmq.utils.MQConstant;
import com.mq.rabbitmq.utils.WebUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.List;

/**
 * rabbitmq增删改查
 */

@RestController
@RequestMapping("/api")
public class MQController {
    private static final Logger logger = LoggerFactory.getLogger(MQController.class);
    @Resource
    MQSender sender;
    @Resource
    WebUtils web;

    //直接模式
    @GetMapping("/sendings")
    public AjaxResponse<String> mq(String msg) {
        sender.send(msg);
        return AjaxResponse.success(msg);
    }

    //广播模式
    @GetMapping("/sendings/fanout")
    public AjaxResponse<String> fanout(String msg) {
        sender.sendFanout(msg);
        return AjaxResponse.success(msg);
    }

    //主题模式
    @GetMapping("/sendings/topic")
    public AjaxResponse<String> topic(String msg) {
        sender.sendTopic(msg);
        return AjaxResponse.success(msg);
    }

    //消息头模式
    @GetMapping("/sendings/header")
    public AjaxResponse<String> header(String msg) {
        sender.sendHeader(msg);
        return AjaxResponse.success(msg);
    }

    //获取全部队列
    @GetMapping("/queues")
    public AjaxResponse queues() throws Exception {
        String returnData=web.doGet(MQConstant.getQUEUES());
        JSONArray array=JSONObject.parseArray(returnData);
        List<QueueBean> list = JSONObject.parseArray(array.toJSONString(), QueueBean.class);
        return  AjaxResponse.success(list);
    }



}

 

MQConstant
public class MQConstant {
	public static final String 	OVERVIEW= "/api/overview";
	public static final String  CLUSTER_NAME= "/api/cluster-name";
	public static final String  NODES= "/api/nodes";
	public static final String  NODES_NAME= "/api/nodes/name";
	public static final String  EXTENSIONS = "/api/extensions";
	public static final String  DEFINITIONS= "/api/definitions";
	public static final String  ALL_CONFIGURATION= "/api/all-configuration";//(deprecated)
	public static final String  DEFINITIONS_VHOST= "/api/definitions/vhost";
	public static final String  CONNECTIONS= "/api/connections";
	public static final String  VHOSTS_VHOST_CONNECTIONS= "/api/vhosts/vhost/connections";
	public static final String  CONNECTIONS_NAME= "/api/connections/name";
	public static final String  CONNECTIONS_NAME_CHANNELS= "/api/connections/name/channels";
	public static final String  CHANNELS= "/api/channels";
	public static final String  VHOSTS_VHOST_CHANNELS= "/api/vhosts/vhost/channels";
	public static final String  CHANNELS_CHANNEL= "/api/channels/channel";
	public static final String  CONSUMERS= "/api/consumers";
	public static final String  CONSUMERS_VHOST= "/api/consumers/vhost";
	/*exchange*/
	public static final String  EXCHANGES= "/api/exchanges";
	public static final String  EXCHANGES_VHOST= "/api/exchanges/vhost";
	public static final String  EXCHANGES_VHOST_NAME= "/api/exchanges/vhost/name";
	public static final String  EXCHANGES_VHOST_NAME_BINDINGS_SOURCE= "/api/exchanges/vhost/name/bindings/source";
	public static final String  ECHANGES_VHOST_NAME_BINDINGS_DESTINATION= "/api/exchanges/vhost/name/bindings/destination";
	public static final String  EXCHANGES_VHOST_NAME_PUBLISH= "/api/exchanges/vhost/name/publish";
	/*queue*/
	public static final String  QUEUES= "/api/queues"; //获取全部队列信息
	public static final String  QUEUES_VHOST= "/api/queues/vhost";
	public static final String  QUEUES_VHOST_NAME= "/api/queues/vhost/name";
	public static final String  QUEUES_VHOST_NAME_BINDINGS= "/api/queues/vhost/name/bindings";
	public static final String  QUEUES_VHOST_NAME_CONTENTS= "/api/queues/vhost/name/contents";
	public static final String  QUEUES_VHOST_NAME_ACTIONS= "/api/queues/vhost/name/actions";
	public static final String  QUEUES_VHOST_NAME_GET= "/api/queues/vhost/name/get";
	/*binding*/
	public static final String  BINDINGS= "/api/bindings";
	public static final String  BINDINGS_VHOST= "/api/bindings/vhost";
	public static final String  BINDINGS_VHOST_E_EXCHANGE_Q_QUEUE= "/api/bindings/vhost/e/exchange/q/queue";
	public static final String  BINDINGS_VHOST_E_EXCHANGE_Q_QUEUE_PROPS= "/api/bindings/vhost/e/exchange/q/queue/props";
	public static final String  BINDINGS_VHOST_E_SOURCE_E_DESTINATION= "/api/bindings/vhost/e/source/e/destination";
	public static final String  BINDINGS_VHOST_E_SOURCE_EDESTINATION_PROPS= "/api/bindings/vhost/e/source/e/destination/props";
	/*vhost*/
	public static final String  VHOSTS= "/api/vhosts";
	public static final String  VHOSTS_NAME= "/api/vhosts/name";
	public static final String  VHOSTS_NAME_PERMISSIONS= "/api/vhosts/name/permissions";
	public static final String  VHOSTS_NAME_TOPIC_PERMISSIONS= "/api/vhosts/name/topic-permissions";
	public static final String  VHOSTS_NAME_START_NODE= "/api/vhosts/name/start/node";
	/*user*/
	public static final String  USERS= "/api/users/";
	public static final String  USER_WITHOUT_PERMISSIONS= "/api/users/without-permissions";
	public static final String  USERS_BULK_DELETE= "/api/users/bulk-delete";
	public static final String  USERS_NAME= "/api/users/name";
	public static final String  USERS_USER_PERMISSIONS= "/api/users/user/permissions";
	public static final String  USERS_USER_TOPIC_PERMISSIONS= "/api/users/user/topic-permissions";

	public static final String  WHOAMI= "/api/whoami";
	public static final String  PERMISSIONS= "/api/permissions";
	public static final String  PERMISSIONS_VHOST_USER= "/api/permissions/vhost/user";
	public static final String  TOPIC_PERMISSIONS= "/api/topic-permissions";
	public static final String  TOPIC_PERMISSIONS_VHOST_USEER= "/api/topic-permissions/vhost/user";
	public static final String  PARAMETERS= "/api/parameters";
	public static final String  PARAMETERS_COMPONENT= "/api/parameters/component";
	public static final String  PARAMETERS_COMPONENT_VHOST= "/api/parameters/component/vhost";
	public static final String  PARAMETERS_COMPONENT_VHOST_NAME= "/api/parameters/component/vhost/name";
	public static final String  GLOBAL_PARANTERS= "/api/global-parameters";
	public static final String  GLOBAL_PARAMETERS_NAME= "/api/global-parameters/name";
	public static final String  POLICIES= "/api/policies";
	public static final String  POLICIES_VHOST= "/api/policies/vhost";
	public static final String  POLICIES_VHOST_NAME= "/api/policies/vhost/name";
	public static final String  OPERATOR_POLICIES= "/api/operator-policies";
	public static final String  OPERATOR_POLICIES_VHOST= "/api/operator-policies/vhost";
	public static final String  OPERATOR_POLICIES_VHOST_NAME= "/api/operator-policies/vhost/name";
	public static final String  ALIVENESS_TEST_VHOST= "/api/aliveness-test/vhost";
	public static final String  HEALTHCHECKS_NODE= "/api/healthchecks/node";
	public static final String  HEALTHCHECKS_NODE_NODE= "/api/healthchecks/node/node";
	public static final String  VHOST_LIMITS= "/api/vhost-limits";
	public static final String  VHOST_LIMITS_VHOST= "/api/vhost-limits/vhost";
	public static final String VHOST_LIMITS_VHOST_NAME = "/api/vhost-limits/vhost/name";

 

 

 

 

Exchange是按照什么逻辑将消息路由到Queue的？


Binding

所谓绑定就是将一个特定的 Exchange 和一个特定的 Queue 绑定起来。Exchange 和Queue的绑定可以是多对多的关系。

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

RabbitMQ笔记

Binding key

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；生产者将消息发送给Exchange时，一般会指定一个routing key(生产者提供路由key);当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。
在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。
binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

routing key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。
在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。
RabbitMQ为routing key设定的长度限制为255 bytes。

 

virtual host

在rabbitmq server上可以创建多个虚拟的message broker(消息代理)，又叫做virtual hosts (vhosts)(虚拟主机)。每一个vhost本质上是一个mini-rabbitmq server，分别管理各自的exchange，和bindings。vhost相当于物理的server，可以为不同app提供边界隔离，使得应用安全的运行在不同的vhost实例上，相互之间不会干扰。producer和consumer连接rabbit server需要指定一个vhost。

virtual host其实是一个虚拟概念，类似于权限控制组，一个virtual host里面可以有若干个Exchange和Queue,但是权限控制的最小粒度是virtual host

 

7.通信过程

假设P1和C1注册了相同的Broker，Exchange和Queue。P1发送的消息最终会被C1消费。基本的通信流程大概如下所示：

  1. P1生产消息，发送给服务器端的Exchange
  2. Exchange收到消息，根据ROUTINKEY，将消息转发给匹配的Queue1
  3. Queue1收到消息，将消息发送给订阅者C1
  4. C1收到消息，发送ACK给队列确认收到消息(收到以后消息就可以删了,消息回执)
  5. Queue1收到ACK，删除队列中缓存的此条消息

Consumer收到消息时需要显式的向rabbit broker发送basic.ack消息或者consumer订阅消息时设置auto_ack参数为true。在通信过程中，队列对ACK的处理有以下几种情况：

  1. 如果consumer接收了消息，发送ack, rabbitmq会删除队列中这个消息，发送另一条消息给consumer。
  2. 如果cosumer接受了消息, 但在发送ack之前断开连接，rabbitmq会认为这条消息没有被deliver(发送),在consumer在次连接的时候，这条消息会被redeliver(再次发送)。
  3. 如果consumer接受了消息，但是程序中有bug,忘记了ack,rabbitmq不会重复发送消息。
  4. rabbitmq2.0.0和之后的版本支持consumer reject某条（类）消息，可以通过设置requeue参数中的reject为true达到目地，那么rabbitmq将会把消息发送给下一个注册的consumer。

 

 

8.API

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。

ConnectionFactory为Connection的制造工厂。

Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

9.消息回执

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。
这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

另外pub message是没有ack的。

10消息持久化

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

11.预取数量

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

RabbitMQ笔记

12.RPC

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。
但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

RabbitMQ笔记

RabbitMQ中实现RPC的机制是：

  • 客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
  • 服务器端收到消息并处理
  • 服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
  • 客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

 

利用RabbitMQ提供的这些功能就可以处理我们绝大部分的异步业务了。

 

 

 

 

13.安装

 

1.安装Erlang 

2.安装RabbitMQ

3.在rabbitmq的安装目录下找到sbin目录，进入并在此目录打开cmd窗口(进入sbin的cmd窗口)。输入rabbitmq-plugins enable rabbitmq_management (需要重启服务)

4.输入:http://localhost:15672浏览器访问,  用户名/密码 guest/guest

 

 

 

 

 

 

 

参考文档:

https://blog.csdn.net/wangbing25307/article/details/80845641
https://blog.csdn.net/chizizhixin/article/details/78488735
https://www.cnblogs.com/jun-ma/p/4840869.html
https://blog.csdn.net/whycold/article/details/41119807

 

 

 

相关标签： rabbitmq

上一篇： Nacos作为服务注册中心演示

下一篇： ActiveMQ 学习札记二

推荐阅读