private final Random random = new Random();

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
	int length = invokers.size(); // Number of invokers 总个数
	int totalWeight = 0; // The sum of weights 总权重
	boolean sameWeight = true; // Every invoker has the same weight? 权重是否相同
	for (int i = 0; i < length; i++) {
		int weight = getWeight(invokers.get(i), invocation);
		totalWeight += weight; // Sum 累计总权重
		if (sameWeight && i > 0
				&& weight != getWeight(invokers.get(i - 1), invocation)) {
			sameWeight = false;	// 计算权重是否相同
		}
	}
	if (totalWeight > 0 && !sameWeight) {
		// If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
		// 如果权重不相同，并且总权重大于0，则按总权重数随机
		int offset = random.nextInt(totalWeight);
		// Return a invoker based on the random value. 确定随机值落在哪个片断上
		for (int i = 0; i < length; i++) {
			offset -= getWeight(invokers.get(i), invocation);
			if (offset < 0) {
				return invokers.get(i);
			}
		}
	}
	// If all invokers have the same weight value or totalWeight=0, return evenly.
	// 如果权重相同，或总权重为0，则均等随机获取
	return invokers.get(random.nextInt(length));
}

private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
	String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
	int length = invokers.size(); // Number of invokers 总个数
	int maxWeight = 0; // The maximum weight 最大权重
	int minWeight = Integer.MAX_VALUE; // The minimum weight 最小权重
	final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();
	int weightSum = 0; // 总权重
	for (int i = 0; i < length; i++) {
		int weight = getWeight(invokers.get(i), invocation);
		maxWeight = Math.max(maxWeight, weight); // Choose the maximum weight 替换
		minWeight = Math.min(minWeight, weight); // Choose the minimum weight 替换
		if (weight > 0) {
			invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
			weightSum += weight;
		}
	}
	AtomicPositiveInteger sequence = sequences.get(key);
	if (sequence == null) {
		sequences.putIfAbsent(key, new AtomicPositiveInteger());
		sequence = sequences.get(key);
	}
	int currentSequence = sequence.getAndIncrement();
	// 按公约后的权重设置轮询比率
	if (maxWeight > 0 && minWeight < maxWeight) {
		int mod = currentSequence % weightSum;
		for (int i = 0; i < maxWeight; i++) {
			for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
				final Invoker<T> k = each.getKey();
				final IntegerWrapper v = each.getValue();
				if (mod == 0 && v.getValue() > 0) {
					return k;
				}
				if (v.getValue() > 0) {
					v.decrement();
					mod--;
				}
			}
		}
	}
	// Round robin 轮循依次获取
	return invokers.get(currentSequence % length);
}

private final Random random = new Random();

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
	int length = invokers.size(); // Number of invokers 总个数
	int leastActive = -1; // The least active value of all invokers 最小的活跃数
	int leastCount = 0; // The number of invokers having the same least active value (leastActive) 相同最小活跃的个数
	int[] leastIndexs = new int[length]; // The index of invokers having the same least active value (leastActive) 相同最小活跃数的下标
	int totalWeight = 0; // The sum of weights 总权重
	int firstWeight = 0; // Initial value, used for comparision 第一个权重，用于计算是否相同
	boolean sameWeight = true; // Every invoker has the same weight value? 是否所有权重相同
	for (int i = 0; i < length; i++) {
		Invoker<T> invoker = invokers.get(i);
		int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // Active number 获取活跃数
		int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // Weight 获取权重
		if (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value. 发现有小于最小活跃数的
			leastActive = active; // Record the current least active value 记录最小活跃数
			leastCount = 1; // Reset leastCount, count again based on current leastCount 重新统计拥有相同最小活跃数的个数
			leastIndexs[0] = i; // Reset 重新记录最小活跃数的下标
			totalWeight = weight; // Reset 重新累计总权重
			firstWeight = weight; // Record the weight the first invoker 记录第一个权重
			sameWeight = true; // Reset, every invoker has the same weight value? 还原权重相等标识
		} else if (active == leastActive) { // If current invoker's active value equals with leaseActive, then accumulating. 如果活跃数相等
			leastIndexs[leastCount++] = i; // Record index number of this invoker 累计
			totalWeight += weight; // Add this invoker's weight to totalWeight. 累计
			// If every invoker has the same weight? 判断所有权重是否相同
			if (sameWeight && i > 0
					&& weight != firstWeight) {
				sameWeight = false;
			}
		}
	}
	// assert(leastCount > 0)
	if (leastCount == 1) {
		// If we got exactly one invoker having the least active value, return this invoker directly.
		// 如果只有一个最小活跃数的就直接返回
		return invokers.get(leastIndexs[0]);
	}
	if (!sameWeight && totalWeight > 0) {
		// If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
		// 权重不同，且总权重大于0，和RandomLoadBalance一样按总权重随机
		int offsetWeight = random.nextInt(totalWeight);
		// Return a invoker based on the random value.
		for (int i = 0; i < leastCount; i++) {
			int leastIndex = leastIndexs[i];
			offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
			if (offsetWeight <= 0)
				return invokers.get(leastIndex);
		}
	}
	// If all invokers have the same weight value or totalWeight=0, return evenly.
	// 权重相同，和RandomLoadBalance一样均等随机获取
	return invokers.get(leastIndexs[random.nextInt(leastCount)]);
}

