Java NIO:allocate或allocateDirect，当我每次写入SocketChannel时创建ByteBuffer时
我有一个每600毫秒运行一次的游戏服务器，在一个循环中操作一个字节数组，然后在循环结束时将字节数组写入客户端
由于在循环结束时需要写入多少字节的性质不确定，我为循环结束时的每次写入创建了一个字节缓冲区。我不确定的是，它是否会更快地直接进行
从我从其他类似问题中收集到的信息来看，direct可能更好，因为当缓冲区写入SocketChannel时，如果它还不是direct，则可能会创建direct字节缓冲区。再澄清一下就好了
如果我的问题不够清楚，以下是我的网络背后的代码： 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());
}
}
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;
}
}
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);
}
}
# 1 楼答案
读取直接缓冲区可以避免额外的拷贝/本机内存分配，从而加快速度并减少CPU周期。但是，如果您想要处理读取到直接缓冲区的数据，那么无论如何都需要将其复制到
byte[]读入堆缓冲区涉及读入临时直接缓冲区，然后将其内容复制回堆。直接缓冲区和堆缓冲区之间的区别在于，直接缓冲区的内容
malloc已加密（即脱离Java堆）。堆缓冲区反过来由byte[]支持，因此在不复制它的情况下，您无法获取指向其内容的原始指针以执行IO（GC可以跨堆移动对象，从而导致堆损坏）尽管JNI中有一个“关键区域”，您可以在其中获取原始指针，而无需进行复制，但它对GC有影响，因为对象被固定到其当前位置，或者GC被完全禁用。我刚才还在试验它Measuring performance of java.io.InputStream