摘要:集群用途是將多個服務提供者合并為一個,并將這個暴露給服務消費者。比如發請求,接受服務提供者返回的數據等。如果包含,表明對應的服務提供者可能因網絡原因未能成功提供服務。如果不包含,此時還需要進行可用性檢測,比如檢測服務提供者網絡連通性等。
1.簡介
為了避免單點故障,現在的應用至少會部署在兩臺服務器上。對于一些負載比較高的服務,會部署更多臺服務器。這樣,同一環境下的服務提供者數量會大于1。對于服務消費者來說,同一環境下出現了多個服務提供者。這時會出現一個問題,服務消費者需要決定選擇哪個服務提供者進行調用。另外服務調用失敗時的處理措施也是需要考慮的,是重試呢,還是拋出異常,亦或是只打印異常等。為了處理這些問題,Dubbo 定義了集群接口 Cluster 以及及 Cluster Invoker。集群 Cluster 用途是將多個服務提供者合并為一個 Cluster Invoker,并將這個 Invoker 暴露給服務消費者。這樣一來,服務消費者只需通過這個 Invoker 進行遠程調用即可,至于具體調用哪個服務提供者,以及調用失敗后如何處理等問題,現在都交給集群模塊去處理。集群模塊是服務提供者和服務消費者的中間層,為服務消費者屏蔽了服務提供者的情況,這樣服務消費者就可以處理遠程調用相關事宜。比如發請求,接受服務提供者返回的數據等。這就是集群的作用。
Dubbo 提供了多種集群實現,包含但不限于 Failover Cluster、Failfast Cluster 和 Failsafe Cluster 等。每種集群實現類的用途不同,接下來我會一一進行分析。
2. 集群容錯在對集群相關代碼進行分析之前,這里有必要先來介紹一下集群容錯的所有組件。包含 Cluster、Cluster Invoker、Directory、Router 和 LoadBalance 等,先來看圖。
* 圖片來源:Dubbo 官方文檔
這張圖來自 Dubbo 官方文檔,接下來我會按照這張圖介紹集群工作過程。集群工作過程可分為兩個階段,第一個階段是在服務消費者初始化期間,集群 Cluster 實現類為服務消費者創建 Cluster Invoker 實例,即上圖中的 merge 操作。第二個階段是在服務消費者進行遠程調用時。以 FailoverClusterInvoker 為例,該類型 Cluster Invoker 首先會調用 Directory 的 list 方法列舉 Invoker 列表(可將 Invoker 簡單理解為服務提供者)。Directory 的用途是保存 Invoker,可簡單類比為 List
以上就是集群工作的整個流程,這里并沒介紹集群是如何容錯的。Dubbo 主要提供了這樣幾種容錯方式:
Failover Cluster - 失敗自動切換
Failfast Cluster - 快速失敗
Failsafe Cluster - 失敗安全
Failback Cluster - 失敗自動恢復
Forking Cluster - 并行調用多個服務提供者
這里暫時只對這幾種容錯模式進行簡單的介紹,在接下來的章節中,我會重點分析這幾種容錯模式的具體實現。好了,關于集群的工作流程和容錯模式先說到這,接下來進入源碼分析階段。
3.源碼分析 3.1 Cluster 實現類分析我在上一章提到了集群接口 Cluster 和 Cluster Invoker,這兩者是不同的。Cluster 是接口,而 Cluster Invoker 是一種 Invoker。服務提供者的選擇邏輯,以及遠程調用失敗后的的處理邏輯均是封裝在 Cluster Invoker 中。那么 Cluster 接口和相關實現類有什么用呢?用途比較簡單,用于生成 Cluster Invoker,僅此而已。下面我們來看一下源碼。
public class FailoverCluster implements Cluster {
public final static String NAME = "failover";
@Override
public Invoker join(Directory directory) throws RpcException {
// 創建并返回 FailoverClusterInvoker 對象
return new FailoverClusterInvoker(directory);
}
}
如上,FailoverCluster 總共就包含這幾行代碼,用于創建 FailoverClusterInvoker 對象,很簡單。下面再看一個。
public class FailbackCluster implements Cluster {
public final static String NAME = "failback";
@Override
public Invoker join(Directory directory) throws RpcException {
// 創建并返回 FailbackClusterInvoker 對象
return new FailbackClusterInvoker(directory);
}
}
如上,FailbackCluster 的邏輯也是很簡單,無需解釋了。所以接下來,我們把重點放在各種 Cluster Invoker 上
3.2 Cluster Invoker 分析我們首先從各種 Cluster Invoker 的父類 AbstractClusterInvoker 源碼開始說起。前面說過,集群工作過程可分為兩個階段,第一個階段是在服務消費者初始化期間,這個在服務引用那篇文章中已經分析過了,這里不再贅述。第二個階段是在服務消費者進行遠程調用時,此時 AbstractClusterInvoker 的 invoke 方法會被調用。列舉 Invoker,負載均衡等操作均會在此階段被執行。因此下面先來看一下 invoke 方法的邏輯。
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance = null;
// 綁定 attachments 到 invocation 中.
Map contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addAttachments(contextAttachments);
}
// 列舉 Invoker
List> invokers = list(invocation);
if (invokers != null && !invokers.isEmpty()) {
// 加載 LoadBalance
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 調用 doInvoke 進行后續操作
return doInvoke(invocation, invokers, loadbalance);
}
// 抽象方法,由子類實現
protected abstract Result doInvoke(Invocation invocation, List> invokers,
LoadBalance loadbalance) throws RpcException;
AbstractClusterInvoker 的 invoke 方法主要用于列舉 Invoker,以及加載 LoadBalance。最后再調用模板方法 doInvoke 進行后續操作。下面我們來看一下 Invoker 列舉方法 list(Invocation) 的邏輯,如下:
protected List> list(Invocation invocation) throws RpcException {
// 調用 Directory 的 list 方法
List> invokers = directory.list(invocation);
return invokers;
}
如上,AbstractClusterInvoker 中的 list 方法做的事情很簡單,只是簡單的調用了 Directory 的 list 方法,沒有其他更多的邏輯了。Directory 的 list 方法我在前面的文章中已經分析過了,這里就不贅述了。
接下來,我們把目光轉移到 AbstractClusterInvoker 的各種實現類上,來看一下這些實現類是如何實現 doInvoke 方法邏輯的。
3.2.1 FailoverClusterInvokerFailoverClusterInvoker 在調用失敗時,會自動切換 Invoker 進行重試。在無明確配置下,Dubbo 會使用這個類作為缺省 Cluster Invoker。下面來看一下該類的邏輯。
public class FailoverClusterInvoker extends AbstractClusterInvoker {
// 省略部分代碼
@Override
public Result doInvoke(Invocation invocation, final List> invokers, LoadBalance loadbalance) throws RpcException {
List> copyinvokers = invokers;
checkInvokers(copyinvokers, invocation);
// 獲取重試次數
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}
RpcException le = null;
List> invoked = new ArrayList>(copyinvokers.size());
Set providers = new HashSet(len);
// 循環調用,失敗重試
for (int i = 0; i < len; i++) {
if (i > 0) {
checkWhetherDestroyed();
// 在進行重試前重新列舉 Invoker,這樣做的好處是,如果某個服務掛了,
// 通過調用 list 可得到最新可用的 Invoker 列表
copyinvokers = list(invocation);
// 對 copyinvokers 進行判空檢查
checkInvokers(copyinvokers, invocation);
}
// 通過負載均衡選擇 Invoker
Invoker invoker = select(loadbalance, invocation, copyinvokers, invoked);
// 添加到 invoker 到 invoked 列表中
invoked.add(invoker);
// 設置 invoked 到 RPC 上下文中
RpcContext.getContext().setInvokers((List) invoked);
try {
// 調用目標 Invoker 的 invoke 方法
Result result = invoker.invoke(invocation);
return result;
} catch (RpcException e) {
if (e.isBiz()) {
throw e;
}
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
// 若重試均失敗,則拋出異常
throw new RpcException(..., "Failed to invoke the method ...");
}
}
如上,FailoverClusterInvoker 的 doInvoke 方法首先是獲取重試次數,然后根據重試次數進行循環調用,失敗后進行重試。在 for 循環內,首先是通過負載均衡組件選擇一個 Invoker,然后再通過這個 Invoker 的 invoke 方法進行遠程調用。如果失敗了,記錄下異常,并進行重試。重試時會再次調用父類的 list 方法列舉 Invoker。整個流程大致如此,不是很難理解。下面我們看一下 select 方法的邏輯。
protected Invoker select(LoadBalance loadbalance, Invocation invocation, List> invokers, List> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
// 獲取調用方法名
String methodName = invocation == null ? "" : invocation.getMethodName();
// 獲取 sticky 配置,sticky 表示粘滯連接。所謂粘滯連接是指讓服務消費者盡可能的
// 調用同一個服務提供者,除非該提供者掛了再進行切換
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
{
// 檢測 invokers 列表是否包含 stickyInvoker,如果不包含,
// 說明 stickyInvoker 代表的服務提供者掛了,此時需要將其置空
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
// 在 sticky 為 true,且 stickyInvoker != null 的情況下。如果 selected 包含
// stickyInvoker,表明 stickyInvoker 對應的服務提供者可能因網絡原因未能成功提供服務。
// 但是該提供者并沒掛,此時 invokers 列表中仍存在該服務提供者對應的 Invoker。
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
// availablecheck 表示是否開啟了可用性檢查,如果開啟了,則調用 stickyInvoker 的
// isAvailable 方法進行檢查,如果檢查通過,則直接返回 stickyInvoker。
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
}
// 如果線程走到當前代碼處,說明前面的 stickyInvoker 為空,或者不可用。
// 此時調用繼續調用 doSelect 選擇 Invoker
Invoker invoker = doSelect(loadbalance, invocation, invokers, selected);
// 如果 sticky 為 true,則將負載均衡組件選出的 Invoker 賦值給 stickyInvoker
if (sticky) {
stickyInvoker = invoker;
}
return invoker;
}
如上,select 方法的主要邏輯集中在了對粘滯連接特性的支持上。首先是獲取 sticky 配置,然后再檢測 invokers 列表中是否包含 stickyInvoker,如果不包含,則認為該 stickyInvoker 不可用,此時將其置空。這里的 invokers 列表可以看做是存活著的服務提供者列表,如果這個列表不包含 stickyInvoker,那自然而然的認為 stickyInvoker 掛了,所以置空。如果 stickyInvoker 存在于 invokers 列表中,此時要進行下一項檢測 ---- 檢測 selected 中是否包含 stickyInvoker。如果包含的話,說明 stickyInvoker 在此之前沒有成功提供服務(但其仍然處于存活狀態)。此時我們認為這個服務不可靠,不應該在重試期間內再次被調用,因此這個時候不會返回該 stickyInvoker。如果 selected 不包含 stickyInvoker,此時還需要進行可用性檢測,比如檢測服務提供者網絡連通性等。當可用性檢測通過,才可返回 stickyInvoker,否則調用 doSelect 方法選擇 Invoker。如果 sticky 為 true,此時會將 doSelect 方法選出的 Invoker 賦值給 stickyInvoker。
以上就是 select 方法的邏輯,這段邏輯看起來不是很復雜,但是信息量比較大。不搞懂 invokers 和 selected 兩個入參的含義,以及粘滯連接特性,這段代碼應該是沒法看懂的。大家在閱讀這段代碼時,不要忽略了對背景知識的理解。其他的不多說了,繼續向下分析。
private Invoker doSelect(LoadBalance loadbalance, Invocation invocation, List> invokers, List> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
if (invokers.size() == 1)
return invokers.get(0);
if (loadbalance == null) {
// 如果 loadbalance 為空,這里通過 SPI 加載 Loadbalance,默認為 RandomLoadBalance
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
// 通過負載均衡組件選擇 Invoker
Invoker invoker = loadbalance.select(invokers, getUrl(), invocation);
// 如果 selected 包含負載均衡選擇出的 Invoker,或者該 Invoker 無法經過可用性檢查,此時進行重選
if ((selected != null && selected.contains(invoker))
|| (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
try {
// 進行重選
Invoker rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
if (rinvoker != null) {
// 如果 rinvoker 不為空,則將其賦值給 invoker
invoker = rinvoker;
} else {
// rinvoker 為空,定位 invoker 在 invokers 中的位置
int index = invokers.indexOf(invoker);
try {
// 獲取 index + 1 位置處的 Invoker,以下代碼等價于:
// invoker = invokers.get((index + 1) % invokers.size());
invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
} catch (Exception e) {
logger.warn("... may because invokers list dynamic change, ignore.");
}
}
} catch (Throwable t) {
logger.error("cluster reselect fail reason is : ...");
}
}
return invoker;
}
doSelect 主要做了兩件事,第一是通過負載均衡組件選擇 Invoker。第二是,如果選出來的 Invoker 不穩定,或不可用,此時需要調用 reselect 方法進行重選。若 reselect 選出來的 Invoker 為空,此時定位 invoker 在 invokers 列表中的位置 index,然后獲取 index + 1 處的 invoker,這也可以看做是重選邏輯的一部分。關于負載均衡的選擇邏輯,我將會在下篇文章進行詳細分析。下面我們來看一下 reselect 方法的邏輯。
private Invoker reselect(LoadBalance loadbalance, Invocation invocation,
List> invokers, List> selected, boolean availablecheck)
throws RpcException {
List> reselectInvokers = new ArrayList>(invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());
// 根據 availablecheck 進行不同的處理
if (availablecheck) {
// 遍歷 invokers 列表
for (Invoker invoker : invokers) {
// 檢測可用性
if (invoker.isAvailable()) {
// 如果 selected 列表不包含當前 invoker,則將其添加到 reselectInvokers 中
if (selected == null || !selected.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
}
// reselectInvokers 不為空,此時通過負載均衡組件進行選擇
if (!reselectInvokers.isEmpty()) {
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
// 不檢查 Invoker 可用性
} else {
for (Invoker invoker : invokers) {
// 如果 selected 列表不包含當前 invoker,則將其添加到 reselectInvokers 中
if (selected == null || !selected.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
if (!reselectInvokers.isEmpty()) {
// 通過負載均衡組件進行選擇
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}
{
// 若線程走到此處,說明 reselectInvokers 集合為空,此時不會調用負載均衡組件進行篩選。
// 這里從 selected 列表中查找可用的 Invoker,并將其添加到 reselectInvokers 集合中
if (selected != null) {
for (Invoker invoker : selected) {
if ((invoker.isAvailable())
&& !reselectInvokers.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
}
if (!reselectInvokers.isEmpty()) {
// 再次進行選擇,并返回選擇結果
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
}
return null;
}
reselect 方法總結下來其實只做了兩件事情,第一是查找可用的 Invoker,并將其添加到 reselectInvokers 集合中。第二,如果 reselectInvokers 不為空,則通過負載均衡組件再次進行選擇。其中第一件事情又可進行細分,一開始,reselect 從 invokers 列表中查找有效可用的 Invoker,若未能找到,此時再到 selected 列表中繼續查找。關于 reselect 方法就先分析到這,繼續分析其他的 Cluster Invoker。
3.2.2 FailbackClusterInvokerFailbackClusterInvoker 會在調用失敗后,返回一個空結果給服務提供者。并通過定時任務對失敗的調用進行重傳,適合執行消息通知等操作。下面來看一下它的實現邏輯。
public class FailbackClusterInvoker extends AbstractClusterInvoker {
private static final long RETRY_FAILED_PERIOD = 5 * 1000;
private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2,
new NamedInternalThreadFactory("failback-cluster-timer", true));
private final ConcurrentMap> failed = new ConcurrentHashMap>();
private volatile ScheduledFuture retryFuture;
@Override
protected Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
// 選擇 Invoker
Invoker invoker = select(loadbalance, invocation, invokers, null);
// 進行調用
return invoker.invoke(invocation);
} catch (Throwable e) {
// 如果調用過程中發生異常,此時僅打印錯誤日志,不拋出異常
logger.error("Failback to invoke method ...");
// 記錄調用信息
addFailed(invocation, this);
// 返回一個空結果給服務消費者
return new RpcResult();
}
}
private void addFailed(Invocation invocation, AbstractClusterInvoker router) {
if (retryFuture == null) {
synchronized (this) {
if (retryFuture == null) {
// 創建定時任務,每隔5秒執行一次
retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
// 對失敗的調用進行重試
retryFailed();
} catch (Throwable t) {
// 如果發生異常,僅打印異常日志,不拋出
logger.error("Unexpected error occur at collect statistic", t);
}
}
}, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);
}
}
}
// 添加 invocation 和 invoker 到 failed 中,
// 這里的把 invoker 命名為 router,很奇怪,明顯名不副實
failed.put(invocation, router);
}
void retryFailed() {
if (failed.size() == 0) {
return;
}
// 遍歷 failed,對失敗的調用進行重試
for (Map.Entry> entry : new HashMap>(failed).entrySet()) {
Invocation invocation = entry.getKey();
Invoker invoker = entry.getValue();
try {
// 再次進行調用
invoker.invoke(invocation);
// 調用成功,則從 failed 中移除 invoker
failed.remove(invocation);
} catch (Throwable e) {
// 僅打印異常,不拋出
logger.error("Failed retry to invoke method ...");
}
}
}
}
這個類主要由3個方法組成,首先是 doInvoker,該方法負責初次的遠程調用。若遠程調用失敗,則通過 addFailed 方法將調用信息存入到 failed 中,等待定時重試。addFailed 在開始階段會根據 retryFuture 為空與非,來決定是否開啟定時任務。retryFailed 方法則是包含了失敗重試的邏輯,該方法會對 failed 進行遍歷,然后依次對 Invoker 進行調用。調用成功則將 Invoker 從 failed 中移除,調用失敗則忽略失敗原因。
以上就是 FailbackClusterInvoker 的執行邏輯,不是很復雜,繼續往下看。
3.2.3 FailfastClusterInvokerFailfastClusterInvoker 只會進行一次調用,失敗后立即拋出異常。適用于冪等操作,比如新增記錄。樓主日常開發中碰到過一次程序連續插入三條同樣的記錄問題,原因是新增記錄過程中包含了一些耗時操作,導致接口超時。而我當時使用的是 Dubbo 默認的 Cluster Invoker,即 FailoverClusterInvoker。其會在調用失敗后進行重試,所以導致插入服務提供者插入了3條同樣的數據。如果當時考慮使用 FailfastClusterInvoker,就不會出現這種問題了。當然此時接口仍然會超時,所以更合理的做法是使用 Dubbo 異步特性。或者優化服務邏輯,避免超時。
其他的不多說了,下面直接看源碼吧。
public class FailfastClusterInvoker extends AbstractClusterInvoker {
@Override
public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
// 選擇 Invoker
Invoker invoker = select(loadbalance, invocation, invokers, null);
try {
// 調用 Invoker
return invoker.invoke(invocation);
} catch (Throwable e) {
if (e instanceof RpcException && ((RpcException) e).isBiz()) {
// 拋出異常
throw (RpcException) e;
}
// 拋出異常
throw new RpcException(..., "Failfast invoke providers ...");
}
}
}
上面代碼比較簡單了,首先是通過 select 方法選擇 Invoker,然后進行遠程調用。如果調用失敗,則立即拋出異常。FailfastClusterInvoker 就先分析到這,下面分析 FailsafeClusterInvoker。
3.2.4 FailsafeClusterInvokerFailsafeClusterInvoker 是一種失敗安全的 Cluster Invoker。所謂的失敗安全是指,當調用過程中出現異常時,FailsafeClusterInvoker 僅會打印異常,而不會拋出異常。Dubbo 官方給出的應用場景是寫入審計日志等操作,這個場景我在日常開發中沒遇到過,沒發言權,就不多說了。下面直接分析源碼。
public class FailsafeClusterInvoker extends AbstractClusterInvoker {
@Override
public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
// 選擇 Invoker
Invoker invoker = select(loadbalance, invocation, invokers, null);
// 進行遠程調用
return invoker.invoke(invocation);
} catch (Throwable e) {
// 打印錯誤日志,但不拋出
logger.error("Failsafe ignore exception: " + e.getMessage(), e);
// 返回空結果忽略錯誤
return new RpcResult();
}
}
}
FailsafeClusterInvoker 的邏輯和 FailfastClusterInvoker 的邏輯一樣簡單,因此就不多說了。繼續下面分析。
3.2.5 ForkingClusterInvokerForkingClusterInvoker 會在運行時通過線程池創建多個線程,并發調用多個服務提供者。只要有一個服務提供者成功返回了結果,doInvoke 方法就會立即結束運行。ForkingClusterInvoker 的應用場景是在一些對實時性要求比較高讀操作(注意是讀操作,并行寫操作可能不安全)下使用,但這將會耗費更多的服務資源。下面來看該類的實現。
public class ForkingClusterInvoker extends AbstractClusterInvoker {
private final ExecutorService executor = Executors.newCachedThreadPool(
new NamedInternalThreadFactory("forking-cluster-timer", true));
@Override
public Result doInvoke(final Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
final List> selected;
// 獲取 forks 配置
final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
// 獲取超時配置
final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
// 如果 forks 配置不合理,則直接將 invokers 賦值給 selected
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {
selected = new ArrayList>();
// 循環選出 forks 個 Invoker,并添加到 selected 中
for (int i = 0; i < forks; i++) {
// 選擇 Invoker
Invoker invoker = select(loadbalance, invocation, invokers, selected);
if (!selected.contains(invoker)) {
selected.add(invoker);
}
}
}
// ----------------------? 分割線1 ?---------------------- //
RpcContext.getContext().setInvokers((List) selected);
final AtomicInteger count = new AtomicInteger();
final BlockingQueue