private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();

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;
	// 生成调用列表的hashcode
	int identityHashCode = System.identityHashCode(invokers);
	// 根据方法名key获取一致性hash选择器，内部类ConsistentHashSelector
	ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
	if (selector == null || selector.identityHashCode != identityHashCode) {
		selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode));
		selector = (ConsistentHashSelector<T>) selectors.get(key);
	}
	// 选择节点
	return selector.select(invocation);
}

private static final class ConsistentHashSelector<T> {
	// 虚拟节点
	private final TreeMap<Long, Invoker<T>> virtualInvokers;
	// 副本数
	private final int replicaNumber;
	// 调用节点的hashcode，用于判断加入selector选择器时是否重复
	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.arguments，获取参数
		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]);
		}
		// 创建虚拟节点，对每一个Invoker生成replicaNumber个虚拟节点并存放于virtualInvokers中
		for (Invoker<T> invoker : invokers) {
			String address = invoker.getUrl().getAddress();
			for (int i = 0; i < replicaNumber / 4; i++) {
				byte[] digest = md5(address + i);
				for (int h = 0; h < 4; h++) {
					long m = hash(digest, h);
					virtualInvokers.put(m, invoker);
				}
			}
		}
	}
	
	public Invoker<T> select(Invocation invocation) {
		// 获取参数
		String key = toKey(invocation.getArguments());
		byte[] digest = md5(key);
		// 返回虚拟节点中对应hash值的Invoker
		return selectForKey(hash(digest, 0));
	}
	
	// ...
}

public interface AsyncService {
    CompletableFuture<String> sayHello(String name);
}

public class AsyncServiceImpl implements AsyncService {
    public CompletableFuture<String> sayHello(String name) {
		// 通过return CompletableFuture.supplyAsync()，业务执行已从Dubbo线程切换到业务线程，避免了对Dubbo线程池的阻塞
        // 建议为supplyAsync提供自定义线程池，避免使用JDK公用线程池
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "async response from provider.";
        });
    }
}

// Dubbo提供了一个类似Serverlet 3.0的异步接口AsyncContext，在没有CompletableFuture签名接口的情况下，也可以实现Provider端的异步执行
public interface AsyncService {
    String sayHello(String name);
}

public class AsyncServiceImpl implements AsyncService {
    public String sayHello(String name) {
        final AsyncContext asyncContext = RpcContext.startAsync();
        new Thread(() -> {
            // 如果要使用上下文，则必须要放在第一句执行
            asyncContext.signalContextSwitch();
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 写回响应
            asyncContext.write("Hello " + name + ", response from provider.");
        }).start();
        return null;
    }
}

// 调用直接返回CompletableFuture
CompletableFuture<String> future = asyncService.sayHello("async call request");
// 增加回调
future.whenComplete((v, t) -> {
    if (t != null) {
        t.printStackTrace();
    } else {
        System.out.println("Response: " + v);
    }
});
// 这将早于结果输出
System.out.println("Executed before response return.");

// JavassistProxyFactory
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
	return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}


