首页 文章 精选 留言 我的

精选列表

搜索[CEL策略],共10000篇文章
优秀的个人博客,低调大师

破解“记忆迷宫”:Agent长程上下文、向量检索与遗忘策略实战

2026年7月,AI Agent从单轮问答迈向长程任务协作,“记不住、找不准、忘不掉”成为复杂场景落地的核心瓶颈。NeurIPS最新研究显示,78%的Agent在多步推理中因上下文丢失导致任务失败率超40%。行业共识已从“无限堆Token”转向“结构化记忆工程”,通过分层存储、语义索引与主动遗忘机制,让Agent在海量信息中精准召回、高效推理。这标志Agent进入认知架构时代,可管理的记忆能力已成为决定其能否胜任专业级任务的关键分水岭。

优秀的个人博客,低调大师

每日一博 | Dubbo 负载均衡策略之一致性哈希

本文主要讲解了一致性哈希算法的原理以及其存在的数据倾斜的问题,然后引出解决数据倾斜问题的方法,最后分析一致性哈希算法在Dubbo中的使用。通过这篇文章,可以了解到一致性哈希算法的原理以及这种算法存在的问题和解决方案。 一、负载均衡 在这里引用dubbo官网的一段话—— LoadBalance 中文意思为负载均衡,它的职责是将网络请求,或者其他形式的负载“均摊”到不同的机器上。避免集群中部分服务器压力过大,而另一些服务器比较空闲的情况。通过负载均衡,可以让每台服务器获取到适合自己处理能力的负载。在为高负载服务器分流的同时,还可以避免资源浪费,一举两得。负载均衡可分为软件负载均衡和硬件负载均衡。在我们日常开发中,一般很难接触到硬件负载均衡。但软件负载均衡还是可以接触到的,比如 Nginx。在 Dubbo 中,也有负载均衡的概念和相应的实现。Dubbo 需要对服务消费者的调用请求进行分配,避免少数服务提供者负载过大。服务提供者负载过大,会导致部分请求超时。因此将负载均衡到每个服务提供者上,是非常必要的。Dubbo 提供了4种负载均衡实现,分别是基于权重随机算法的 RandomLoadBalance、基于最少活跃调用数算法的 LeastActiveLoadBalance、基于hash 一致性的 ConsistentHashLoadBalance,以及基于加权轮询算法的 RoundRobinLoadBalance。这几个负载均衡算法代码不是很长,但是想看懂也不是很容易,需要对这几个算法的原理有一定了解才行。 二、哈希算法 图1 无哈希算法请求 如上所示,假设0,1,2号服务器都存储的有用户信息,那么当我们需要获取某用户信息时,因为我们不知道该用户信息存放在哪一台服务器中,所以需要分别查询0,1,2号服务器。这样获取数据的效率是极低的。 对于这样的场景,我们可以引入哈希算法。 图2 引入哈希算法后的请求 还是上面的场景,但前提是每一台服务器存放用户信息时是根据某一种哈希算法存放的。所以取用户信息的时候,也按照同样的哈希算法取即可。 假设我们要查询用户号为100的用户信息,经过某个哈希算法,比如这里的userId mod n,即100 mod 3结果为1。所以用户号100的这个请求最终会被1号服务器接收并处理。 这样就解决了无效查询的问题。 但是这样的方案会带来什么问题呢? 扩容或者缩容时,会导致大量的数据迁移。最少也会影响50%的数据。 图3 增加一台服务器 为了说明问题,加入一台服务器3。服务器的数量n就从3变成了4。还是查询用户号为100的用户信息时,100 mod 4结果为0。这时,请求就被0号服务器接收了。 当服务器数量为3时,用户号为100的请求会被1号服务器处理。 当服务器数量为4时,用户号为100的请求会被0号服务器处理。 所以,当服务器数量增加或者减少时,一定会涉及到大量数据迁移的问题。 对于上述哈希算法其优点是简单易用,大多数分库分表规则就采取的这种方式。一般是提前根据数据量,预先估算好分区数。 其缺点是由于扩容或收缩节点导致节点数量变化时,节点的映射关系需要重新计算,会导致数据进行迁移。所以扩容时通常采用翻倍扩容,避免数据映射全部被打乱,导致全量迁移的情况,这样只会发生50%的数据迁移。 三、一致性哈希算法 **一致性 hash 算法由麻省理工学院的 Karger 及其合作者于1997年提出的,算法提出之初是用于大规模缓存系统的负载均衡。**它的工作过程是这样的,首先根据 ip 或者其他的信息为缓存节点生成一个 hash,并将这个 hash 投射到 [0, 232 - 1] 的圆环上。当有查询或写入请求时,则为缓存项的 key 生成一个 hash 值。然后查找第一个大于或等于该 hash 值的缓存节点,并到这个节点中查询或写入缓存项。如果当前节点挂了,则在下一次查询或写入缓存时,为缓存项查找另一个大于其 hash 值的缓存节点即可。大致效果如下图所示,每个缓存节点在圆环上占据一个位置。如果缓存项的 key 的 hash 值小于缓存节点 hash 值,则到该缓存节点中存储或读取缓存项。比如下面绿色点对应的缓存项将会被存储到 cache-2 节点中。由于 cache-3 挂了,原本应该存到该节点中的缓存项最终会存储到cache-4节点中。 图4 一致性哈希算法 在一致性哈希算法中,不管是增加节点,还是宕机节点,受影响的区间仅仅是增加或者宕机服务器在哈希环空间中,逆时针方向遇到的第一台服务器之间的区间,其它区间不会受到影响。 但是一致性哈希也是存在问题的: 图5 数据倾斜 当节点很少的时候可能会出现这样的分布情况,A服务会承担大部分请求。这种情况就叫做数据倾斜。 那如何解决数据倾斜的问题呢? 加入虚拟节点。 首先一个服务器根据需要可以有多个虚拟节点。假设一台服务器有n个虚拟节点。那么哈希计算时,可以使用IP+端口+编号的形式进行哈希值计算。其中的编号就是0到n的数字。由于IP+端口是一样的,所以这n个节点都是指向的同一台机器。 图6 引入虚拟节点 在没有加入虚拟节点之前,A服务器承担了绝大多数的请求。但是假设每个服务器有一个虚拟节点(A-1,B-1,C-1),经过哈希计算后落在了如上图所示的位置。那么A服务器的承担的请求就在一定程度上(图中标注了五角星的部分)分摊给了B-1、C-1虚拟节点,实际上就是分摊给了B、C服务器。 一致性哈希算法中,加入虚拟节点,可以解决数据倾斜问题。 四、一致性哈希在DUBBO 中的应用 图7 Dubbo中一致性哈希环 这里相同颜色的节点均属于同一个服务提供者,比如 Invoker1-1,Invoker1-2,……, Invoker1-160。这样做的目的是通过引入虚拟节点,让 Invoker 在圆环上分散开来,避免数据倾斜问题。所谓数据倾斜是指,由于节点不够分散,导致大量请求落到了同一个节点上,而其他节点只会接收到了少量请求的情况。比如: 图8 数据倾斜问题 如上,由于 Invoker-1 和 Invoker-2 在圆环上分布不均,导致系统中75%的请求都会落到 Invoker-1 上,只有 25% 的请求会落到 Invoker-2 上。解决这个问题办法是引入虚拟节点,通过虚拟节点均衡各个节点的请求量。 到这里背景知识就普及完了,接下来开始分析源码。我们先从 ConsistentHashLoadBalance 的 doSelect 方法开始看起,如下: public class ConsistentHashLoadBalance extends AbstractLoadBalance { private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>(); @Override protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { String methodName = RpcUtils.getMethodName(invocation); String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName; // 获取 invokers 原始的 hashcode int identityHashCode = System.identityHashCode(invokers); ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key); // 如果 invokers 是一个新的 List 对象,意味着服务提供者数量发生了变化,可能新增也可能减少了。 // 此时 selector.identityHashCode != identityHashCode 条件成立 if (selector == null || selector.identityHashCode != identityHashCode) { // 创建新的 ConsistentHashSelector selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode)); selector = (ConsistentHashSelector<T>) selectors.get(key); } // 调用 ConsistentHashSelector 的 select 方法选择 Invoker return selector.select(invocation); } private static final class ConsistentHashSelector<T> {...} } 如上,doSelect 方法主要做了一些前置工作,比如检测 invokers 列表是不是变动过,以及创建 ConsistentHashSelector。这些工作做完后,接下来开始调用 ConsistentHashSelector 的 select 方法执行负载均衡逻辑。在分析 select 方法之前,我们先来看一下一致性 hash 选择器 ConsistentHashSelector 的初始化过程,如下: private static final class ConsistentHashSelector<T> { // 使用 TreeMap 存储 Invoker 虚拟节点 private final TreeMap<Long, Invoker<T>> virtualInvokers; private final int replicaNumber; private final int identityHashCode; private final int[] argumentIndex; ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) { this.virtualInvokers = new TreeMap<Long, Invoker<T>>(); this.identityHashCode = identityHashCode; URL url = invokers.get(0).getUrl(); // 获取虚拟节点数,默认为160 this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160); // 获取参与 hash 计算的参数下标值,默认对第一个参数进行 hash 运算 String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0")); argumentIndex = new int[index.length]; for (int i = 0; i < index.length; i++) { argumentIndex[i] = Integer.parseInt(index[i]); } for (Invoker<T> invoker : invokers) { String address = invoker.getUrl().getAddress(); for (int i = 0; i < replicaNumber / 4; i++) { // 对 address + i 进行 md5 运算,得到一个长度为16的字节数组 byte[] digest = md5(address + i); // 对 digest 部分字节进行4次 hash 运算,得到四个不同的 long 型正整数 for (int h = 0; h < 4; h++) { // h = 0 时,取 digest 中下标为 0 ~ 3 的4个字节进行位运算 // h = 1 时,取 digest 中下标为 4 ~ 7 的4个字节进行位运算 // h = 2, h = 3 时过程同上 long m = hash(digest, h); // 将 hash 到 invoker 的映射关系存储到 virtualInvokers 中, // virtualInvokers 需要提供高效的查询操作,因此选用 TreeMap 作为存储结构 virtualInvokers.put(m, invoker); } } } } } ConsistentHashSelector 的构造方法执行了一系列的初始化逻辑,比如从配置中获取虚拟节点数以及参与 hash 计算的参数下标,默认情况下只使用第一个参数进行 hash。需要特别说明的是,ConsistentHashLoadBalance 的负载均衡逻辑只受参数值影响,具有相同参数值的请求将会被分配给同一个服务提供者。ConsistentHashLoadBalance 不 关系权重,因此使用时需要注意一下。 在获取虚拟节点数和参数下标配置后,接下来要做的事情是计算虚拟节点 hash 值,并将虚拟节点存储到 TreeMap 中。到此,ConsistentHashSelector 初始化工作就完成了。接下来,我们来看看 select 方法的逻辑。 public Invoker<T> select(Invocation invocation) { // 将参数转为 key String key = toKey(invocation.getArguments()); // 对参数 key 进行 md5 运算 byte[] digest = md5(key); // 取 digest 数组的前四个字节进行 hash 运算,再将 hash 值传给 selectForKey 方法, // 寻找合适的 Invoker return selectForKey(hash(digest, 0)); } private Invoker<T> selectForKey(long hash) { // 到 TreeMap 中查找第一个节点值大于或等于当前 hash 的 Invoker Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry(); // 如果 hash 大于 Invoker 在圆环上最大的位置,此时 entry = null, // 需要将 TreeMap 的头节点赋值给 entry if (entry == null) { entry = virtualInvokers.firstEntry(); } // 返回 Invoker return entry.getValue(); } 如上,选择的过程相对比较简单了。首先是对参数进行 md5 以及 hash 运算,得到一个 hash 值。然后再拿这个值到 TreeMap 中查找目标 Invoker 即可。 到此关于 ConsistentHashLoadBalance 就分析完了。 在阅读ConsistentHashLoadBalance 源码之前,建议读者先补充背景知识,不然看懂代码逻辑会有很大难度。 五、应用场景 DNS负载均衡最早的负载均衡技术是通过DNS来实现的,在DNS中为多个地址配置同一个名字,因而查询这个名字的客户机将得到其中一个地址,从而使得不同的客户访问不同的服务器,达到负载均衡的目的。DNS负载均衡是一种简单而有效的方法,但是它不能区分服务器的差异,也不能反映服务器的当前运行状态。 代理服务器负载均衡使用代理服务器,可以将请求转发给内部的服务器,使用这种加速模式显然可以提升静态网页的访问速度。然而,也可以考虑这样一种技术,使用代理服务器将请求均匀转发给多台服务器,从而达到负载均衡的目的。 地址转换网关负载均衡支持负载均衡的地址转换网关,可以将一个外部IP地址映射为多个内部IP地址,对每次TCP连接请求动态使用其中一个内部地址,达到负载均衡的目的。 协议内部支持负载均衡除了这三种负载均衡方式之外,有的协议内部支持与负载均衡相关的功能,例如HTTP协议中的重定向能力等,HTTP运行于TCP连接的最高层。 NAT负载均衡NAT(Network Address Translation网络地址转换)简单地说就是将一个IP地址转换为另一个IP地址,一般用于未经注册的内部地址与合法的、已获注册的Internet IP地址间进行转换。适用于解决Internet IP地址紧张、不想让网络外部知道内部网络结构等的场合下。 反向代理负载均衡普通代理方式是代理内部网络用户访问internet上服务器的连接请求,客户端必须指定代理服务器,并将本来要直接发送到internet上服务器的连接请求发送给代理服务器处理。反向代理(Reverse Proxy)方式是指以代理服务器来接受internet上的连接请求,然后将请求转发给内部网络上的服务器,并将从服务器上得到的结果返回给internet上请求连接的客户端,此时代理服务器对外就表现为一个服务器。反向代理负载均衡技术是把将来自internet上的连接请求以反向代理的方式动态地转发给内部网络上的多台服务器进行处理,从而达到负载均衡的目的。 混合型负载均衡在有些大型网络,由于多个服务器群内硬件设备、各自的规模、提供的服务等的差异,可以考虑给每个服务器群采用最合适的负载均衡方式,然后又在这多个服务器群间再一次负载均衡或群集起来以一个整体向外界提供服务(即把这多个服务器群当做一个新的服务器群),从而达到最佳的性能。将这种方式称之为混合型负载均衡。此种方式有时也用于单台均衡设备的性能不能满足大量连接请求的情况下。 作者:京东物流 乔杰 来源:京东云开发者社区

优秀的个人博客,低调大师

四种策略确保 RabbitMQ 消息发送可靠性!你用哪种?

@[toc] 微服务可以设计成消息驱动的微服务,响应式系统也可以基于消息中间件来做,从这个角度来说,在互联网应用开发中,消息中间件真的是太重要了。 今天,以 RabbitMQ 为例,松哥来和大家聊一聊消息中间消息发送可靠性的问题。 注意,以下内容我主要和大家讨论如何确保消息生产者将消息发送成功,并不涉及消息消费的问题。 1. RabbitMQ 消息发送机制 大家知道,RabbitMQ 中的消息发送引入了 Exchange(交换机)的概念,消息的发送首先到达交换机上,然后再根据既定的路由规则,由交换机将消息路由到不同的 Queue(队列)中,再由不同的消费者去消费。 大致的流程就是这样,所以要确保消息发送的可靠性,主要从两方面去确认: 消息成功到达 Exchange 消息成功到达 Queue 如果能确认这两步,那么我们就可以认为消息发送成功了。 如果这两步中任一步骤出现问题,那么消息就没有成功送达,此时我们可能要通过重试等方式去重新发送消息,多次重试之后,如果消息还是不能到达,则可能就需要人工介入了。 经过上面的分析,我们可以确认,要确保消息成功发送,我们只需要做好三件事就可以了: 确认消息到达 Exchange。 确认消息到达 Queue。 开启定时任务,定时投递那些发送失败的消息。 2. RabbitMQ 的努力 上面提出的三个步骤,第三步需要我们自己实现,前两步 RabbitMQ 则有现成的解决方案。 如何确保消息成功到达 RabbitMQ?RabbitMQ 给出了两种方案: 开启事务机制 发送方确认机制 这是两种不同的方案,不可以同时开启,只能选择其中之一,如果两者同时开启,则会报如下错误: 我们分别来看。以下所有案例都在 Spring Boot 中展开,文末可以下载相关源码。 2.1 开启事务机制 Spring Boot 中开启 RabbitMQ 事务机制的方式如下: 首先需要先提供一个事务管理器,如下: @Bean RabbitTransactionManager transactionManager(ConnectionFactory connectionFactory) { return new RabbitTransactionManager(connectionFactory); } 接下来,在消息生产者上面做两件事:添加事务注解并设置通信信道为事务模式: @Service public class MsgService { @Autowired RabbitTemplate rabbitTemplate; @Transactional public void send() { rabbitTemplate.setChannelTransacted(true); rabbitTemplate.convertAndSend(RabbitConfig.JAVABOY_EXCHANGE_NAME,RabbitConfig.JAVABOY_QUEUE_NAME,"hello rabbitmq!".getBytes()); int i = 1 / 0; } } 这里注意两点: 发送消息的方法上添加 @Transactional 注解标记事务。 调用 setChannelTransacted 方法设置为 true 开启事务模式。 这就 OK 了。 在上面的案例中,我们在结尾来了个 1/0 ,这在运行时必然抛出异常,我们可以尝试运行该方法,发现消息并未发送成功。 当我们开启事务模式之后,RabbitMQ 生产者发送消息会多出四个步骤: 客户端发出请求,将信道设置为事务模式。 服务端给出回复,同意将信道设置为事务模式。 客户端发送消息。 客户端提交事务。 服务端给出响应,确认事务提交。 上面的步骤,除了第三步是本来就有的,其他几个步骤都是平白无故多出来的。所以大家看到,事务模式其实效率有点低,这并非一个最佳解决方案。我们可以想想,什么项目会用到消息中间件?一般来说都是一些高并发的项目,这个时候并发性能尤为重要。 所以,RabbitMQ 还提供了发送方确认机制(publisher confirm)来确保消息发送成功,这种方式,性能要远远高于事务模式,一起来看下。 2.2 发送方确认机制 2.2.1 单条消息处理 首先我们移除刚刚关于事务的代码,然后在 application.properties 中配置开启消息发送方确认机制,如下: spring.rabbitmq.publisher-confirm-type=correlated spring.rabbitmq.publisher-returns=true 第一行是配置消息到达交换器的确认回调,第二行则是配置消息到达队列的回调。 第一行属性的配置有三个取值: none:表示禁用发布确认模式,默认即此。 correlated:表示成功发布消息到交换器后会触发的回调方法。 simple:类似 correlated,并且支持 waitForConfirms() 和 waitForConfirmsOrDie() 方法的调用。 接下来我们要开启两个监听,具体配置如下: @Configuration public class RabbitConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback { public static final String JAVABOY_EXCHANGE_NAME = "javaboy_exchange_name"; public static final String JAVABOY_QUEUE_NAME = "javaboy_queue_name"; private static final Logger logger = LoggerFactory.getLogger(RabbitConfig.class); @Autowired RabbitTemplate rabbitTemplate; @Bean Queue queue() { return new Queue(JAVABOY_QUEUE_NAME); } @Bean DirectExchange directExchange() { return new DirectExchange(JAVABOY_EXCHANGE_NAME); } @Bean Binding binding() { return BindingBuilder.bind(queue()) .to(directExchange()) .with(JAVABOY_QUEUE_NAME); } @PostConstruct public void initRabbitTemplate() { rabbitTemplate.setConfirmCallback(this); rabbitTemplate.setReturnsCallback(this); } @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { if (ack) { logger.info("{}:消息成功到达交换器",correlationData.getId()); }else{ logger.error("{}:消息发送失败", correlationData.getId()); } } @Override public void returnedMessage(ReturnedMessage returned) { logger.error("{}:消息未成功路由到队列",returned.getMessage().getMessageProperties().getMessageId()); } } 关于这个配置类,我说如下几点: 定义配置类,实现 RabbitTemplate.ConfirmCallback 和 RabbitTemplate.ReturnsCallback 两个接口,这两个接口,前者的回调用来确定消息到达交换器,后者则会在消息路由到队列失败时被调用。 定义 initRabbitTemplate 方法并添加 @PostConstruct 注解,在该方法中为 rabbitTemplate 分别配置这两个 Callback。 这就可以了。 接下来我们对消息发送进行测试。 首先我们尝试将消息发送到一个不存在的交换机中,像下面这样: rabbitTemplate.convertAndSend("RabbitConfig.JAVABOY_EXCHANGE_NAME",RabbitConfig.JAVABOY_QUEUE_NAME,"hello rabbitmq!".getBytes(),new CorrelationData(UUID.randomUUID().toString())); 注意第一个参数是一个字符串,不是变量,这个交换器并不存在,此时控制台会报如下错误: 接下来我们给定一个真实存在的交换器,但是给一个不存在的队列,像下面这样: rabbitTemplate.convertAndSend(RabbitConfig.JAVABOY_EXCHANGE_NAME,"RabbitConfig.JAVABOY_QUEUE_NAME","hello rabbitmq!".getBytes(),new CorrelationData(UUID.randomUUID().toString())); 注意此时第二个参数是一个字符串,不是变量。 可以看到,消息虽然成功达到交换器了,但是没有成功路由到队列(因为队列不存在)。 这是一条消息的发送,我们再来看看消息的批量发送。 2.2.2 消息批量处理 如果是消息批量处理,那么发送成功的回调监听是一样的,这里不再赘述。 这就是 publisher-confirm 模式。 相比于事务,这种模式下的消息吞吐量会得到极大的提升。 3. 失败重试 失败重试分两种情况,一种是压根没找到 MQ 导致的失败重试,另一种是找到 MQ 了,但是消息发送失败了。 两种重试我们分别来看。 3.1 自带重试机制 前面所说的事务机制和发送方确认机制,都是发送方确认消息发送成功的办法。如果发送方一开始就连不上 MQ,那么 Spring Boot 中也有相应的重试机制,但是这个重试机制就和 MQ 本身没有关系了,这是利用 Spring 中的 retry 机制来完成的,具体配置如下: spring.rabbitmq.template.retry.enabled=true spring.rabbitmq.template.retry.initial-interval=1000ms spring.rabbitmq.template.retry.max-attempts=10 spring.rabbitmq.template.retry.max-interval=10000ms spring.rabbitmq.template.retry.multiplier=2 从上往下配置含义依次是: 开启重试机制。 重试起始间隔时间。 最大重试次数。 最大重试间隔时间。 间隔时间乘数。(这里配置间隔时间乘数为 2,则第一次间隔时间 1 秒,第二次重试间隔时间 2 秒,第三次 4 秒,以此类推) 配置完成后,再次启动 Spring Boot 项目,然后关掉 MQ,此时尝试发送消息,就会发送失败,进而导致自动重试。 3.2 业务重试 业务重试主要是针对消息没有到达交换器的情况。 如果消息没有成功到达交换器,根据我们第二小节的讲解,此时就会触发消息发送失败回调,在这个回调中,我们就可以做文章了! 整体思路是这样: 首先创建一张表,用来记录发送到中间件上的消息,像下面这样: 每次发送消息的时候,就往数据库中添加一条记录。这里的字段都很好理解,有三个我额外说下: status:表示消息的状态,有三个取值,0,1,2 分别表示消息发送中、消息发送成功以及消息发送失败。 tryTime:表示消息的第一次重试时间(消息发出去之后,在 tryTime 这个时间点还未显示发送成功,此时就可以开始重试了)。 count:表示消息重试次数。 其他字段都很好理解,我就不一一啰嗦了。 在消息发送的时候,我们就往该表中保存一条消息发送记录,并设置状态 status 为 0,tryTime 为 1 分钟之后。 在 confirm 回调方法中,如果收到消息发送成功的回调,就将该条消息的 status 设置为1(在消息发送时为消息设置 msgId,在消息发送成功回调时,通过 msgId 来唯一锁定该条消息)。 另外开启一个定时任务,定时任务每隔 10s 就去数据库中捞一次消息,专门去捞那些 status 为 0 并且已经过了 tryTime 时间记录,把这些消息拎出来后,首先判断其重试次数是否已超过 3 次,如果超过 3 次,则修改该条消息的 status 为 2,表示这条消息发送失败,并且不再重试。对于重试次数没有超过 3 次的记录,则重新去发送消息,并且为其 count 的值+1。 大致的思路就是上面这样,松哥这里就不给出代码了,松哥的 vhr 里边邮件发送就是这样的思路来处理的,完整代码大家可以参考 vhr 项目(https://github.com/lenve/vhr)。 当然这种思路有两个弊端: 去数据库走一遭,可能拖慢 MQ 的 Qos,不过有的时候我们并不需要 MQ 有很高的 Qos,所以这个应用时要看具体情况。 按照上面的思路,可能会出现同一条消息重复发送的情况,不过这都不是事,我们在消息消费时,解决好幂等性问题就行了。 当然,大家也要注意,消息是否要确保 100% 发送成功,也要看具体情况。 4. 小结 好啦,这就是关于消息生产者的一些常见问题以及对应的解决方案,下篇文章松哥和大家探讨如果保证消息消费成功并解决幂等性问题。 本文涉及到的相关源代码大家可以在这里下载:https://github.com/lenve/javaboy-code-samples。

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

用户登录
用户注册