nio 问题

Question

服务器是用nio写的
我的接受和发送端bytebuffer开的都是8192空间为什么在接收端把bytebuffer给分成两次并且一次是3838另一次是4380

Answer 1

int i=1;
     InputStream fis = null;
String filePrefix = new StringBuffer("file ").append(terminateTransID).append(" ").toString();

ByteBuffer w_buff = ByteBuffer.allocate(maxReadBytesEveryTime + filePrefix.getBytes().length); // 建立连接
InetSocketAddress targetAddress = new InetSocketAddress(toIP, TransferServerConfig.getPort());
SocketChannel socketChannel;
Selector selector;
try
{  
socketChannel = SocketChannel.open();
socketChannel.connect(targetAddress);
socketChannel.configureBlocking(false);
selector = Selector.open();
socketChannel.register(selector, SelectionKey.OP_WRITE); while (!socketChannel.finishConnect());
} catch (IOException e1)
{
// 传输文件流时建立Socket连接失败
runvar.put("目标IP", toIP);
runvar.put("传输服务器端口", String.valueOf(TransferServerConfig.getPort()));

throw new IDTException("7008", runvar);

} // 准备读取压缩文件的输入流
File zipFile = new File(zipFilePath);
long remainedFileLength = zipFile.length();
try
{
fis = new BufferedInputStream(new FileInputStream(zipFile));
} catch (FileNotFoundException e1)
{
} int bytesWritten = 0;
Calendar begin = Calendar.getInstance();
Calendar end;
long interval;
long limit; // 大文件内容分多次发送

while (remainedFileLength > 0)
{
// 读取压缩文件
try
{
bytesRead = fis.read(fileContent);
} catch (IOException e1)
{
// 传输文件流时文件流读写错误
throw new IDTException("7010");
}
remainedFileLength = remainedFileLength - bytesRead; // 发送文件内容报文
w_buff.clear();
w_buff.position(0);
w_buff.put(filePrefix.getBytes());
// 最后一次文件内容
if (bytesRead < maxReadBytesEveryTime)
{
byte[] lastFileContent = new byte[bytesRead];
System.arraycopy(fileContent, 0, lastFileContent, 0, bytesRead);
w_buff.put(lastFileContent);
} else
w_buff.put((fileContent)); w_buff.flip();
try
{   
System.out.println(i++ +"||"+w_buff.limit());
while (w_buff.hasRemaining()){
if(i%500==0){
try
{
Thread.sleep(5);
} catch (InterruptedException e)
{
e.printStackTrace();
}
}
socketChannel.write(w_buff);
}
} catch (IOException e)
{
// 传输文件流时写数据错误
runvar.put("目标IP", toIP);
runvar.put("传输服务器端口", String.valueOf(TransferServerConfig.getPort()));
runvar.put("传输数据", String.valueOf(w_buff));
throw new IDTException("7009", runvar);
}
w_buff.clear();
// 带宽限制
bytesWritten = bytesWritten + bytesRead;
if (bytesWritten > 2 * 1024 * 1024)
{
end = Calendar.getInstance();
interval = end.getTimeInMillis() - begin.getTimeInMillis();
limit = bytesWritten * 1000 / (4096 * 1024);
if (limit > interval)
{
try
{
Thread.sleep(limit - interval);
} catch (InterruptedException e)
{
// 执行带宽限制出错
throw new IDTException("7016");
}
}
bytesWritten = 0;
begin = Calendar.getInstance();
}
}
     try
{
fis.close();
} catch (IOException e)
{
e.printStackTrace();
}
这是发送端，已经测试过每次发送buffer都是满的最后一次一做处理了

Answer 2

public void run() {
int limit = 0;
byte[] messageBeginStr = new byte[9]; // 保存消息报文开始串
byte[] fileBeginStr = new byte[5]; // 保存文件报文开始串
byte[] digestStr = new byte[7];// 摘要 byte[] terminateTransID = new byte[20]; // 保存文件请求报文标识
byte[] digestContent = new byte[16];// 摘要内容
byte[] fileContent = new byte[maxReadBytesEveryTime];// 文件体 String fileNameStr;
String terminateTransIDStr; SocketChannel socketChannel;
String readContext = "";
ByteBuffer buffer = ByteBuffer.allocate(maxReadBytesEveryTime + 26);
SelectionKey selectionKey; try {
while (selector.select() > 0) {
// 依次处理selector上的每个已选择的SelectionKey
for (Iterator<SelectionKey> iter = selector.selectedKeys().iterator(); iter.hasNext();) {
selectionKey = (SelectionKey) iter.next(); // 从selector上的已选择Key集中删除正在处理的SelectionKey
iter.remove(); // 如果SelectionKey对应的通道包含客户端的连接请求
if (selectionKey.isAcceptable()) {
serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
// 调用accept方法接受连接，产生服务器端对应的SocketChannel
socketChannel = serverSocketChannel.accept();
// 设置采用非阻塞模式
socketChannel.configureBlocking(false);
// 将该SocketChannel也注册到selector
socketChannel.register(selector, SelectionKey.OP_READ);
} // 如果SelectionKey对应的通道有数据需要读取
if (selectionKey.isReadable()) { // 获取该SelectionKey对应的Channel,该Channel中有可读的数据
socketChannel = (SocketChannel) selectionKey.channel(); // 开始读取数据
try {
while (socketChannel.read(buffer) > 0) {
buffer.flip();
limit = buffer.limit();
buffer.position(0);
if (buffer.limit() >= 9) {// message类型报文
buffer.get(messageBeginStr, 0, 9);
}
buffer.position(0);
if (compare(messageBeginStr, messageBegin)) {
readContext = new String(charset.decode(buffer).array()).trim();
Controler c = new Controler();
c.process(readContext);
buffer.clear();
continue;
}
buffer.position(0);
if (buffer.limit() >= 5) {
buffer.get(fileBeginStr, 0, 5);// 文件头
buffer.position(0);
buffer.get(digestStr, 0, 7);// 摘要
}
if (compare(fileBeginStr, fileBegin)) {// 文件头
buffer.position(5);
buffer.get(terminateTransID, 0, 20);
terminateTransIDStr = new String(terminateTransID, "utf-8");
buffer.position(26);
buffer.get(fileContent, 0, limit - 26); //文件体
buffer.flip();
fileNameStr = this.getFileName(terminateTransIDStr);
receiver.saveFileStream(fileContent, new String(terminateTransID), fileNameStr, limit - 26);
} else if (compare(digestStr, digest)) {//摘要
buffer.position(7);
buffer.get(terminateTransID, 0, 20);
terminateTransIDStr = new String(terminateTransID, "utf-8");
fileNameStr = this.getFileName(terminateTransIDStr);
buffer.position(28);
buffer.get(digestContent, 0, limit - 28);
receiver.finishFileTransfer(terminateTransIDStr, fileNameStr, digestContent);
}
// 将SelectionKey对应的Channel设置成准备下一次读取
buffer.clear();
selectionKey.interestOps(SelectionKey.OP_READ);
}
} catch (IOException ex) {
if (selectionKey.channel() != null) {
selectionKey.cancel();
selectionKey.channel().close();
}
}
}
}
}
} catch (ClosedChannelException e) {
e.printStackTrace();
System.out.println("传输服务关闭连接失败!");
} catch (IOException e) {
e.printStackTrace();
System.out.println("传输服务读写错误!");
} catch (IDTException e) {
e.printStackTrace();
System.out.println("传输服务报文处理错误!");
}
}
这是服务器端,并不是每次都buffer都被拆分。