class NioWorker implements Runnable {    ...    private final Queue writeTaskQueue = new LinkedTransferQueue();    ...}

public class DubboInvoker<T> extends AbstractInvoker<T> {    ...   @Override    protected Result doInvoke(final Invocation invocation) throws Throwable {         ...         return (Result) currentClient.request(inv, timeout).get(); //currentClient.request(inv, timeout)返回了一個(gè)DefaultFuture    }    ...}

public class DefaultFuture implements ResponseFuture {
    private static final Map CHANNELS = new ConcurrentHashMap();    private static final Map FUTURES = new ConcurrentHashMap();
    // invoke id.    private final long id;      //Dubbo請(qǐng)求的id，每個(gè)消費(fèi)者都是一個(gè)從0開(kāi)始的long類(lèi)型    private final Channel channel;    private final Request request;    private final int timeout;    private final Lock lock = new ReentrantLock();    private final Condition done = lock.newCondition();    private final long start = System.currentTimeMillis();    private volatile long sent;    private volatile Response response;    private volatile ResponseCallback callback;    public DefaultFuture(Channel channel, Request request, int timeout) {        this.channel = channel;        this.request = request;        this.id = request.getId();        this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);        // put into waiting map.        FUTURES.put(id, this);    //等待時(shí)以id為key把Future放入全局的Future Map中，這樣回復(fù)數(shù)據(jù)回來(lái)了可以根據(jù)id找到對(duì)應(yīng)的Future通知主線(xiàn)程        CHANNELS.put(id, channel);    }

public void run() {????for?(;;)?{       ....????????????SelectorUtil.select(selector);
????????????proce***egisterTaskQueue();????????????processWriteTaskQueue();?//先處理隊(duì)列里的寫(xiě)請(qǐng)求            processSelectedKeys(selector.selectedKeys()); //再處理select事件,讀寫(xiě)都可能有       ....    }}
private void processWriteTaskQueue() throws IOException {    for (;;) {        final Runnable task = writeTaskQueue.poll();//這個(gè)隊(duì)列就是調(diào)用線(xiàn)程把請(qǐng)求放進(jìn)去的隊(duì)列        if (task == null) {            break;        }        task.run(); //寫(xiě)數(shù)據(jù)        cleanUpCancelledKeys();    }}

void writeFromTaskLoop(final NioSocketChannel ch) {    if (!ch.writeSuspended) { //這個(gè)地方很重要，如果writeSuspended了，那么就直接跳過(guò)這次寫(xiě)        write0(ch);    }}
private void write0(NioSocketChannel channel) {    ......    final int writeSpinCount = channel.getConfig().getWriteSpinCount(); //netty可配置的一個(gè)參數(shù)，默認(rèn)是16    synchronized (channel.writeLock) {        channel.inWriteNowLoop = true;        for (;;) {            for (int i = writeSpinCount; i > 0; i --) { //每次最多嘗試16次                localWrittenBytes = buf.transferTo(ch);                if (localWrittenBytes != 0) {                    writtenBytes += localWrittenBytes;                    break;????????????????}                if (buf.finished()) {                    break;                }            }
            if (buf.finished()) {                // Successful write - proceed to the next message.                buf.release();                channel.currentWriteEvent = null;                channel.currentWriteBuffer = null;                evt = null;                buf = null;                future.setSuccess();            } else {                // Not written fully - perhaps the kernel buffer is full.                //重點(diǎn)在這，如果寫(xiě)16次還沒(méi)寫(xiě)完，可能是內(nèi)核緩沖區(qū)滿(mǎn)了，writeSuspended被設(shè)置為true                addOpWrite = true;                channel.writeSuspended = true;                ......            }            ......            if (open) {                if (addOpWrite) {                    setOpWrite(channel);                } else if (removeOpWrite) {                    clearOpWrite(channel);                }            }            ......        }        fireWriteComplete(channel, writtenBytes);    }

void writeFromSelectorLoop(final SelectionKey k) {        NioSocketChannel ch = (NioSocketChannel) k.attachment();        ch.writeSuspended = false;        write0(ch);    }

public void run() {????for?(;;)?{        ....????????SelectorUtil.select(selector);
        proce***egisterTaskQueue();        processWriteTaskQueue();        processSelectedKeys(selector.selectedKeys()); //再處理select事件,讀寫(xiě)都可能有        ....    }}
    private void processSelectedKeys(Set selectedKeys) throws IOException {        for (Iterator i = selectedKeys.iterator(); i.hasNext();) {            SelectionKey k = i.next();            i.remove();            try {                int readyOps = k.readyOps();                if ((readyOps & SelectionKey.OP_READ) != 0 || readyOps == 0) {                    if (!read(k)) {                        // Connection already closed - no need to handle write.                        continue;                    }                }                if ((readyOps & SelectionKey.OP_WRITE) != 0) {                    writeFromSelectorLoop(k);                }            } catch (CancelledKeyException e) {                close(k);            }
            if (cleanUpCancelledKeys()) {                break; // break the loop to avoid ConcurrentModificationException            }        }    }    private boolean read(SelectionKey k) {       ......
            // Fire the event.            fireMessageReceived(channel, buffer);  //讀取完后，最終會(huì)調(diào)用這個(gè)函數(shù)，發(fā)送一個(gè)收到信息的事件       ......
    }

public class AllChannelHandler extends WrappedChannelHandler {    ......    public void received(Channel channel, Object message) throws RemotingException {        ExecutorService cexecutor = getExecutorService();        try {            cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));        } catch (Throwable t) {            //TODO A temporary solution to the problem that the exception information can not be sent to the opposite end after the thread pool is full. Need a refactoring            //fix The thread pool is full, refuses to call, does not return, and causes the consumer to wait for time out            if(message instanceof Request && t instanceof RejectedExecutionException){                Request request = (Request)message;                if(request.isTwoWay()){                    String msg = "Server side(" + url.getIp() + "," + url.getPort() + ") threadpool is exhausted ,detail msg:" + t.getMessage();                    Response response = new Response(request.getId(), request.getVersion());                    response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR);                    response.setErrorMessage(msg);                    channel.send(response);                    return;                }            }            throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);        }    }    ......}

public class HeaderExchangeHandler implements ChannelHandlerDelegate {    ......    Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {        Response res = new Response(req.getId(), req.getVersion());        ......        // find handler by message class.        Object msg = req.getData();        try {            // handle data.            Object result = handler.reply(channel, msg);   //真正的業(yè)務(wù)邏輯類(lèi)            res.setStatus(Response.OK);            res.setResult(result);        } catch (Throwable e) {            res.setStatus(Response.SERVICE_ERROR);            res.setErrorMessage(StringUtils.toString(e));        }        return res;    }
    public void received(Channel channel, Object message) throws RemotingException {       ......
            if (message instanceof Request) {                // handle request.                Request request = (Request) message;
                    if (request.isTwoWay()) {                        Response response = handleRequest(exchangeChannel, request); //處理業(yè)務(wù)邏輯，得到一個(gè)Response                        channel.send(response);  //回寫(xiě)response                    }            }       ......
 }

for (int i = writeSpinCount; i > 0; i --) {    localWrittenBytes = buf.transferTo(ch);    if (localWrittenBytes != 0) {        writtenBytes += localWrittenBytes;        break;    }    if (buf.finished()) {        break;    } }
    if (buf.finished()) {         // Successful write - proceed to the next message.         buf.release();         channel.currentWriteEvent = null;         channel.currentWriteBuffer = null;         evt = null;         buf = null;         future.setSuccess();     } else {          // Not written fully - perhaps the kernel buffer is full.          addOpWrite = true;          channel.writeSuspended = true;

public class HeaderExchangeHandler implements ChannelHandlerDelegate {    static void handleResponse(Channel channel, Response response) throws RemotingException {        if (response != null && !response.isHeartbeat()) {            DefaultFuture.received(channel, response);        }    }    public void received(Channel channel, Object message) throws RemotingException {        ......        if (message instanceof Response) {                handleResponse(channel, (Response) message);        }        ......}

 public static void received(Channel channel, Response response) {         ......         DefaultFuture future = FUTURES.remove(response.getId());         if (future != null) {            future.doReceived(response);         }         ......    }

<dubbo:protocol name="dubbo" port="20880" dispatcher="all" threadpool="fixed" threads="2000" />

net.ipv4.tcp_rmem = 4096 873800 16777216net.ipv4.tcp_wmem = 4096 873800 16777216

   //服務(wù)端實(shí)現(xiàn)   public String sayHello(String name) {        counter.getAndIncrement();        long start = System.currentTimeMillis();        try {            Thread.sleep(rand.nextInt(300));        } catch (InterruptedException e) {        }        String result = "Hello " + name + name  + ", response form provider: " + RpcContext.getContext().getLocalAddress();        long end = System.currentTimeMillis();        timer.getAndAdd(end-start);        return result;    }

        for(int i = 0; i < N; i ++) {            threads[i] = new Thread(new Runnable() {                @Override                public void run() {                    while(true) {                        Long start = System.currentTimeMillis();                        String hello = service.sayHello(z);                        Long end = System.currentTimeMillis();                        totalTime.getAndAdd(end-start);                        counter.getAndIncrement();                    }                }});            threads[i].start();        }

(base) niuxinli@ubuntu:~$ ss -itState                            Recv-Q                        Send-Q                                                       Local Address:Port                                                          Peer Address:PortESTAB                            0                             36                                                             192.168.1.7:ssh                                                            192.168.1.4:58931                            cubic wscale:8,2 rto:236 rtt:33.837/8.625 ato:40 mss:1460 pmtu:1500 rcvmss:1460 advmss:1460 cwnd:10 bytes_acked:559805 bytes_received:54694 segs_out:2754 segs_in:2971 data_segs_out:2299 data_segs_in:1398 send 3.5Mbps pacing_rate 6.9Mbps delivery_rate 44.8Mbps busy:36820ms unacked:1 rcv_rtt:513649 rcv_space:16130 rcv_ssthresh:14924 minrtt:0.112ESTAB                            0                             0                                                              192.168.1.7:36666                                                          192.168.1.7:2181                             cubic wscale:7,7 rto:204 rtt:0.273/0.04 ato:40 mss:33344 pmtu:65535 rcvmss:536 advmss:65483 cwnd:10 bytes_acked:2781 bytes_received:3941 segs_out:332 segs_in:170 data_segs_out:165 data_segs_in:165 send 9771.1Mbps lastsnd:4960 lastrcv:4960 lastack:4960 pacing_rate 19497.6Mbps delivery_rate 7621.5Mbps app_limited busy:60ms rcv_space:65535 rcv_ssthresh:66607 minrtt:0.035ESTAB                            0                             27474                                                          192.168.1.7:20880                                                          192.168.1.5:60760                            cubic wscale:7,7 rto:204 rtt:1.277/0.239 ato:40 mss:1448 pmtu:1500 rcvmss:1448 advmss:1448 cwnd:625 ssthresh:20 bytes_acked:96432644704 bytes_received:49286576300 segs_out:68505947 segs_in:36666870 data_segs_out:67058676 data_segs_in:35833689 send 5669.5Mbps pacing_rate 6801.4Mbps delivery_rate 627.4Mbps app_limited busy:1340536ms rwnd_limited:400372ms(29.9%) sndbuf_limited:433724ms(32.4%) unacked:70 retrans:0/5 rcv_rtt:1.308 rcv_space:336692 rcv_ssthresh:2095692 notsent:6638 minrtt:0.097

(base) niuxinli@ubuntu:~$ netstat -natActive Internet connections (servers and established)Proto Recv-Q Send-Q Local Address           Foreign Address         Statetcp        0      0 0.0.0.0:20880           0.0.0.0:*               LISTENtcp        0     36 192.168.1.7:22          192.168.1.4:58931       ESTABLISHEDtcp        0      0 192.168.1.7:36666       192.168.1.7:2181        ESTABLISHEDtcp        0  65160 192.168.1.7:20880       192.168.1.5:60760       ESTABLISHED

 Recv-Q       Established: The count of bytes not copied by the user program connected to this socket.
   Send-Q       Established: The count of bytes not acknowledged by the remote host.