前言 我们知道，消息从发送到签收的整个过程是 Producer-->Broker/Exchange-->Broker/Queue-->Consumer，因此如果只是要保证消息的可靠投递，我们需要考虑的仅是前两个阶段...，因为消息只要成功到达队列，就算投递成功。...比如投递消息时指定的Exchange不存在，那么阶段一就会失败 如果投递到Exchange成功，但是指定的路由件错误或者别的原因，消息没有从Exchange到达Queue，那就是第二阶段出错。...而从生产者和消费者角度来看，消息成功投递到队列才算成功投递，因此阶段一和阶段而都属于生产者一方需要关注，阶段三属于消费者一方，这里只考虑消息的成功投递，因此不考虑消费者的签收部分。...在多线程环境中，不能保证下一个操作使用相同的通道。但是，有时您可能希望更好地控制通道的使用，并确保在同一通道上执行全部操作。
前言 我们知道，消息从发送到签收的整个过程是 Producer-->Broker/Exchange-->Broker/Queue-->Consumer，因此如果只是要保证消息的可靠投递，我们需要考虑的仅是前两个阶段...，因为消息只要成功到达队列，就算投递成功。...比如投递消息时指定的Exchange不存在，那么阶段一就会失败 如果投递到Exchange成功，但是指定的路由件错误或者别的原因，消息没有从Exchange到达Queue，那就是第二阶段出错。...[在这里插入图片描述] 而从生产者和消费者角度来看，消息成功投递到队列才算成功投递，因此阶段一和阶段而都属于生产者一方需要关注，阶段三属于消费者一方，这里只考虑消息的成功投递，因此不考虑消费者的签收部分...在多线程环境中，不能保证下一个操作使用相同的通道。但是，有时您可能希望更好地控制通道的使用，并确保在同一通道上执行全部操作。
生产端的可靠性投递 保障消息的成功发出 保障MQ节点的成功接收 发送端收到MQ节点确认应答 完善的消息进行补偿机制 解决方案 消息落库，对消息状态进行打标 消息的延迟投递，做二次确认，回调检查 消息落库架构图...比如消息入库的时候需要带上时间，我们取出状态为0的消息形成一个集合，然后过滤该集合的时间为2分钟以上的消息进行重新发送。...由于我们的业务入库是必须的，所以我们要考虑消息入库是否可以取消。所以我们要考虑第二种方式：消息的延迟投递，做二次确认，回调检查。 ?...RPC或者Restful重新调用upstream中的业务库检查后，再重新发送一条step1消息，走下一个流程。...以上的这个方案并不能保证100%的消息投递，但是它对于第一种方案的好处是少进行了一次DB操作，保证高并发的性能。
image.png 「消息投递时 可能发生丢失的场景：」 生产者------msg------> MQ 。可开启消息投递结果回调，确保每条消息都收到了回调。 MQ。...将Queue与消息设置成可持久化，搭建镜像集群队列。 MQ-------callback---->生产者。回调时失败，某条消息在一段时间内未收到回调，则默认投递失败，生产者需要再次投递该消息到MQ。...private String logPath; private int logRetentionDays; } } 定时扫描任务编写 /** * 扫描数据库中需要重新投递的消息并重新投递...if (PAGE_SIZE == messages.size()) { service(++pageNum); } } /** * 重新投递消息...开启任务调度再次进行投递(投递次数+1，且更新下次投递时间) 当投递次数达到最大投递次数，下一次，将消息设置成投递失败 调度日志 image.png # Next 消息可靠消费 消费端限流保护 死信队列
RabbitMQ消息可靠性投递 什么是消息的可靠性投递？即保证消息百分百发送到消息队列中去，消息发送端需要接受到mq服务端接受到消息的确认应答。...除此之外还应有完善的消息补偿机制，发送失败的消息可以再感知并二次处理。...新版依赖可靠性投递默认是关闭的，使用以下方法开启： #旧版，确认消息发送成功，通过实现ConfirmCallBack接口，消息发送到交换器Exchange后触发回调 spring.rabbitmq.publisher-confirms...=true #新版 第二步 修改交换机投递到队列失败的策略 # 为true,则交换机处理消息到路由失败，则会返回给生产者 spring.rabbitmq.template.mandatory=true...returnCallback"); // 为true,则交换机处理消息到路由失败，则会返回给生产者，开启强制消息投递（mandatory为设置为true），但消息未被路由至任何一个queue
本文的上篇《IM消息机制（一）：保证在线实时消息的可靠投递》中，我们讨论了在线实时消息的投递可以通过应用层的确认、发送方的超时重传、接收方的去重等手段来保证业务层面消息的不丢不重。...但实时在线投递针对的是消息收发双方都在线的情况（如当发送方用户A发送消息给接收方用户B时，用户B是在线的），那如果消息的接收方用户B不在线，系统是如何保证消息的可达性的呢？这就是本文要讨论的问题。...二、典型离线消息表的设计以及拉取离线消息的过程 ① 存储离线消看书的表主要字段大致如下： -- 消息接收者ID receiver_uid varchar(50), -- 消息的唯一指纹码（即消息ID...五、优化离线消息的拉取过程，保证离线消息不会丢失 如何保证可达性，上述步骤第三步执行完毕之后，第四个步骤离线消息返回给客户端过程中，服务器挂点，路由器丢消息，或者客户端crash了，那离线消息岂不是丢了么...如同在线消息的应用层ACK机制一样，离线消息拉时，不能够直接删除数据库中的离线消息，而必须等应用层的离线消息ACK（说明用户B真的收到离线消息了），才能删除数据库中的离线消息。
[image.png] 消息投递时 可能发生丢失的场景： 生产者------msg------> MQ 。可开启消息投递结果回调，确保每条消息都收到了回调。 MQ。...将Queue与消息设置成可持久化，搭建镜像集群队列。 MQ-------callback---->生产者。回调时失败，某条消息在一段时间内未收到回调，则默认投递失败，生产者需要再次投递该消息到MQ。...，可以用来保证消息100%投递到rabbitMQ。...private String logPath; private int logRetentionDays; } } 定时扫描任务编写 /** * 扫描数据库中需要重新投递的消息并重新投递...if (PAGE_SIZE == messages.size()) { service(++pageNum); } } /** * 重新投递消息
RabbitMQ消息的可靠性主要包括两方面，一方面是通过实现消费的重试机制（通过@Retryable来实现重试，可以设置重试次数和重试频率，但是要保证幂等性），另一方面就是实现消息生产者的可靠投递...（注意消费单幂等），下面主要讲下生产者实现的可靠消息投递。...代码类似这样： 创建channel send message wait for ack(or 超时) close channel 返回成功or失败 同样的，由于每次发送message都要重新建立连接...（相比于丢失消息，重发消息要好解决的多，我们可以在consumer端做到幂等）。...Boolean send = producr.send(messageWithTime); logger.info("进行重新投递消息
在开发业务系统时，某些业务场景需要消息定时发送或延时发送（类似：飞信的短信定时发送需求），这时候就需要用到activemq的消息延时投递，详细的文档可参考官网说明，本文只介绍二种常用的用法： 注：本文采用...spring的JmsTemplate来发送消息 步骤1、首先要修改activemq.xml配置文件，启用延时投递 1 <broker xmlns="http://activemq.apache.org/...org.apache.commons.lang3.StringUtils; import org.springframework.jms.core.MessagePostProcessor; /** * MQ延时<em>投递</em>处理器...注：在开启<em>消息</em>持久化存储的前提下，就算把相应的queue在webconsole面板中删除（即删除队列），只要<em>投递</em>的时间尚未到，该<em>消息</em>也不会删除，仍然能正常延时<em>投递</em>。...此外，在queues面板中，如何查看某条具体的<em>消息</em>，也可以通过属性发现这条<em>消息</em>是延时<em>消息</em>，参考下图： ?
消息队列的八股文最喜欢问这个。消息可靠性投递，这个对于消息队列非常重要，很多时候，我们不是人为的去干预，但是仍然出现其他意外的事情。...导致消息在传递过程中可能会面临丢失、重复、损坏等问题，这就要求我们必须重视消息的可靠性投递。 什么是消息可靠性投递？...，当目标交换机投递失败时，把消息投递至备份交换机 故障情况2 故障情况2：消息成功存入消息队列，但是消息队列服务器宕机了。...创建配置类 在这里我们为什么要创建这个配置类呢？...使用@PostConstruct注解的方法必须满足以下条件： 方法不能有任何参数。 方法必须是非静态的。 方法不能返回任何值。
重试机制：自动重试：在消费者端，可以通过使用basic.recover()方法进行消息的自动重试。当该方法被调用时，RabbitMQ将重新投递消息，直到投递成功或者消息被拒绝。...延迟队列方式：RabbitMQ还支持通过使用延迟队列（dead-letter queue）实现消息的重试。在这种方式中，当消息一次投递失败后，消息将被重新投递到延迟队列中。...延迟队列的作用是将消息保留一段时间，然后再将其重新投递到原队列中进行处理。confirm机制和return机制：confirm机制：用于确保消息从生产者到交换机的过程中被正确处理。...如果消息未能成功到达交换机，生产者将收到确认失败的通知，并可以选择重新发送消息。return机制：用于确保消息从交换机到队列的过程中被正确处理。...如果消息在路由过程中出现问题（如找不到匹配的队列），RabbitMQ将向生产者发送一个return通知，其中包含有关失败原因的信息。生产者可以根据这些信息选择重新发送消息或执行其他操作。
一、什么是延迟消息投递 延迟消息投递在MQ应用场景中十分普遍，它是指消息在发送到 MQ 服务端后并不会立马投递，而是根据消息中的属性延迟固定时间后才投递给消费者，一般分为定时消息和延迟消息两种： 定时消息...延迟消息：Producer 将消息发送到 MQ 服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到 Consumer 进行消费。...二、延迟消息投递的使用场景 延迟消息投递是要暂缓对当前消息的处理，在未来的某个时间点再触发投递，实际的应用场景非常多，比如异常检测重试、订单超时取消、预约提醒等。...三、如何使用Pulsar延迟消息投递 Pulsar 最早是在 2.4.0 引入了延迟消息投递的特性，在 Pulsar 中使用延迟消息，可以精确指定延迟投递的时间，有 deliverAfter 和 deliverAt...六、Pulsar延迟消息投递未来工作 Pulsar 目前的延迟消息投递方案简单高效，但处理大规模延迟消息时仍然存在风险。关于延迟消息投递，社区和数据平台部 MQ 团队下一步将聚焦在支持大规模延迟消息。
，结果服务器宕机了 投递失败：RabbitMQ把这个消息投递给对应的消费者了，但是消费者宕机了，导致这条消息没能正常消费。...那么对于这三种情况，我们分别要处理的问题也就是以下三个 生产者保证消息可靠投递 RabbitMQ持久化 消费者保证消息可靠消费 我们一个一个来解决 生产者保证消息可靠投递 为了保证消息被正确投递到消息中间件...，RabbitMQ提供了如下两个配置来保证消息投递的可靠性。...如果设置了Mandatory属性则当消息不能被正确路由到队列中去时将会触发Return Method，这样我们可以在Return Method中进行相关业务处理，如果Mandatory没有设置则当消息不能正确路由到队列中去的时候...域，表示到这个序列号之前的所有消息都已经得到了处理(multiple如果为true则表示小于等于deliveryTag的消息都被投递成功，如果为false则表示只有等于deliveryTag的消息已经被投递成功
如果发送到A消费者的消息一直不确认，只有等到A消费者与rabbitmq的连接中断，rabbitmq才会考虑将A消费者未确认的消息重新投递给另一个消费者 Spring Boot中针对消息ack的方式 有三种方式...所以要保证消息的可靠投递，只需要保证这3个阶段的可靠投递即可 生产阶段 这个阶段的可靠投递主要靠ConfirmListener（发布者确认）和ReturnListener（失败通知） 前面已经介绍过了...id，并且保存消息id和消息的对应关系，因为RabbitTemplate.ConfirmCallback只能拿到消息id，并不能拿到消息内容，所以需要我们自己保存这种映射关系。...前文已经介绍了原生api ack的方式和Spring Boot框架ack的方式 总而言之，在生产环境中，我们一般都是单条手动ack，消费失败后不会重新入队（因为很大概率还会再次失败），而是将消息重新投递到死信队列...，消息会被直接丢弃，如果指定了死信队列的话，会被投递到死信队列
《webim如何保证消息的可靠投递》 上一章和大家分享了webim消息的实时性问题 消息的可靠性，即消息的不丢失和不重复，也是im系统中的一个难点。...只能说明im-server成功接收到了消息，并不能说明client-B接收到了消息。...会发现，一条消息的发送，分别包含（上）（下）两个半场，即msg的R/A/N三个报文，ack的R/A/N三个报文，一个应用层即时通讯消息的可靠投递，共涉及6个报文，这就是im系统中消息投递的最核心技术（如果某个...五、可靠消息投递存在什么问题 期望六个报文完成消息的可靠投递，但实际情况下： 1）msg:R，msg:A报文可能丢失，此时直接提示“发送失败”即可，问题不大 2）msg:N，ack:R，ack:A，ack...:N这四个报文都可能丢失（原因如第二章所述，可能是服务器奔溃、网络抖动、或者客户端奔溃），此时client-A都收不到期待的ack:N报文，即client-A不能确认client-B是否收到“你好”，那怎么办呢
RabbitMQ消息的可靠性主要包括两方面，一方面是通过实现消费的重试机制（通过@Retryable来实现重试，可以设置重试次数和重试频率，但是要保证幂等性），另一方面就是实现消息生产者的可靠投递...（注意消费单幂等），下面主要讲下生产者实现的可靠消息投递。 ...代码类似这样： 创建channel send message wait for ack(or 超时) close channel 返回成功or失败 同样的，由于每次发送message都要重新建立连接...（相比于丢失消息，重发消息要好解决的多，我们可以在consumer端做到幂等）。...Boolean send = producr.send(messageWithTime); logger.info("进行重新投递消息
，在很多时候也很难控制网络通信的质量，这也就是为什么『网络是稳定、可信赖的』分布式系统中常见的谬论之一。...message-delivery-problems 网络请求由于超时的问题，消息的发送者只能通过重试的方式对消息进行重发，但是这就可能会导致消息的重复发送与处理，然而如果超时后不重新发送消息也可能导致消息的丢失...at-most-once 无论该请求是否发送给了接受者，发送者都不会重新发送这条消息；这其实就是最最基本的消息投递语义，然而消息可能由于网络或者节点的故障出现丢失。...最少一次 为了解决最多一次时的消息丢失问题，消息的发送者需要在网络出现超时重新发送相同的消息，也就是引入超时重试的机制，在发送者发出消息会监听消息的响应，如果超过了一定时间也没有得到响应就会重新发送该消息...序列号 使用序列号保证投递顺序的方式其实与 TCP 协议中使用的 SEQ 非常相似，因为网络并不能保证所有数据包传输的顺序并且每个栈帧的传输大小有限，所以 TCP 协议在发送数据包时加入 SEQ，接受方可以通过
，发送方client-A收到msg:A后，只能说明im-server成功接收到了消息，并不能说明client-B接收到了消息。...一个应用层即时通讯消息的可靠投递，共涉及6个报文，这就是im系统中消息投递的最核心技术（如果某个im系统不包含这6个报文，不要谈什么消息的可靠性）。...六、可靠消息投递存在什么问题 期望六个报文完成消息的可靠投递，但实际情况下： msg:R，msg:A 报文可能丢失： 此时直接提示“发送失败”即可，问题不大 msg:N，ack:R，ack:A，ack:...N这四个报文都可能丢失： （原因如1.4所述，可能是服务器奔溃、网络抖动、或者客户端奔溃），此时client-A都收不到期待的ack:N报文，即client-A不能确认client-B是否收到“你好”...ack，群消息就没有这么简单了，群消息存在一个扩散系数，im群消息的可靠投递问题感兴趣的可查阅相关资料。
介绍 要想保证消息的可靠型投递，无非保证如下3个阶段的正常执行即可。...生产者将消息成功投递到broker broker将投递过程的消息持久化下来 消费者能从broker消费到消息 发送端消息重试 producer向broker发送消息后，没有收到broker的ack时，rocketmq...重试队列和死信队列都是按照Consumer Group划分的」 重试队列topic名字：%RETRY% + consumerGroup 死信队列topic名字：%DLQ% + consumerGroup 「为什么重试队列和死信队列要按照...中 将消息顺序写到CommitLog中 将消息对应的信息分发到对应的ConsumerQueue中（topic为SCHEDULE_TOPIC_XXXX总共有18个queue，对应18个延迟级别） 定时任务不断判断消息是否到达投递时间...，没有到达则后续执行投递 如果到达投递时间，则从commitLog中拉取消息的内容，重新设置消息topic，queueId为原来的（原来的topic，queueId在消息扩展属性中），然后将消息投递到commitLog
然而网络在计算机的世界中是最不可控的，如果我们通过网络请求调用其他服务的接口，可能就会由于种种原因没有将消息送达至目标的服务，对于当前服务我们并不能控制网络的传输，在很多时候也很难控制网络通信的质量，这也就是为什么...每一次确定的响应都需要这次请求在一个往返以及被调用节点中正确处理，流量既不能被中间代理丢包，也不能由于目标节点的错误导致无法发出响应，只有在同时满足了这两个条件的情况下，我们才能得到确定的响应结果。...络请求由于超时的问题，消息的发送者只能通过重试的方式对消息进行重发，但是这就可能会导致消息的重复发送与处理，然而如果超时后不重新发送消息也可能导致消息的丢失，所以如何在不可靠的通信方式中，保证消息不重不漏是非常关键的...无论该请求是否发送给了接受者，发送者都不会重新发送这条消息；这其实就是最最基本的消息投递语义，然而消息可能由于网络或者节点的故障出现丢失。...最少一次 为了解决最多一次时的消息丢失问题，消息的发送者需要在网络出现超时重新发送相同的消息，也就是引入超时重试的机制，在发送者发出消息会监听消息的响应，如果超过了一定时间也没有得到响应就会重新发送该消息
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储
