SocketChannel：为什么如果我快速写入一条消息，每条消息的延迟都很低，但是当我每 30 秒写入一条消息时，延迟却很高？

Question

现在，这个新问题明确地描述了这个问题的进展：为什么 JVM 在繁忙的旋转暂停后对同一代码块显示出更多的延迟？

我在下面提供了一个简单服务器和客户端的源代码，用于演示和隔离问题。基本上我正在计算乒乓（客户端-服务器-客户端）消息的延迟。我首先每 1 毫秒发送一条消息。我等待发送 200k 条消息，以便 HotSpot 有机会优化代码。然后我将暂停时间从 1 毫秒更改为 30 秒。令我惊讶的是，我的写入和读取操作变得相当慢。

我不认为这是 JIT/HotSpot 问题。我能够确定本机 JNI 调用 write ( write0) 和 read 的较慢方法。看起来停顿的时间越长，速度就越慢。

我正在寻找有关如何调试、理解、解释或解决此问题的指导。

服务器.java：

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

public class Server {

    private final ServerSocketChannel serverSocketChannel;
    private final ByteBuffer readBuffer = ByteBuffer.allocateDirect(1024);
    private final int port;
    private final int msgSize;

    public Server(int port, int msgSize) throws IOException {
        this.serverSocketChannel = ServerSocketChannel.open();
        this.port = port;
        this.msgSize = msgSize;
    }

    public void start() throws IOException {
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        final SocketChannel socketChannel = serverSocketChannel.accept(); // blocking mode...
        System.out.println("Client accepted!");
        socketChannel.configureBlocking(false);
        socketChannel.socket().setTcpNoDelay(true);
        Thread t = new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    while(true) {
                        int bytesRead = socketChannel.read(readBuffer);
                        if (bytesRead == -1) {
                            System.out.println("Client disconnected!");
                            return;
                        } else if (bytesRead > 0) {
                            if (readBuffer.position() == msgSize) {
                                // have a full message there...
                                readBuffer.flip();
                                int bytesSent = socketChannel.write(readBuffer);
                                if (bytesSent != msgSize) throw new RuntimeException("Could not send full message out: " + bytesSent);
                                readBuffer.clear();
                            }
                        }
                    }
                } catch(Exception e) {
                    throw new RuntimeException(e);
                }
            }
        });
        t.start();
        serverSocketChannel.close();
    }

    public static void main(String[] args) throws Exception {

        Server s = new Server(9999, 8);
        s.start();
    }
}

客户端.java:

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

public class Client implements Runnable {

    private static final int WARMUP = 200000;

    private final SocketChannel socketChannel;
    private final String host;
    private final int port;
    private final ByteBuffer outBuffer;
    private final ByteBuffer inBuffer = ByteBuffer.allocateDirect(1024);
    private final int msgSize;
    private final StringBuilder sb = new StringBuilder(1024);

    private int interval;
    private int totalMessagesSent;
    private long timeSent;
    private int mod;


    public Client(String host, int port, int msgSize) throws IOException {
        this.socketChannel = SocketChannel.open();
        this.host = host;
        this.port = port;
        this.outBuffer = ByteBuffer.allocateDirect(msgSize);
        this.msgSize = msgSize;
        for(int i = 0; i < msgSize; i++) outBuffer.put((byte) i);
        outBuffer.flip();
        this.interval = 1;
        this.mod = 20000;
    }

    public static long busySleep(long t) {
        long x = 0;
        for(int i = 0; i < t * 20000; i++) {
            x += System.currentTimeMillis() / System.nanoTime();
        }
        return x;
    }

    public void start() throws Exception {
        this.socketChannel.configureBlocking(false);
        this.socketChannel.socket().setTcpNoDelay(true);
        this.socketChannel.connect(new InetSocketAddress(host, port));

        while(!socketChannel.finishConnect()) {
            System.out.println("Waiting to connect");
            Thread.sleep(1000);
        }
        System.out.println("Please wait as output will appear every minute or so. After " + WARMUP + " messages you will see the problem.");
        Thread t = new Thread(this);
        t.start();
    }

    private final void printResults(long latency, long timeToWrite, long timeToRead, long zeroReads, long partialReads, long realRead) {
        sb.setLength(0);
        sb.append(new java.util.Date().toString());
        sb.append(" Results: totalMessagesSent=").append(totalMessagesSent);
        sb.append(" currInterval=").append(interval);
        sb.append(" latency=").append(latency);
        sb.append(" timeToWrite=").append(timeToWrite);
        sb.append(" timeToRead=").append(timeToRead);
        sb.append(" realRead=").append(realRead);
        sb.append(" zeroReads=").append(zeroReads);
        sb.append(" partialReads=").append(partialReads);
        System.out.println(sb);
    }

    @Override
    public void run() {

        try {

            while(true) {

                busySleep(interval);

                outBuffer.position(0);

                timeSent = System.nanoTime();

                int bytesSent = socketChannel.write(outBuffer);
                long timeToWrite = System.nanoTime() - timeSent;
                if (bytesSent != msgSize) throw new IOException("Can't write message: " + bytesSent);

                inBuffer.clear();
                long zeroReads = 0;
                long partialReads = 0;
                long timeToRead = System.nanoTime();
                long realRead = 0;
                while(inBuffer.position() != msgSize) {
                    realRead = System.nanoTime();
                    int bytesRead = socketChannel.read(inBuffer);
                    if (bytesRead == 0) {
                        zeroReads++;
                    } else if (bytesRead == -1) {
                        System.out.println("Other side disconnected!");
                        return;
                    } else if (bytesRead != msgSize) {
                        partialReads++;
                        realRead = -1;
                    } else {
                        realRead = System.nanoTime() - realRead;
                    }
                }

                long now = System.nanoTime();

                timeToRead = now - timeToRead;

                long latency = now - timeSent;

                if (++totalMessagesSent % mod == 0 || totalMessagesSent == 1) {
                    printResults(latency, timeToWrite, timeToRead, zeroReads, partialReads, realRead);
                }

                if (totalMessagesSent == WARMUP) {
                    this.interval = 30000;
                    this.mod = 1;
                }
            }

        } catch(Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws Exception {

        Client client = new Client("localhost", 9999, 8);
        client.start();
    }
}

我执行java -server -cp . Server和java -server -cp . Client. 客户端的输出是：

---

根据@dunni 的请求，将延迟 30 秒更改为 1 秒。同样的问题：

Answer 1

您遇到的一个问题是，当没有数据可供读取时，JVM、CPU 及其缓存就会进入休眠状态。一旦发生这种情况，机器在获取数据之前必须比问题严重时执行更多的操作。

• CPU 速度可能会降低以节省电量。例如一半正常。它可以在愚蠢的繁忙循环中执行此操作。
• 该线程未运行，必须在新 CPU 上重新启动。（在你的情况下，这应该很少见）
• CPU 的缓存可能已断电，必须从 L3 缓存或主内存逐步加载
• 即使在线程返回后，由于缓存会拉入更多数据/代码，它的运行速度也会比正常情况慢最多 100 微秒。
• 您将每秒收到 100 多次不可屏蔽中断，并且无法关闭。

简而言之，如果您需要一致的延迟，您需要

• 关闭电源管理。
• 不要放弃CPU，即忙等待。（你正在做什么）
• 在隔离的 CPU 上运行，以亲和力绑定线程。
• 禁用该内核上的所有可屏蔽中断。
• 使用用户空间驱动程序而不是内核进行网络连接。

注意：鉴于每个操作似乎都需要大约 2 倍的时间，我会首先考虑电源管理。