本文曾经收录到Github仓库，该仓库蕴含计算机根底、Java根底、多线程、JVM、数据库、Redis、Spring、Mybatis、SpringMVC、SpringBoot、分布式、微服务、设计模式、架构、校招社招分享等外围知识点，欢送star~
Github地址：https://github.com/Tyson0314/Java-learning
什么是RabbitMQ？
RabbitMQ是一个由erlang开发的音讯队列。音讯队列用于利用间的异步合作。
RabbitMQ的组件
Message：由音讯头和音讯体组成。音讯体是不通明的，而音讯头则由一系列的可选属性组成，这些属性包含routing-key、priority、delivery-mode（是否持久性存储）等。
Publisher：音讯的生产者。
Exchange：接管音讯并将音讯路由到一个或多个Queue。default exchange 是默认的直连交换机，名字为空字符串，每个新建队列都会主动绑定到默认交换机上，绑定的路由键名称与队列名称雷同。
Binding：通过Binding将Exchange和Queue关联，这样Exchange就晓得将音讯路由到哪个Queue中。
Queue：存储音讯，队列的个性是先进先出。一个音讯可散发到一个或多个队列。
Virtual host：每个 vhost 实质上就是一个 mini 版的 RabbitMQ 服务器，领有本人的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的根底，必须在连贯时指定，RabbitMQ 默认的 vhost 是 / 。当多个不同的用户应用同一个RabbitMQ server提供的服务时，能够划分出多个vhost，每个用户在本人的vhost创立exchange和queue。
Broker：音讯队列服务器实体。
什么时候应用MQ
对于一些不须要立刻失效的操作，能够拆分进去，异步执行，应用音讯队列实现。
以常见的订单零碎为例，用户点击下单按钮之后的业务逻辑可能包含：扣减库存、生成相应单据、发短信告诉。这种场景下就能够用 MQ 。将短信告诉放到 MQ 异步执行，在下单的主流程（比方扣减库存、生成相应单据）实现之后发送一条音讯到 MQ， 让主流程疾速完结，而由另外的线程生产MQ的音讯。
RabbitMQ的优缺点
毛病：应用erlang实现，不利于二次开发和保护；性能较kafka差，长久化音讯和ACK确认的状况下生产和生产音讯单机吞吐量大概在1-2万左右，kafka单机吞吐量在十万级别。
长处：有治理界面，方便使用；可靠性高；功能丰富，反对音讯长久化、音讯确认机制、多种音讯散发机制。
RabbitMQ 有哪些重要的角色？
RabbitMQ 中重要的角色有：生产者、消费者和代理。
- 生产者：音讯的创建者，负责创立和推送数据到音讯服务器；
- 消费者：音讯的接管方，用于解决数据和确认音讯；
- 代理：就是 RabbitMQ 自身，用于表演“快递”的角色，自身不生产音讯，只是表演“快递”的角色。
Exchange 类型
Exchange散发音讯时依据类型的不同散发策略不同，目前共四种类型：direct、fanout、topic、headers 。headers 模式依据音讯的headers进行路由，此外 headers 交换器和 direct 交换器完全一致，但性能差很多。
Exchange规定。
| 类型名称 | 类型形容 |
|---|---|
| fanout | 把所有发送到该Exchange的音讯路由到所有与它绑定的Queue中 |
| direct | Routing Key==Binding Key |
| topic | 含糊匹配 |
| headers | Exchange不依赖于routing key与binding key的匹配规定来路由音讯，而是依据发送的音讯内容中的header属性进行匹配。 |
direct
direct替换机会将音讯路由到binding key 和 routing key齐全匹配的队列中。它是齐全匹配、单播的模式。
fanout
所有发到 fanout 类型交换机的音讯都会路由到所有与该交换机绑定的队列下来。fanout 类型转发音讯是最快的。
topic
topic交换机应用routing key和binding key进行含糊匹配，匹配胜利则将音讯发送到相应的队列。routing key和binding key都是句点号“. ”分隔的字符串，binding key中能够存在两种特殊字符“*”与“##”，用于做含糊匹配，其中“*”用于匹配一个单词，“##”用于匹配多个单词。
headers
headers交换机是依据发送的音讯内容中的headers属性进行路由的。在绑定Queue与Exchange时指定一组键值对；当音讯发送到Exchange时，RabbitMQ会取到该音讯的headers（也是一个键值对的模式），比照其中的键值对是否齐全匹配Queue与Exchange绑定时指定的键值对；如果齐全匹配则音讯会路由到该Queue，否则不会路由到该Queue。
音讯失落
音讯失落场景：生产者生产音讯到RabbitMQ Server音讯失落、RabbitMQ Server存储的音讯失落和RabbitMQ Server到消费者音讯失落。
音讯失落从三个方面来解决：生产者确认机制、消费者手动确认音讯和长久化。
生产者确认机制
生产者发送音讯到队列，无奈确保发送的音讯胜利的达到server。
解决办法：
- 事务机制。在一条音讯发送之后会使发送端阻塞，期待RabbitMQ的回应，之后能力持续发送下一条音讯。性能差。
- 开启生产者确认机制，只有音讯胜利发送到交换机之后，RabbitMQ就会发送一个ack给生产者（即便音讯没有Queue接管，也会发送ack）。如果音讯没有胜利发送到交换机，就会发送一条nack音讯，提醒发送失败。
在 Springboot 是通过 publisher-confirms 参数来设置 confirm 模式：
spring: rabbitmq: ##开启 confirm 确认机制 publisher-confirms: true在生产端提供一个回调办法，当服务端确认了一条或者多条音讯后，生产者会回调这个办法，依据具体的后果对音讯进行后续解决，比方从新发送、记录日志等。
// 音讯是否胜利发送到Exchangefinal RabbitTemplate.ConfirmCallback confirmCallback = (CorrelationData correlationData, boolean ack, String cause) -> { log.info("correlationData: " + correlationData); log.info("ack: " + ack); if(!ack) { log.info("异样解决...."); } };rabbitTemplate.setConfirmCallback(confirmCallback);路由不可达音讯
生产者确认机制只确保音讯正确达到交换机，对于从交换机路由到Queue失败的音讯，会被抛弃掉，导致音讯失落。
对于不可路由的音讯，有两种解决形式：Return音讯机制和备份交换机。
Return音讯机制
Return音讯机制提供了回调函数 ReturnCallback，当音讯从交换机路由到Queue失败才会回调这个办法。须要将mandatory 设置为 true ，能力监听到路由不可达的音讯。
spring: rabbitmq: ##触发ReturnCallback必须设置mandatory=true, 否则Exchange没有找到Queue就会抛弃掉音讯, 而不会触发ReturnCallback template.mandatory: true通过 ReturnCallback 监听路由不可达音讯。
final RabbitTemplate.ReturnCallback returnCallback = (Message message, int replyCode, String replyText, String exchange, String routingKey) -> log.info("return exchange: " + exchange + ", routingKey: " + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText);rabbitTemplate.setReturnCallback(returnCallback);当音讯从交换机路由到Queue失败时，会返回 return exchange: , routingKey: MAIL, replyCode: 312, replyText: NO_ROUTE。
备份交换机
备份交换机alternate-exchange 是一个一般的exchange，当你发送音讯到对应的exchange时，没有匹配到queue，就会主动转移到备份交换机对应的queue，这样音讯就不会失落。
消费者手动音讯确认
有可能消费者收到音讯还没来得及解决MQ服务就宕机了，导致音讯失落。因为音讯者默认采纳主动ack，一旦消费者收到音讯后会告诉MQ Server这条音讯曾经解决好了，MQ 就会移除这条音讯。
解决办法：消费者设置为手动确认音讯。消费者解决完逻辑之后再给broker回复ack，示意音讯曾经胜利生产，能够从broker中删除。当音讯者生产失败的时候，给broker回复nack，依据配置决定从新入队还是从broker移除，或者进入死信队列。只有没收到消费者的 acknowledgment，broker 就会始终保留着这条音讯，但不会 requeue，也不会调配给其余 消费者。
消费者设置手动ack：
##设置生产端手动 ackspring.rabbitmq.listener.simple.acknowledge-mode=manual音讯解决完，手动确认：
@RabbitListener(queues = RabbitMqConfig.MAIL_QUEUE) public void onMessage(Message message, Channel channel) throws IOException { try { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); } long deliveryTag = message.getMessageProperties().getDeliveryTag(); //手工ack；第二个参数是multiple，设置为true，示意deliveryTag序列号之前（包含本身）的音讯都曾经收到，设为false则示意收到一条音讯 channel.basicAck(deliveryTag, true); System.out.println("mail listener receive: " + new String(message.getBody())); }当音讯生产失败时，生产端给broker回复nack，如果consumer设置了requeue为false，则nack后broker会删除音讯或者进入死信队列，否则音讯会从新入队。
长久化
如果RabbitMQ服务异样导致重启，将会导致音讯失落。RabbitMQ提供了长久化的机制，将内存中的音讯长久化到硬盘上，即便重启RabbitMQ，音讯也不会失落。
音讯长久化须要满足以下条件：
- 音讯设置长久化。公布音讯前，设置投递模式delivery mode为2，示意音讯须要长久化。
- Queue设置长久化。
- 交换机设置长久化。
当公布一条音讯到交换机上时，Rabbit会先把音讯写入长久化日志，而后才向生产者发送响应。一旦从队列中生产了一条音讯的话并且做了确认，RabbitMQ会在长久化日志中移除这条音讯。在生产音讯前，如果RabbitMQ重启的话，服务器会主动重建交换机和队列，加载长久化日志中的音讯到相应的队列或者交换机上，保障音讯不会失落。
镜像队列
当MQ产生故障时，会导致服务不可用。引入RabbitMQ的镜像队列机制，将queue镜像到集群中其余的节点之上。如果集群中的一个节点生效了，能主动地切换到镜像中的另一个节点以保障服务的可用性。
通常每一个镜像队列都蕴含一个master和多个slave，别离对应于不同的节点。发送到镜像队列的所有音讯总是被间接发送到master和所有的slave之上。除了publish外所有动作都只会向master发送，而后由master将命令执行的后果播送给slave，从镜像队列中的生产操作实际上是在master上执行的。
音讯反复生产怎么解决？
音讯反复的起因有两个：1.生产时音讯反复，2.生产时音讯反复。
生产者发送音讯给MQ，在MQ确认的时候呈现了网络稳定，生产者没有收到确认，这时候生产者就会从新发送这条音讯，导致MQ会接管到反复音讯。
消费者生产胜利后，给MQ确认的时候呈现了网络稳定，MQ没有接管到确认，为了保障音讯不失落，MQ就会持续给消费者投递之前的音讯。这时候消费者就接管到了两条一样的音讯。因为反复音讯是因为网络起因造成的，无奈防止。
解决办法：发送音讯时让每个音讯携带一个全局的惟一ID，在生产音讯时先判断音讯是否曾经被生产过，保障音讯生产逻辑的幂等性。具体生产过程为：
- 消费者获取到音讯后先依据id去查问redis/db是否存在该音讯
- 如果不存在，则失常生产，生产结束后写入redis/db
- 如果存在，则证实音讯被生产过，间接抛弃
生产端怎么进行限流？
当 RabbitMQ 服务器积压大量音讯时，队列里的音讯会大量涌入生产端，可能导致生产端服务器奔溃。这种状况下须要对生产端限流。
Spring RabbitMQ 提供参数 prefetch 能够设置单个申请解决的音讯个数。如果消费者同时解决的音讯达到最大值的时候，则该消费者会阻塞，不会生产新的音讯，直到有音讯 ack 才会生产新的音讯。
开启生产端限流：
##在单个申请中解决的音讯个数，unack的最大数量spring.rabbitmq.listener.simple.prefetch=2原生 RabbitMQ 还提供 prefetchSize 和 global 两个参数。Spring RabbitMQ没有这两个参数。
//单条音讯大小限度，0代表不限度//global：限度限流性能是channel级别的还是consumer级别。当设置为false，consumer级别，限流性能失效，设置为true没有了限流性能，因为channel级别尚未实现。void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;什么是死信队列？
生产失败的音讯寄存的队列。
音讯生产失败的起因：
- 音讯被回绝并且音讯没有从新入队（requeue=false）
- 音讯超时未生产
- 达到最大队列长度
设置死信队列的 exchange 和 queue，而后进行绑定：
@Bean public DirectExchange dlxExchange() { return new DirectExchange(RabbitMqConfig.DLX_EXCHANGE); } @Bean public Queue dlxQueue() { return new Queue(RabbitMqConfig.DLX_QUEUE, true); } @Bean public Binding bindingDeadExchange(Queue dlxQueue, DirectExchange deadExchange) { return BindingBuilder.bind(dlxQueue).to(deadExchange).with(RabbitMqConfig.DLX_QUEUE); }在一般队列加上两个参数，绑定一般队列到死信队列。当音讯生产失败时，音讯会被路由到死信队列。
@Bean public Queue sendSmsQueue() { Map<String,Object> arguments = new HashMap<>(2); // 绑定该队列到私信交换机 arguments.put("x-dead-letter-exchange", RabbitMqConfig.DLX_EXCHANGE); arguments.put("x-dead-letter-routing-key", RabbitMqConfig.DLX_QUEUE); return new Queue(RabbitMqConfig.MAIL_QUEUE, true, false, false, arguments); }生产者残缺代码：
@Component@Slf4jpublic class MQProducer { @Autowired RabbitTemplate rabbitTemplate; @Autowired RandomUtil randomUtil; @Autowired UserService userService; final RabbitTemplate.ConfirmCallback confirmCallback = (CorrelationData correlationData, boolean ack, String cause) -> { log.info("correlationData: " + correlationData); log.info("ack: " + ack); if(!ack) { log.info("异样解决...."); } }; final RabbitTemplate.ReturnCallback returnCallback = (Message message, int replyCode, String replyText, String exchange, String routingKey) -> log.info("return exchange: " + exchange + ", routingKey: " + routingKey + ", replyCode: " + replyCode + ", replyText: " + replyText); public void sendMail(String mail) { //貌似线程不平安 范畴100000 - 999999 Integer random = randomUtil.nextInt(100000, 999999); Map<String, String> map = new HashMap<>(2); String code = random.toString(); map.put("mail", mail); map.put("code", code); MessageProperties mp = new MessageProperties(); //在生产环境中这里不必Message，而是应用 fastJson 等工具将对象转换为 json 格局发送 Message msg = new Message("tyson".getBytes(), mp); msg.getMessageProperties().setExpiration("3000"); //如果生产端要设置为手工 ACK ，那么生产端发送音讯的时候肯定发送 correlationData ，并且全局惟一，用以惟一标识音讯。 CorrelationData correlationData = new CorrelationData("1234567890"+new Date()); rabbitTemplate.setMandatory(true); rabbitTemplate.setConfirmCallback(confirmCallback); rabbitTemplate.setReturnCallback(returnCallback); rabbitTemplate.convertAndSend(RabbitMqConfig.MAIL_QUEUE, msg, correlationData); //存入redis userService.updateMailSendState(mail, code, MailConfig.MAIL_STATE_WAIT); }}消费者残缺代码：
@Slf4j@Componentpublic class DeadListener { @RabbitListener(queues = RabbitMqConfig.DLX_QUEUE) public void onMessage(Message message, Channel channel) throws IOException { try { Thread.sleep(5000); } catch (InterruptedException e) { e.printStackTrace(); } long deliveryTag = message.getMessageProperties().getDeliveryTag(); //手工ack channel.basicAck(deliveryTag,false); System.out.println("receive--1: " + new String(message.getBody())); }}当一般队列中有死信时，RabbitMQ 就会主动的将这个音讯从新公布到设置的死信交换机去，而后被路由到死信队列。能够监听死信队列中的音讯做相应的解决。
说说pull模式
pull模式次要是通过channel.basicGet办法来获取音讯，示例代码如下：
GetResponse response = channel.basicGet(QUEUE_NAME, false);System.out.println(new String(response.getBody()));channel.basicAck(response.getEnvelope().getDeliveryTag(),false);怎么设置音讯的过期工夫？
在生产端发送音讯的时候能够给音讯设置过期工夫，单位为毫秒(ms)
Message msg = new Message("tyson".getBytes(), mp);msg.getMessageProperties().setExpiration("3000");也能够在创立队列的时候指定队列的ttl，从音讯入队列开始计算，超过该工夫的音讯将会被移除。
参考链接
RabbitMQ根底
Springboot整合RabbitMQ
RabbitMQ之音讯长久化
RabbitMQ发送邮件代码
线上rabbitmq问题
最初给大家分享一个Github仓库，下面有大彬整顿的300多本经典的计算机书籍PDF，包含C语言、C++、Java、Python、前端、数据库、操作系统、计算机网络、数据结构和算法、机器学习、编程人生等，能够star一下，下次找书间接在下面搜寻，仓库继续更新中~
Github地址：https://github.com/Tyson0314/java-books