Topic和Direct、Fanout匹配解析

RabbitMQ 日志的初学者指南—如何查看、定位和分析日志

RabbitMQ: 开启消息历史记录

RabbitMQ是如何确定消息是否投递到队列中的

Rabbitmq消息确认机制，看完再也不怕Rabbitmq消息丢失了

Springboot中使用ConfirmCallback和ReturnCallback

RabbitTemplate的 发布确认 和 事务

解决使用RabbitTemplate操作RabbitMQ，发生The channelMax limit is reached. Try later.问题

【RabbitMQ-9】自定义配置线程池（线程池资源不足-MQ初始化队列&&MQ动态扩容影响）

springboot整合rabbitmq和ThreadPool实现异步调用

springboot 整合RabbitMQ yml配置文件配置交换机 队列信息

分布式事务之rabbitmq肉身实战

rabbitmq 怎么实现多个消费者同时接收一个队列的消息？

RabbitMQ，ZeroMQ，Kafka 是一个层级的东西吗？相互之间有哪些优缺点？

RabbitMQ在国内为什么没有那么流行？

消息队列的使用场景是怎样的？

学习哪个消息队列和 RPC 框架比较好呢？

顺序消息

长事务死锁等各种风险

只需要发布一个产品ID变更的通知，由下游系统来处理，可能更为合理

信息采集处理

Interface ApplicationContextAware 和 InitializingBean来获取我们在Configuration class中声明的exchange, queue, binding beans并调用channel的相应方法来声明

BasicRecover

消息重试

for the fastest possible throughput, manual acks should be disabled.

retry只能在自动ack模式下使用。如果一定要在手动ack模式下使用retry功能，需保证消息能在有限次重试过程中可以重试成功，否则超过重试次数，又没办法执行ack或者nack，消息就会一直处于unack,并不会转发到死信队列

#onMessage() from ChannelAwareMessageListener class. Then you can do it this way

可从队列删除 Message A，以避免一旦取出消息隐藏时长过期后该消息被再次接受并处理

长轮询

采取 Pull 的方式问题就简单了许多，由于 Consumer 是主动到服务端拉取数据，此时只需要降低自己访问频率即可。举例：如前端是 flume 等日志收集业务，不断向 CMQ 生产消息，CMQ 向后端投递，后端业务如数据分析等业务，效率可能低于生产者。

MessageReCoverer

五次重试后，消费处于一个未被确认的状态。因为需要你手动 ack！下次服务重启的时候，会继续消费这条消息。

仅仅是消费者内部进行了重试，换句话说就是重试跟mq没有任何关系。上述消费者代码不能添加try{}catch(){}，一旦捕获了异常，在自动 ack 模式下，就相当于消息正确处理了，消息直接被确认掉了，不会触发重试的。

它会根据方法的执行情况来决定是否确认还是拒绝

AmqpRejectAndDontRequeueException 异常的时候，则消息会被拒绝，且 requeue = false（不重新入队列）

则消息会被拒绝，且 requeue = true（如果此时只有一个消费者监听该队列，则有发生死循环的风险，多消费端也会造成资源的极大浪费，这个在开发过程中一定要避免的）。可以通过 setDefaultRequeueRejected（默认是true）去设置

channel.basicNack 与 channel.basicReject 的区别在于basicNack可以批量拒绝多条消息，而basicReject一次只能拒绝一条消息。

如果某个服务忘记确认 ACK 了，则 RabbitMQ 不会再发送此消息数据给它，只要程序还在运行，没确认的消息就一直是 Unacked 状态，无法被 RabbitMQ 重新投递。

抛出异常

requeue：被拒绝的是否重新入队列

broker回传给生产者的确认消息中deliver-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理。

对象内部只有一个 id 属性，用来表示当前消息的唯一性

：队列名称

清除给定队列的消息。

如果channel.basicNack(8, false, false);表示deliveryTag=8的消息处理失败且将该消息直接丢弃。

当consumer对消息进行ack以后就会将此消息移除，从而放入新的消息

Jackson2JsonMessageConverter

会触发: channel.basicNack(tag, false, true);, 这样会告诉rabbitmq该消息消费失败, 需要重新入队

可以看到, 虽然消息确实被消费了, 但是由于是手动确认模式, 而最后又没手动确认, 所以, 消息仍被rabbitmq保存

CorrelationData correlationData = new CorrelationData(correlationDataId.toString());

