@Override
 @SuppressWarnings({"unchecked", "rawtypes"})
 public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
     try {
         checkInvokers(invokers, invocation);
         final List<Invoker<T>> selected;
         final int forks = getUrl().getParameter(FORKS_KEY, DEFAULT_FORKS);
         final int timeout = getUrl().getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
         if (forks <= 0 || forks >= invokers.size()) {
             selected = invokers;
         } else {
             selected = new ArrayList<>(forks);
             while (selected.size() < forks) {
                 Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
                 if (!selected.contains(invoker)) {
                     //Avoid add the same invoker several times.
                     selected.add(invoker);
                 }
             }
         }
         RpcContext.getContext().setInvokers((List) selected);
         final AtomicInteger count = new AtomicInteger();
         final BlockingQueue<Object> ref = new LinkedBlockingQueue<>();
         for (final Invoker<T> invoker : selected) {
             executor.execute(() -> {
                 try {
                     Result result = invoker.invoke(invocation);
                     ref.offer(result);
                 } catch (Throwable e) {
                     int value = count.incrementAndGet();
                     if (value >= selected.size()) {
                         ref.offer(e);
                     }
                 }
             });
         }
         try {
             Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
             if (ret instanceof Throwable) {
                 Throwable e = (Throwable) ret;
                 throw new RpcException("");
             }
             return (Result) ret;
         } catch (InterruptedException e) {
             throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e);
         }
     } finally {
         // clear attachments which is binding to current thread.
         RpcContext.getContext().clearAttachments();
     }
 }
复制代码