@Override
        public void run() {
            while (isRunning) {
                try {
                    while (!brokerQueue.isEmpty()) {
                        for (String serviceKey : brokerQueue.keySet()) {
                            ConnectionNode node = brokerQueue.get(serviceKey);
                            ServiceHolder h = servicesCache.get(serviceKey);
                            if (h != null) {
                                brokerQueue.remove(serviceKey);
                                continue;
                            }
                            RpcServiceInvoker invoker =
                                    new RpcServiceInvoker(clientFactory, node.clazzType,
                                            node.getConfig(), node.getAddressInfo());
                            Object service =
                                    Proxy.newProxyInstance(node.clazzType.getClassLoader(),
                                            new Class[]{node.clazzType}, invoker);
                            try {
                                invoker.getClientOnce();
                                resetRmtAddrErrCount(node.getHostAndPortStr());
                            } catch (Throwable e) {
                                if (e instanceof LocalConnException) {
                                    addRmtAddrErrCount(node.getHostAndPortStr());
                                }
                                brokerQueue.remove(serviceKey);
                                continue;
                            }
                            servicesCache.putIfAbsent(serviceKey, new ServiceHolder(service, invoker));
                            updateTime.put(serviceKey, System.currentTimeMillis());
                            brokerQueue.remove(serviceKey);
                        }
                    }
                    long cur = System.currentTimeMillis();
                    if (cur - lastCheckTime.get() >= 30000) {
                        ArrayList<String> tmpKeyList = new ArrayList<>();
                        for (Map.Entry<String, Long> entry : updateTime.entrySet()) {
                            if (entry.getKey() == null || entry.getValue() == null) {
                                continue;
                            }
                            if (cur - entry.getValue() > DEFAULT_IDLE_TIME) {
                                tmpKeyList.add(entry.getKey());
                            }
                        }
                        for (String serviceKey : tmpKeyList) {
                            servicesCache.remove(serviceKey);
                            updateTime.remove(serviceKey);
                        }
                        rmvAllExpiredRecords();
                        rmvExpiredUnavailableBrokers();
                        lastCheckTime.set(System.currentTimeMillis());
                    }
                } catch (Throwable e2) {
                    logger.warn("[connectionManager]: runner found throw error info ", e2);
                }
                ThreadUtils.sleep(80);
            }
        }
ConnectionManager
Manage and recycle network connection resources
inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/RpcServiceFactory.java