在需要使用消息的return机制时候，mandatory参数必须设置为true

Routing key does not match but still message gets sent to queue

java.net.SocketException: Connection reset

Consumer utilisation

而 fanout 的广播模式不利于增加多个 Queue

增加fetch值（这个根据我们的实际经验，lan内影响很有限）

消息消费速度变慢

使用@Input和@output都是用MessageChannel，这是不对的。@Output对MessageChannel，@Input对应SubscribableChannel

最好不要自定义输入输出在同一个类里面。这样，如果我们只调用生产者发送消息。会导致提示Dispatcher has no subscribers for channel

会导致提示Dispatcher has no subscribers for channel。并且会让我们发送消息的次数莫名减少几次

第四个参数是autoDelete：true表示服务器不在使用这个队列是会自动删除它

第三个参数是exclusive：true表示一个队列只能被一个消费者占有并消费

MQ中间件与最终用户提供的应用程序代码(生产者/消费者)之间的桥梁

编写消费者代码

The index is the index of the input or output binding. It is always 0 for typical single input/output function, so it’s only relevant for Functions with multiple input and output arguments.

StreamBridge模式

output - <functionName> + -out- + <index>

@EnableBinding：将定义通道的接口绑定到某个 Bean 以便于我们可以通过该 Bean 操作通道进行发送和接收消息。

发送延迟消息非常简单，首先我们需要在生产者、消费者的配置文件中指定交换机的类型是延迟交换机

'''demoRoutingKey'''

spring.cloud.stream.rocketmq.bindings..producer.sync=true

Since it's an expression, you'll need quotes: 'cities' or if the same producer sends to both, something like headers['whereToSendHeader'].

安装插件后会生成新的Exchange类型x-delayed-message，该类型消息支持延迟投递机制,接收到消息后并未立即将消息投递至目标队列中，而是存储在mnesia(一个分布式数据系统)表中，检测消息延迟时间，如达到可投递时间时并将其通过x-delayed-type类型标记的交换机类型投递至目标队列。

可以通过在类上监听队列，然后在类中写带有不同参数类型的方法，来接收不同的操作推送的信息

It will work if the @RabbitListener is defined at the method level instead of the class level because we can infer the generic type from the listener method parameter.

发布-订阅、消费组、分区

rabbiimq延时插件

args.put("x-delayed-type", "direct");

编写配置类

两者中较小的值，即队列无消费者连接的消息过期时间，或者消息在队列中一直未被消费的过期时间

我们看到两个消费者都收到了消息

创建一个消费者组

我们需要监听之前创建的通道greetingChannel。让我们为它创建一个绑定

使用SubscribableChannel和@Input注解连接到greetingChannel，消息数据将被推送这里

发现是给用户授予了角色，只能登录控制台，但是没有给读写以及管理队列的权限，通过控制台admin按钮查看

真正地消费/处理消息

UnicastingDispatcher，必然也会存在广播的消息分发器，那就是 BroadcastingDispatcher，它被 PublishSubscribeChannel 这个消息通道所使用。广播消息分发器会把消息分发给所有的 MessageHandler

黄色部分涉及到各消息中间件的 Binder 实现以及 MQ 基本的订阅发布功能

RabbitMQ的基因中没有延时队列这回事，它不能直接指定一个队列类型为延时队列，然后去延时处理，但是经过上面两节的铺垫，我们可以将TTL+DLX相结合，这就能组成一个延时队列。

The management plugin is included in the RabbitMQ distribution. Like any other plugin, it must be enabled before it can be used. That's done using rabbitmq-plugins:

创建一个消费者，绑定消费队列及死信交换机，交换机默认为direct模型，死信交换机也是，arguments绑定死信交换机和key。（注解支持的具体参数文末会附上）

如果配置了死信队列，它将被重新publish到死信交换机，死信交换机将死信投递到一个队列上，这个队列就是死信队列

key = {"info","error","warning"}

集群消费

广播消费

它采用pull机制，而 不是一般MQ的push模型

消息队列既可以做管子，也可以当做池子

RabbitMQ很优秀，但RabbitMQ对消息堆积的支持并不好，当大量消息积压的时候，会导致 RabbitMQ 的性能急剧下降

搜索团队从kafka直接消费消息

但第二个问题却让我束手无措

都是可以根据 RoutingKey 把消息路由到不同的队列

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割