Java NIO : allocate or allocateDirect when creating a ByteBuffer every time I write to a SocketChannel

Question

我有一个每 600 毫秒运行一次的游戏服务器，在一个周期内操作一个字节数组，然后在周期结束时将字节数组写入客户端。

由于在循环结束时需要写入多少字节的不确定性，我在循环结束时为每次写入创建了一个字节缓冲区。我不确定的是直接还是不直接是否会更快。

根据我从其他类似问题中收集到的信息，直接可能会更好，因为如果缓冲区尚未直接写入 SocketChannel，则无论如何都可以创建直接字节缓冲区。任何更多的澄清会很好。

如果我的问题不够清楚，下面是我的网络背后的代码: https://pastebin.com/M9wm88BA

package com.palidino.nio;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class NioServer implements Runnable {
    private List<Session> sessions = new ArrayList<>();
    private Map<String, Integer> connectionCounts = new HashMap<>();
    private InetSocketAddress hostAddress;
    private ServerSocketChannel serverSocketChannel;
    private Selector selector;
    private SessionHandler sessionHandler;
    private boolean running;

    private ByteBuffer readBuffer;
    private byte[] readBytes;

    private int sessionIdleTimeout;
    private int maxConnectionsPerIPAddress;
    private int socketBufferSize = 16384;

    public NioServer() throws IOException {
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
    }

    public void start(String remoteAddress, int port) throws IOException {
        if (hostAddress != null) {
            throw new IllegalStateException("Server already started");
        }
        if (sessionHandler == null) {
            throw new IllegalStateException("SsessionHandler can't be null");
        }
        readBuffer = ByteBuffer.allocateDirect(socketBufferSize);
        readBytes = new byte[socketBufferSize];
        hostAddress = new InetSocketAddress(remoteAddress, port);
        serverSocketChannel.socket().setReceiveBufferSize(socketBufferSize);
        serverSocketChannel.socket().bind(hostAddress);
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("Starting server on " + remoteAddress + ":" + port);
        new Thread(this, "NioServer").start();
    }

    public void stop() {
        try {
            if (serverSocketChannel != null) {
                serverSocketChannel.close();
                serverSocketChannel = null;
            }
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
    }

    public void setSessionHandler(SessionHandler sessionHandler) {
        this.sessionHandler = sessionHandler;
    }

    public void setSessionIdleTimeout(int seconds) {
        if (hostAddress != null) {
            throw new IllegalStateException("Server already started");
        }
        if (seconds <= 0) {
            throw new IllegalArgumentException("seconds must be greater than 0");
        }
        sessionIdleTimeout = seconds * 1000;
    }

    public void setMaxConnectionsPerIPAddress(int maxConnectionsPerIPAddress) {
        if (hostAddress != null) {
            throw new IllegalStateException("Server already started");
        }
        if (maxConnectionsPerIPAddress <= 0) {
            throw new IllegalArgumentException("maxConnectionsPerIPAddress must be greater than 0");
        }
        this.maxConnectionsPerIPAddress = maxConnectionsPerIPAddress;
    }

    public void setSocketBufferSize(int socketBufferSize) throws IOException {
        if (hostAddress != null) {
            throw new IllegalStateException("Server already started");
        }
        if (socketBufferSize <= 0) {
            throw new IllegalArgumentException("size must be greater than 0");
        }
        this.socketBufferSize = socketBufferSize;
    }

    @Override
    public void run() {
        if (running) {
            throw new IllegalStateException("Server is already running");
        }
        running = true;
        while (serverSocketChannel.isOpen()) {
            cycle();
        }
        running = false;
    }

    private void cycle() {
        try {
            selector.select();
            for (Iterator<SelectionKey> it = selector.selectedKeys().iterator(); it.hasNext();) {
                SelectionKey selectionKey = it.next();
                it.remove();
                Session session = null;
                try {
                    if (serverSocketChannel == null || !serverSocketChannel.isOpen()) {
                        break;
                    }
                    session = (Session) selectionKey.attachment();
                    if (selectionKey.isValid() && selectionKey.isAcceptable()) {
                        session = accept(selectionKey);
                    }
                    if (session == null) {
                        continue;
                    }
                    if (selectionKey.isValid() && selectionKey.isReadable()) {
                        read(selectionKey);
                    }
                    if (selectionKey.isValid() && selectionKey.isWritable()) {
                        write(selectionKey);
                    }
                } catch (Exception e2) {
                    error(e2, session);
                }
            }
            checkSessions();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private Session accept(SelectionKey selectionKey) throws IOException {
        Session session = null;
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
        SocketChannel socketChannel = serverSocketChannel.accept();
        socketChannel.socket().setSendBufferSize(socketBufferSize);
        socketChannel.configureBlocking(false);
        String remoteAddress = socketChannel.socket().getInetAddress().getHostAddress();
        int connectionCount = getConnectionCount(remoteAddress);
        if (maxConnectionsPerIPAddress > 0 && connectionCount >= maxConnectionsPerIPAddress) {
            socketChannel.close();
        } else {
            connectionCounts.put(remoteAddress, connectionCount + 1);
            session = new Session(socketChannel, remoteAddress, socketChannel.register(selector, SelectionKey.OP_READ));
            sessionHandler.accept(session);
            sessions.add(session);
        }
        return session;
    }

    private void read(SelectionKey selectionKey) throws IOException {
        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
        if (!socketChannel.isOpen()) {
            return;
        }
        Session session = (Session) selectionKey.attachment();
        readBuffer.clear();
        int numberBytesRead;
        ByteArrayOutputStream readStream = new ByteArrayOutputStream();
        while ((numberBytesRead = socketChannel.read(readBuffer)) > 0) {
            readBuffer.flip();
            readBuffer.get(readBytes, 0, numberBytesRead);
            readStream.write(readBytes, 0, numberBytesRead);
            readBuffer.clear();
            session.updateLastRead();
        }
        if (readStream.size() > 0) {
            sessionHandler.read(session, readStream.toByteArray());
        }
        if (numberBytesRead == -1) {
            session.close();
        }
    }

    private void write(SelectionKey selectionKey) throws IOException {
        SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
        if (!socketChannel.isOpen()) {
            return;
        }
        Session session = (Session) selectionKey.attachment();
        if (session.getWriteEvents().isEmpty()) {
            return;
        }
        try {
            while (!session.getWriteEvents().isEmpty()) {
                WriteEvent writeEvent = session.getWriteEvents().peek();
                socketChannel.write(writeEvent.getBuffer());
                if (writeEvent.getBuffer().remaining() > 0) {
                    break;
                }
                if (writeEvent.getHandler() != null) {
                    writeEvent.getHandler().complete(session, true);
                }
                session.getWriteEvents().poll();
            }
        } catch (Exception e) {
            error(e, session);
        }
        if (selectionKey.isValid() && session.getWriteEvents().isEmpty()) {
            selectionKey.interestOps(SelectionKey.OP_READ);
        }
    }

    private void error(Exception exception, Session session) throws IOException {
        try {
            sessionHandler.error(exception, session);
        } catch (Exception e) {
            if (session != null) {
                session.close();
            }
            e.printStackTrace();
        }
    }

    private void checkSessions() {
        if (sessions.isEmpty()) {
            return;
        }
        for (Iterator<Session> it = sessions.iterator(); it.hasNext();) {
            Session session = it.next();
            SelectionKey selectionKey = session.getSelectionKey();
            if (selectionKey.isValid() && !session.getWriteEvents().isEmpty()) {
                selectionKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
            }
            if (session.idleTimeout(sessionIdleTimeout)) {
                session.close();
            }
            if (session.isOpen()) {
                continue;
            }
            String remoteAddress = session.getRemoteAddress();
            int connectionCount = getConnectionCount(remoteAddress);
            if (connectionCount > 1) {
                connectionCounts.put(remoteAddress, connectionCount - 1);
            } else {
                connectionCounts.remove(remoteAddress);
            }
            if (sessionHandler != null) {
                sessionHandler.closed(session);
            }
            if (selectionKey.isValid()) {
                selectionKey.cancel();
            }
            while (!session.getWriteEvents().isEmpty()) {
                WriteEvent writeEvent = session.getWriteEvents().poll();
                if (writeEvent.getHandler() != null) {
                    writeEvent.getHandler().complete(session, false);
                }
            }
            it.remove();
        }
    }

    private int getConnectionCount(String remoteAddress) {
        return connectionCounts.containsKey(remoteAddress) ? connectionCounts.get(remoteAddress) : 0;
    }

    public void printStats() {
        System.out
                .println("NIOServer: sessions: " + sessions.size() + "; connectionCounts: " + connectionCounts.size());
    }
}

https://pastebin.com/TxPQN7JZ

package com.palidino.nio;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

public class Session {
    private SocketChannel socketChannel;
    private SelectionKey selectionKey;
    private String remoteAddress;
    private long lastRead;
    private Queue<WriteEvent> writeEvents = new ConcurrentLinkedQueue<>();
    private Object attachment;

    public Session(SocketChannel socketChannel, String remoteAddress, SelectionKey selectionKey) {
        this.socketChannel = socketChannel;
        this.remoteAddress = remoteAddress;
        this.selectionKey = selectionKey;
        selectionKey.attach(this);
        updateLastRead();
    }

    public void write(byte[] bytes) {
        write(bytes, 0, bytes.length, null);
    }

    public void write(byte[] bytes, WriteEventHandler handler) {
        write(bytes, 0, bytes.length, handler);
    }

    public void write(byte[] bytes, int offset, int length) {
        write(bytes, 0, bytes.length, null);
    }

    public void write(byte[] bytes, int offset, int length, WriteEventHandler handler) {
        addWriteEvent(new WriteEvent(bytes, offset, length, handler));
    }

    public void write(ByteBuffer buffer) {
        write(buffer, null);
    }

    public void write(ByteBuffer buffer, WriteEventHandler handler) {
        addWriteEvent(new WriteEvent(buffer, handler));
    }

    private void addWriteEvent(WriteEvent writeEvent) {
        writeEvents.offer(writeEvent);
        if (selectionKey.isValid()) {
            selectionKey.selector().wakeup();
        }
    }

    public void close() {
        try {
            socketChannel.close();
        } catch (IOException ioe) {
        }
    }

    public boolean isOpen() {
        return socketChannel.isOpen();
    }

    public SocketChannel getSocketChannel() {
        return socketChannel;
    }

    public String getRemoteAddress() {
        return remoteAddress;
    }

    public long getLastRead() {
        return lastRead;
    }

    public boolean idleTimeout(int timeoutMillis) {
        return timeoutMillis > 0 && System.currentTimeMillis() - lastRead > timeoutMillis;
    }

    public void setAttachment(Object attachment) {
        this.attachment = attachment;
    }

    public Object getAttachment() {
        return attachment;
    }

    SelectionKey getSelectionKey() {
        return selectionKey;
    }

    void updateLastRead() {
        lastRead = System.currentTimeMillis();
    }

    Queue<WriteEvent> getWriteEvents() {
        return writeEvents;
    }
}

https://pastebin.com/r37vPUtJ

package com.palidino.nio;

import java.nio.ByteBuffer;

public class WriteEvent {
    private ByteBuffer buffer;
    private WriteEventHandler handler;

    public WriteEvent(byte[] original, int offset, int length, WriteEventHandler handler) {
        if (original == null) {
            throw new NullPointerException("original can't be null");
        }
        if (offset < 0 || length < 0) {
            throw new NegativeArraySizeException("offset and length must be greater than 0");
        }
        if (offset > original.length || length + offset > original.length) {
            throw new ArrayIndexOutOfBoundsException("length + offset can't be greater than original.length");
        }
        if (original.length == 0 || length == 0) {
            throw new IllegalArgumentException("length must be greater than 0");
        }
        buffer = ByteBuffer.allocateDirect(length);
        buffer.put(original, offset, length);
        buffer.flip();
        buffer = buffer.asReadOnlyBuffer();
        this.handler = handler;
    }

    public WriteEvent(ByteBuffer original, WriteEventHandler handler) {
        buffer = ByteBuffer.allocateDirect(original.capacity());
        ByteBuffer readOnlyCopy = original.asReadOnlyBuffer();
        readOnlyCopy.flip();
        buffer.put(readOnlyCopy);
        buffer.flip();
        buffer = buffer.asReadOnlyBuffer();
        this.handler = handler;
    }

    ByteBuffer getBuffer() {
        return buffer;
    }

    WriteEventHandler getHandler() {
        return handler;
    }
}

或者为每次写入创建一个直接缓冲区，在我的选择器线程中有一个直接缓冲区，并将其大小调整到 fromBuffer 在写入时很少有限制的位置。

private void write(SelectionKey selectionKey) throws IOException {
    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
    if (!socketChannel.isOpen()) {
        return;
    }
    Session session = (Session) selectionKey.attachment();
    if (session.getWriteEvents().isEmpty()) {
        return;
    }
    try {
        while (!session.getWriteEvents().isEmpty()) {
            WriteEvent writeEvent = session.getWriteEvents().peek();
            ByteBuffer fromBuffer = writeEvent.getBuffer();
            int position = fromBuffer.position();
            int limit = fromBuffer.limit();
            try {
                do {
                    directBuffer.clear();
                    if (fromBuffer.remaining() > directBuffer.remaining()) {
                        fromBuffer.limit(fromBuffer.remaining() - directBuffer.remaining());
                    }
                    directBuffer.put(fromBuffer);
                    directBuffer.flip();
                    position += socketChannel.write(directBuffer);
                    fromBuffer.position(position);
                    fromBuffer.limit(limit);
                } while (directBuffer.remaining() == 0 && fromBuffer.remaining() > 0);
                if (fromBuffer.remaining() > 0) {
                    break;
                }
                if (writeEvent.getHandler() != null) {
                    writeEvent.getHandler().complete(session, true);
                }
                session.getWriteEvents().poll();
            } catch (Exception e2) {
                fromBuffer.position(position);
                fromBuffer.limit(limit);
                throw e2;
            }
        }
    } catch (Exception e) {
        error(e, session);
    }
    if (selectionKey.isValid() && session.getWriteEvents().isEmpty()) {
        selectionKey.interestOps(SelectionKey.OP_READ);
    }
}

Answer 1

What I'm uncertain of is if it would be faster making it direct or not.

读入直接缓冲区可以避免额外的复制/ native 内存分配，从而使其更快并消耗更少的 CPU 周期。但是，如果您想处理读取到直接缓冲区的数据，您无论如何都需要将其复制到 byte[] 中。

读入堆缓冲区涉及读入临时直接缓冲区，然后将其内容复制回堆中。直接缓冲区和堆缓冲区之间的区别在于，直接缓冲区的内容经过malloc处理(即脱离 Java 堆)。堆缓冲区反过来由 byte[] 支持，因此您无法获得指向其内容的原始指针以在不复制它的情况下执行 IO(GC 可以在堆中移动对象导致堆损坏)

尽管在 JNI 中有一个“临界区”，您可以在其中获取原始指针而无需制作副本，但它对 GC 有影响，因为对象要么被固定到它的当前位置，要么 GC 被完全禁用。我刚才正在试验它 Measuring performance of java.io.InputStream