NIO 高并发客户端

Question

我正在编写一个工具，它将针对网络服务器生成大量 HTTP 调用。目前，我对每秒可以发出多少个请求很感兴趣。我现在对这些请求的结果不感兴趣。

我正在测量向 google.com 发送 1k 个请求所花费的时间，我得到了 69 毫秒：

但是当我使用 WireShark 嗅探流量时，我发现发送所有 GET 请求需要将近 4 秒：

• 通话开始

• 通话结束

工具已在 Windows 10、I7 1.8 Ghz、32 GB RAM 上从 IntelliJ 运行。

我的问题是：为什么我有这种差异？发送 1k 个 HTTP GET 请求应该很快，但几乎需要 4 秒。我在这里做错了什么？

上面的代码仅用于测试目的，它很丑陋，所以请多多包涵。另外我对蔚来也不是很熟悉。

import org.apache.commons.lang3.time.StopWatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicInteger;


public class UJPPHighTrafficClient {
    public static final Logger logger = LoggerFactory.getLogger(UJPPHighTrafficClient.class);

    public static final int iterations = 1000;

    public static void main(String[] args) {
        doStartClient();
    }

    private static void doStartClient() {
        logger.info("starting the client");   

        UJPPHighTrafficExecutor executor = new UJPPHighTrafficExecutor();
           
        StopWatch watch = new StopWatch();
        watch.start();

        for (int i = 0; i < iterations; i++) {
            executor.run();
        }
        watch.stop();

        logger.info("Run " + iterations + " executions in " + watch.getTime() + " milliseconds");

    }
}


import org.apache.http.HttpHost;
import org.apache.http.HttpResponse;
import org.apache.http.ProtocolVersion;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.config.ConnectionConfig;
import org.apache.http.impl.nio.DefaultHttpClientIODispatch;
import org.apache.http.impl.nio.pool.BasicNIOConnPool;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHttpEntityEnclosingRequest;
import org.apache.http.nio.protocol.*;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOEventDispatch;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.protocol.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.concurrent.atomic.AtomicInteger;

public class UJPPHighTrafficExecutor {
    private final Logger logger = LoggerFactory.getLogger("debug");
    public static ConnectingIOReactor requestsReactor = null;
    private static BasicNIOConnPool clientConnectionPool = null;
    public static HttpAsyncRequester clientRequester = null;
    public static Thread runnerThread = null;
    private static AtomicInteger counter = null;

    public static final int cores = Runtime.getRuntime().availableProcessors() * 2;

    public UJPPHighTrafficExecutor() {
        counter = new AtomicInteger();
        counter.set(0);
        initializeConnectionManager();
    }

    public void initializeConnectionManager() {

        try {
            requestsReactor =
                    new DefaultConnectingIOReactor(IOReactorConfig.
                            custom().
                            setIoThreadCount(cores).
                            build());

            clientConnectionPool = new BasicNIOConnPool(requestsReactor, ConnectionConfig.DEFAULT);

            clientConnectionPool.setDefaultMaxPerRoute(cores);
            clientConnectionPool.setMaxTotal(100);

            clientRequester = initializeHttpClient(requestsReactor);

        } catch (IOReactorException ex) {
            logger.error(" initializeConnectionManager " + ex.getMessage());
        }
    }

    private HttpAsyncRequester initializeHttpClient(final ConnectingIOReactor ioReactor) {
        // Create HTTP protocol processing chain
        HttpProcessor httpproc = HttpProcessorBuilder.create()
                // Use standard client-side protocol interceptors
                .add(new RequestContent(true)).
                        add(new RequestTargetHost()).
                        add(new RequestConnControl())
                .add(new RequestExpectContinue(true)).
                        build();

        // Create HTTP requester
        HttpAsyncRequester requester = new HttpAsyncRequester(httpproc);
        // Create client-side HTTP protocol handler
        HttpAsyncRequestExecutor protocolHandler = new HttpAsyncRequestExecutor();
        // Create client-side I/O event dispatch
        final IOEventDispatch ioEventDispatch =
                new DefaultHttpClientIODispatch(protocolHandler, ConnectionConfig.DEFAULT);

        // Run the I/O reactor in a separate thread
        runnerThread = new Thread("Client") {
            @Override
            public void run() {
                try {
                    ioReactor.execute(ioEventDispatch);
                } catch (InterruptedIOException ex) {
                    logger.error("Interrupted", ex);
                } catch (IOException e) {
                    logger.error("I/O error", e);
                } catch (Exception e) {
                    logger.error("Exception encountered in Client ", e.getMessage(), e);
                }
                logger.info("Client shutdown");
            }
        };
        runnerThread.start();

        return requester;
    }

    public void run() {
        HttpHost httpHost = new HttpHost("google.com", 80, "http");
        final HttpCoreContext coreContext = HttpCoreContext.create();
        ProtocolVersion ver = new ProtocolVersion("HTTP", 1, 1);
        BasicHttpEntityEnclosingRequest request = new BasicHttpEntityEnclosingRequest("GET", "/", ver);

        clientRequester.execute(new BasicAsyncRequestProducer(httpHost, request), new BasicAsyncResponseConsumer(),
                clientConnectionPool, coreContext,
                // Handle HTTP response from a callback
                new FutureCallback<HttpResponse>() {

                    @Override
                    public void completed(final HttpResponse response) {
                        logger.info("Completed " + response.toString());
                        checkCounter();
                    }

                    @Override
                    public void failed(final Exception ex) {
                        logger.info("Failed " + ex.getMessage());
                        checkCounter();
                    }

                    @Override
                    public void cancelled() {
                        logger.info("Cancelled ");
                        checkCounter();
                    }
                });
    }

    private void checkCounter() {
        counter.set(counter.get() + 1);
        if (counter.get() == UJPPHighTrafficClient.iterations) {
            try {
                requestsReactor.shutdown();
            } catch (Exception ex) {

            }

        }
    }
}

Answer 1

您的代码正在计时设置 1000 次 http 连接迭代的时间，而不是完成这些连接的时间，其中许多连接在 3-4 秒后仍在运行。要查看更准确的数字，请将本地字段 t0 放入 UJPPHighTrafficExecutor：

public class UJPPHighTrafficExecutor {
    long t0 = System.nanoTime();

...然后checkCounter() 可以打印完成所有迭代的时间：

private void checkCounter() {
    counter.set(counter.get() + 1);
    if (counter.get() == UJPPHighTrafficClient.iterations) {
        try {
            requestsReactor.shutdown();
        } catch (Exception ex) {

        }
        long t1 = System.nanoTime();
        System.out.println("ELAPSED MILLIS: ~"+TimeUnit.NANOSECONDS.toMillis(t1-t0));
    }
}

这将为 1000 次迭代打印一个更大的数字：

ELAPSED MILLIS: ~xxxx

请注意，counter.set(counter.get() + 1) 不是递增 AtomicInteger 的安全方式，删除该行并在 if 语句内递增：

if (counter.incrementAndGet() == UJPPHighTrafficClient.iterations)