RocketMQ proxy caffeine缓存

TopicRouteService的 Caffeine 缓存配置在构造函数里一次性构建刷新间隔默认20 秒刷新失败会降级继续用旧数据仅首次加载失败才报错。下面展开。一、缓存配置默认值[TopicRouteService 构造函数]this.topicCacheCaffeine.newBuilder().maximumSize(config.getTopicRouteServiceCacheMaxNum())// 20000 条.expireAfterAccess(config.getTopicRouteServiceCacheExpiredSeconds(),TimeUnit.SECONDS)// 300s.refreshAfterWrite(config.getTopicRouteServiceCacheRefreshSeconds(),TimeUnit.SECONDS)// 20s.executor(cacheRefreshExecutor)// 异步刷新线程池.build(newCacheLoaderString,MessageQueueView(){...});对应 [ProxyConfig 的默认值]配置项默认值含义topicRouteServiceCacheMaxNum20000最大缓存条目数LRU 淘汰topicRouteServiceCacheExpiredSeconds3005 分钟访问过期超过该时间未被访问才被移除topicRouteServiceCacheRefreshSeconds2020 秒写入后刷新超过该时间后下次访问触发异步刷新topicRouteServiceThreadPoolNumsPROCESSOR_NUMBER异步刷新线程池大小topicRouteServiceThreadPoolQueueCapacity5000刷新线程池队列容量二、刷新间隔是多少、如何触发刷新间隔是 20 秒refreshAfterWrite(20s)但要注意它是惰性刷新不是后台定时器条目写入缓存后超过 20 秒当有请求访问该 topic时才触发一次异步刷新reload刷新由 [cacheRefreshExecutor]线程池执行刷新完成前调用者立即拿到旧值不会被阻塞。和expireAfterAccess(300s)的区别refreshAfterWrite管「多久刷新一次」expireAfterAccess管「多久不访问就删除」。三、刷新失败怎么处理继续用旧数据关键在 [CacheLoader.reload]OverridepublicMessageQueueViewreload(Stringkey,MessageQueueViewoldValue)throwsException{try{returnload(key);// 刷新成功 → 返回新路由}catch(Exceptione){log.warn(String.format(reload topic route from namesrv. topic: %s,key),e);returnoldValue;// 刷新失败 → 返回旧值继续用}}所以刷新失败如 Namesrv 抖动、网络超时会静默降级返回oldValue继续用旧路由只打一条 warn 日志不抛异常、不影响业务。四、但「首次加载」失败是会报错的注意区分reload和load。首次访问缓存 miss走 [load]publicMessageQueueViewload(Stringtopic)throwsException{try{TopicRouteDatadata...getTopicRouteInfoFromNameServer(topic,3s);returnbuildMessageQueueView(topic,data);}catch(Exceptione){if(TopicRouteHelper.isTopicNotExistError(e)){returnMessageQueueView.WRAPPED_EMPTY_QUEUE;// topic 不存在 → 返回空队列}throwe;// 其他异常 → 向上抛}}topic 不存在→ 返回WRAPPED_EMPTY_QUEUE随后 [getCacheMessageQueueWrapper]检测到空队列抛MQClientException(TOPIC_NOT_EXIST)。Namesrv 不可达/超时等其它异常→load直接throw e首次访问就会报错因为没有旧值可降级。总结场景行为首次加载 topic 不存在返回空队列 → 上层抛TOPIC_NOT_EXIST首次加载 Namesrv 异常load抛异常直接报错无旧值可降级刷新20s 后异步 reload成功返回新路由刷新失败return oldValue继续用旧数据仅 warn 日志一句话默认「写入后 20 秒、访问时惰性异步刷新」刷新失败降级用旧路由不报错只有首次加载失败无缓存可兜底才会抛异常。线程池满如何处理结论刷新线程池满了之后既不阻塞、也不报错而是「丢弃队列里最旧的等待任务」让最新的刷新任务入队。这取决于线程池的拒绝策略。一、线程池是怎么建的[TopicRouteService 构造] 里通过 [ThreadPoolMonitor.createAndMonitor] 创建ThreadUtils.newThreadPoolExecutor(PROCESSOR_NUMBER,// corePoolSizePROCESSOR_NUMBER,// maximumPoolSize core线程数固定60s,// keepAliveTimenewLinkedBlockingQueue(5000),// 有界队列容量 5000threadFactory,newThreadPoolExecutor.DiscardOldestPolicy());// 拒绝策略关键参数来自 [ProxyConfig]线程数固定为PROCESSOR_NUMBER核心 最大不会扩容线程等待队列LinkedBlockingQueue容量5000拒绝策略DiscardOldestPolicy。二、DiscardOldestPolicy满时的行为JDK 的DiscardOldestPolicy语义是publicvoidrejectedExecution(Runnabler,ThreadPoolExecutore){if(!e.isShutdown()){e.getQueue().poll();// 1. 丢弃队头最旧的等待任务e.execute(r);// 2. 重新提交新任务此时队列有空位成功入队}}所以当「所有PROCESSOR_NUMBER个线程都在跑 队列 5000 已满」时新提交的刷新任务会触发它丢弃队列里最旧的那个待刷新任务队头腾出一个空位让最新提交的刷新任务入队执行全程不阻塞调用者也不抛RejectedExecutionException除非线程池已经 shutdown。这跟另外几种常见策略的区别策略满时的行为AbortPolicy默认丢弃新任务并抛RejectedExecutionExceptionCallerRunsPolicy由提交线程自己执行阻塞背压DiscardPolicy静默丢弃新任务DiscardOldestPolicy本处丢弃最旧等待任务接纳新任务三、对业务的实际影响被丢弃的是「最旧」的刷新请求而不是最新请求属于「牺牲老的、保新的」策略。被丢弃的那个 topic 的刷新不会因此永久失效因为 Caffeine 的refreshAfterWrite(20s)是惰性刷新该 topic 的旧路由仍在缓存里下次有访问且超过 20 秒时还会再次触发刷新只是这次刷新被推迟了。极端情况下如果刷新速度持续跟不上线程一直满DiscardOldestPolicy会不断丢最旧任务可能造成某些 topic 的路由长期停留在旧版本——但结合 [reload 失败返回旧值] 的兜底系统仍能工作只是路由更新延迟。一句话刷新线程池满时DiscardOldestPolicy会丢弃队列中最旧的待刷新任务来接纳新任务属于「不阻塞、不报错、保新弃旧」被丢弃的 topic 会靠下一次惰性刷新重新补上。