// InvokerInvocationHandler
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
	String methodName = method.getName();
	Class<?>[] parameterTypes = method.getParameterTypes();
	if (method.getDeclaringClass() == Object.class) {
		return method.invoke(invoker, args);
	}
	if ("toString".equals(methodName) && parameterTypes.length == 0) {
		return invoker.toString();
	}
	if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
		return invoker.hashCode();
	}
	if ("equals".equals(methodName) && parameterTypes.length == 1) {
		return invoker.equals(args[0]);
	}

	RpcInvocation invocation;
	if (RpcUtils.hasGeneratedFuture(method)) {
		Class<?> clazz = method.getDeclaringClass();
		String syncMethodName = methodName.substring(0, methodName.length() - Constants.ASYNC_SUFFIX.length());
		Method syncMethod = clazz.getMethod(syncMethodName, method.getParameterTypes());
		invocation = new RpcInvocation(syncMethod, args);
		invocation.setAttachment(Constants.FUTURE_GENERATED_KEY, "true");
		invocation.setAttachment(Constants.ASYNC_KEY, "true");
	} else {
		invocation = new RpcInvocation(method, args);
		if (RpcUtils.hasFutureReturnType(method)) {
			invocation.setAttachment(Constants.FUTURE_RETURNTYPE_KEY, "true");
			invocation.setAttachment(Constants.ASYNC_KEY, "true");
		}
	}
	// ----> 这里还是consumer的业务线程，所以在这里读取ThreadLocal里的traceId，再放入RpcInvocation的attachment
	// 那么Provider端就可以从收到的RpcInvocation取出透传的traceId
	return invoker.invoke(invocation).recreate();
}

// DubboProtocol
private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
	@Override
	public CompletableFuture<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
		if (message instanceof Invocation) {
			Invocation inv = (Invocation) message;
			Invoker<?> invoker = getInvoker(channel, inv);
			// need to consider backward-compatibility if it's a callback
			if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
				String methodsStr = invoker.getUrl().getParameters().get("methods");
				boolean hasMethod = false;
				if (methodsStr == null || !methodsStr.contains(",")) {
					hasMethod = inv.getMethodName().equals(methodsStr);
				} else {
					String[] methods = methodsStr.split(",");
					for (String method : methods) {
						if (inv.getMethodName().equals(method)) {
							hasMethod = true;
							break;
						}
					}
				}
				if (!hasMethod) {
					logger.warn(new IllegalStateException("The methodName " + inv.getMethodName()
							+ " not found in callback service interface ,invoke will be ignored."
							+ " please update the api interface. url is:"
							+ invoker.getUrl()) + " ,invocation is :" + inv);
					return null;
				}
			}
			RpcContext rpcContext = RpcContext.getContext();
			boolean supportServerAsync = invoker.getUrl().getMethodParameter(inv.getMethodName(), Constants.ASYNC_KEY, false);
			if (supportServerAsync) {
				CompletableFuture<Object> future = new CompletableFuture<>();
				rpcContext.setAsyncContext(new AsyncContextImpl(future));
			}
			rpcContext.setRemoteAddress(channel.getRemoteAddress());
			// ----> 这里是Provider端的实现类的线程中，可以在inv这里取出traceId放入ThreadLocal
			Result result = invoker.invoke(inv);

			if (result instanceof AsyncRpcResult) {
				return ((AsyncRpcResult) result).getResultFuture().thenApply(r -> (Object) r);
			} else {
				return CompletableFuture.completedFuture(result);
			}
		}
		throw new RemotingException(channel, "Unsupported request: "
				+ (message == null ? null : (message.getClass().getName() + ": " + message))
				+ ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
	}
		
	// ...
};

// AbstractInvoker
public Result invoke(Invocation inv) throws RpcException {
	// ...
	RpcInvocation invocation = (RpcInvocation) inv;
	invocation.setInvoker(this);
	if (attachment != null && attachment.size() > 0) {
		invocation.addAttachmentsIfAbsent(attachment);
	}
	Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
	if (contextAttachments != null && contextAttachments.size() != 0) {
		/**
		 * invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
		 * because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered
		 * by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is
		 * a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information).
		 */
		invocation.addAttachments(contextAttachments);
	}
	// ...
	try {
		return doInvoke(invocation);
	} // ...
}