利用NIO建立Socket服务器

2023-07-27 18:23:16

传统的Java 的IO，利用Socket建立服务器，接收客户端连接，一般都是为每一个连接建立一个线程，如果连接数巨大，那么服务器开销也将巨大。。NIO的原理，可以参照图：http://new.51cto.com/files/uploadimg/20080912/150103487.jpg

利用NIO建立Socket服务器
Socket的Channel在Selector上注册某一种动作，Selector通过select操作，监视所有在该Selector注册过的Channel的对应的动作，如果监测到某一对应的动作，则返回selectedKeys，自己手动取到各个SelectionKey进行相应的处理。当然NIO不仅可以接受Socket的Channel，还有文件操作等其他IO操作。

作业的要求：

使用socket编程实现一个简单的文件服务器。客户端程序实现put功能(将一个文件从本地传到文件服务器)和get功能(从文件服务器取一远程文件存为本地文件)。客户端和文件服务器不在同一台机器上。
put [-h hostname] [-p portname] local_filename remote_filename
get [-h hostname] [-p portname] remote_filename local_filename

服务器端不使用nio，直接使用io的socket代码如下：

  1. import java.io.*;
  2. import java.net.ServerSocket;
  3. import java.net.Socket;
  4. public class ServerMain {
  5. public static void main(String[] args) {
  6. class SocketThread extends Thread{
  7. private Socket socket;
  8. private byte[] buf;
  9. private int len = 0;
  10. public SocketThread(Socket socket) {
  11. this.socket = socket;
  12. buf = new byte[1024];
  13. }
  14. @Override
  15. public void run() {
  16. try {
  17. DataInputStream dis = new DataInputStream(socket.getInputStream());
  18. DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
  19. //String command = dis.readUTF();
  20. len = dis.read(buf);
  21. String command = new String(buf,0,len);
  22. System.out.println("command=="+command);
  23. String[] temp =command.split(" ");
  24. command = temp[0];  //命令  是put还是get
  25. String filename = temp[1];  //文件名
  26. File file = new File("C:\\",filename);//假设放在C盘
  27. if(command.equals("get")){
  28. if(!file.exists()){
  29. //dos.writeUTF("notexists");
  30. dos.write("notexists".getBytes());
  31. dos.flush();
  32. System.out.println("没有这个文件，无法提供下载！");
  33. dis.close();
  34. dos.close();
  35. socket.close();
  36. return;
  37. }
  38. //dos.writeUTF("DownloadReady "+file.length());
  39. dos.write("准备下载".getBytes());
  40. dos.flush();
  41. System.out.println("正在接受文件下载...");
  42. DataInputStream fis = new DataInputStream(new BufferedInputStream(new FileInputStream(file)));
  43. while ((len = fis.read(buf))!= -1) {
  44. dos.write(buf, 0, len);
  45. }
  46. dos.flush();
  47. fis.close();
  48. System.out.println("文件传输完成");
  49. }
  50. else {
  51. //dos.writeUTF("UploadReady");
  52. dos.write("UploadReady".getBytes());
  53. dos.flush();
  54. System.out.println("正在接受文件上传...");
  55. DataOutputStream fileOut =
  56. new DataOutputStream(new BufferedOutputStream(new FileOutputStream(file)));
  57. while ((len = dis.read(buf))!=-1) {
  58. fileOut.write(buf, 0, len);
  59. }
  60. System.out.println("上传完毕！");
  61. fileOut.close();
  62. }
  63. dis.close();
  64. dos.close();
  65. socket.close();
  66. } catch (Exception e) {
  67. e.printStackTrace();
  68. }
  69. }
  70. }
  71. System.out.println("等待客户端连接....");
  72. int index = 0;
  73. try {
  74. ServerSocket server = new ServerSocket(9527,300); //端口号9527  允许最大连接数300
  75. while (true) {
  76. Socket socket = server.accept();
  77. System.out.println("收到第"+(++index)+"个连接");
  78. new SocketThread(socket).start(); //对每个连接创建一个线程
  79. }
  80. } catch (Exception e) {
  81. e.printStackTrace();
  82. }
  83. }
  84. }

使用NIO建立的Socket服务器，代码如下：

  1. import java.io.BufferedInputStream;
  2. import java.io.BufferedOutputStream;
  3. import java.io.DataInputStream;
  4. import java.io.DataOutputStream;
  5. import java.io.File;
  6. import java.io.FileInputStream;
  7. import java.io.FileOutputStream;
  8. import java.io.IOException;
  9. import java.net.InetSocketAddress;
  10. import java.nio.ByteBuffer;
  11. import java.nio.CharBuffer;
  12. import java.nio.channels.SelectionKey;
  13. import java.nio.channels.Selector;
  14. import java.nio.channels.ServerSocketChannel;
  15. import java.nio.channels.SocketChannel;
  16. import java.nio.charset.Charset;
  17. import java.nio.charset.CharsetDecoder;
  18. import java.nio.charset.CharsetEncoder;
  19. import java.util.Iterator;
  20. public class NewSocketServer {
  21. private static final int port = 9527;
  22. private Selector selector;
  23. private ByteBuffer clientBuffer = ByteBuffer.allocate(1024);
  24. private CharsetDecoder decoder = Charset.forName("GB2312").newDecoder();
  25. private CharsetEncoder encoder = Charset.forName("GB2312").newEncoder();
  26. //编码解码格式设置成GBK也行.UTF-8不行，中文乱码  （前提都是客户端没有设置任何编码解码格式）
  27. public void setListener() throws Exception{
  28. selector = Selector.open(); //打开选择器
  29. ServerSocketChannel server = ServerSocketChannel.open();  //定义一个 ServerSocketChannel通道
  30. server.socket().bind(new InetSocketAddress(port));  //ServerSocketChannel绑定端口
  31. server.configureBlocking(false);   //配置通道使用非阻塞模式
  32. server.register(selector, SelectionKey.OP_ACCEPT); //该通道在selector上注册  接受连接的动作
  33. while(true)
  34. {
  35. selector.select();   //select() 会阻塞，直到在该selector上注册的channel有对应的消息读入
  36. Iterator iter = selector.selectedKeys().iterator();
  37. while (iter.hasNext()) {
  38. SelectionKey key = (SelectionKey) iter.next();
  39. iter.remove();  // 删除此消息
  40. process(key);   // 当前线程内处理。（为了高效，一般会在另一个线程中处理此消息）
  41. }
  42. }
  43. }
  44. private void process(SelectionKey key) throws IOException {
  45. if (key.isAcceptable()) { // 接收请求
  46. ServerSocketChannel server = (ServerSocketChannel) key.channel();
  47. SocketChannel channel = server.accept();//类似于io的socket，ServerSocketChannel的accept函数返回 SocketChannel
  48. channel.configureBlocking(false);   //设置非阻塞模式
  49. SelectionKey sKey = channel.register(selector, SelectionKey.OP_READ);
  50. sKey.attach("read_command"); //这儿接收到连接请求之后可以为每个连接设置一个ID
  51. }
  52. else if (key.isReadable()) { // 读信息
  53. SocketChannel channel = (SocketChannel) key.channel();
  54. String name = (String) key.attachment();
  55. if(name.equals("read_command")){
  56. int count = channel.read(clientBuffer);
  57. if (count > 0) {
  58. clientBuffer.flip();
  59. CharBuffer charBuffer = decoder.decode(clientBuffer);
  60. String command = charBuffer.toString();
  61. //command形如：get abc.png 或者  put aaa.png
  62. System.out.println("command===="+command);  //得到客户端传来的命令
  63. String[] temp =command.split(" ");
  64. command = temp[0];  //命令  是put还是get
  65. String filename = temp[1];  //文件名
  66. SelectionKey sKey = channel.register(selector,SelectionKey.OP_WRITE);
  67. if(command.equals("put"))sKey.attach("UploadReady#"+filename);  //要保护该通道的文件名
  68. else if(command.equals("get")){
  69. if(!new File("C:\\",filename).exists()){ //假设文件都是在C盘根目录
  70. System.out.println("没有这个文件，无法提供下载！");
  71. sKey.attach("notexists");
  72. }
  73. else sKey.attach("DownloadReady#"+filename); //要保护该通道的文件名
  74. }
  75. } else {
  76. channel.close();
  77. }
  78. }
  79. else if(name.startsWith("read_file")){//这儿可以新开一个线程     文件操作也可以用NIO
  80. DataOutputStream fileOut =
  81. new DataOutputStream(
  82. new BufferedOutputStream(
  83. new FileOutputStream(
  84. new File("C:\\",name.split("#")[1]))));
  85. int passlen = channel.read(clientBuffer);
  86. while (passlen>=0) {
  87. clientBuffer.flip();
  88. fileOut.write(clientBuffer.array(), 0, passlen);
  89. passlen = channel.read(clientBuffer);
  90. }
  91. System.out.println("上传完毕！");
  92. fileOut.close();
  93. channel.close();
  94. }
  95. clientBuffer.clear();
  96. }
  97. else if (key.isWritable()) { // 写事件
  98. SocketChannel channel = (SocketChannel) key.channel();
  99. String flag = (String) key.attachment();
  100. if(flag.startsWith("downloading")){//这儿可以新开一个线程   文件操作也可以用NIO
  101. DataInputStream fis = new DataInputStream(
  102. new BufferedInputStream(
  103. new FileInputStream(
  104. new File("C:\\",flag.split("#")[1]))));
  105. byte[] buf = new byte[1024];
  106. int len =0;
  107. while ((len = fis.read(buf))!= -1) {
  108. channel.write(ByteBuffer.wrap(buf, 0, len));
  109. }
  110. fis.close();
  111. System.out.println("文件传输完成");
  112. channel.close();
  113. }
  114. else if(flag.equals("notexists")){
  115. //channel.write(encoder.encode(CharBuffer.wrap(flag)));
  116. channel.write(ByteBuffer.wrap(flag.getBytes())); //不用编码也行    客户端直接接收    中文也不是乱码
  117. channel.close();
  118. }
  119. else if(flag.startsWith("UploadReady")){
  120. channel.write(encoder.encode(CharBuffer.wrap("UploadReady")));
  121. //这儿如果不重新注册该通道的读操作    selector选择到该通道的将继续永远是写操作，也就无法跳转到上面的接受上传的处理
  122. SelectionKey sKey =channel.register(selector, SelectionKey.OP_READ);//register是覆盖的????!!!
  123. sKey.attach("read_file#"+flag.split("#")[1]);
  124. //key.attach("read_file#"+flag.split("#")[1]); //select不到读操作
  125. }
  126. else if(flag.startsWith("DownloadReady")){
  127. channel.write(ByteBuffer.wrap("准备下载".getBytes()));
  128. //channel.write(encoder.encode(CharBuffer.wrap("准备下载")));
  129. key.attach("downloading#"+flag.split("#")[1]);
  130. }
  131. }
  132. }
  133. public static void main(String[] args) {
  134. try {
  135. System.out.println("等待来至" + port + "端口的客户端连接.....");
  136. new NewSocketServer().setListener();
  137. } catch (Exception e) {
  138. e.printStackTrace();
  139. }
  140. }
  141. }

客户端代码如下：

  1. import java.io.*;
  2. import java.net.InetAddress;
  3. import java.net.Socket;
  4. import java.util.Scanner;
  5. public class ClientMain {
  6. private   int ServerPort = 9527;
  7. private   String ServerAddress = "192.168.1.154";
  8. private   String GetOrPut = "get";
  9. private   String local_filename = "";
  10. private   String remote_filename  = "";
  11. private   byte[] buf;
  12. private   int len;
  13. class SocketThread extends Thread{
  14. @Override
  15. public void run() {
  16. try {
  17. File file = new File("C:\\",local_filename); //假设文件放在C盘
  18. if(!file.exists()&&GetOrPut.equals("put")){
  19. System.out.println("本地没有这个文件，无法上传！");
  20. return;
  21. }
  22. InetAddress loalhost = InetAddress.getLocalHost();
  23. Socket socket = new Socket(ServerAddress,ServerPort,loalhost,44);
  24. //服务器IP地址  端口号   本机IP 本机端口号
  25. DataInputStream dis = new DataInputStream(socket.getInputStream());
  26. DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
  27. //dos.writeUTF(GetOrPut+" "+remote_filename);//服务器端如果是io的socket，writeUTF和writeUTF对接
  28. dos.write((GetOrPut+" "+remote_filename).getBytes());
  29. dos.flush();
  30. //String tempString = dis.writeUTF();
  31. buf = new byte[1024];
  32. len = dis.read(buf);
  33. String tempString = new String(buf,0,len);//服务器反馈的信息
  34. //System.out.println(tempString);
  35. if(tempString.equals("notexists")){
  36. System.out.println("服务器没有这个文件，无法下载！");
  37. dos.close();
  38. dis.close();
  39. socket.close();
  40. return;
  41. }
  42. if(tempString.startsWith("准备下载")){
  43. DataOutputStream fileOut =
  44. new DataOutputStream(new BufferedOutputStream(new FileOutputStream(file)));
  45. while ((len = dis.read(buf))!=-1) {
  46. fileOut.write(buf, 0, len);
  47. }
  48. System.out.println("下载完毕！");
  49. fileOut.close();
  50. dos.close();
  51. dis.close();
  52. socket.close();
  53. }
  54. else if(tempString.equals("UploadReady")){
  55. System.out.println("正在上传文件.......");
  56. DataInputStream fis = new DataInputStream(new BufferedInputStream(new FileInputStream(file)));
  57. while ((len = fis.read(buf))!= -1) {
  58. dos.write(buf, 0, len);
  59. }
  60. dos.flush();
  61. System.out.println("上传完毕！");
  62. fis.close();
  63. dis.close();
  64. dos.close();
  65. socket.close();
  66. }
  67. } catch (Exception e) {
  68. e.printStackTrace();
  69. }
  70. }
  71. }
  72. public boolean checkCommand(String command)
  73. {
  74. if(!command.startsWith("put")&&!command.startsWith("get")){
  75. System.out.println("输入命令错误");
  76. return false;
  77. }
  78. int index = -1;
  79. String temp = "";
  80. String[] tempStrings = null;
  81. if((index=command.indexOf("-h"))>0){
  82. temp = command.substring(index+3);
  83. temp = temp.substring(0, temp.indexOf(' '));
  84. ServerAddress = temp;
  85. }
  86. if((index=command.indexOf("-p"))>0){
  87. temp = command.substring(index+3);
  88. temp = temp.substring(0, temp.indexOf(' '));
  89. ServerPort = Integer.valueOf(temp);
  90. }
  91. tempStrings = command.split(" ");
  92. if(command.startsWith("put")){
  93. GetOrPut = "put";
  94. local_filename = tempStrings[tempStrings.length-2];
  95. remote_filename = tempStrings[tempStrings.length-1];
  96. }
  97. else if(command.startsWith("get")){
  98. GetOrPut = "get";
  99. local_filename = tempStrings[tempStrings.length-1];
  100. remote_filename = tempStrings[tempStrings.length-2];
  101. }
  102. return true;
  103. }
  104. public static void main(String[] args) {
  105. ClientMain thisC= new ClientMain();
  106. Scanner sc = new Scanner(System.in);
  107. String commandString = "";
  108. do {
  109. System.out.println("请输入命令：");
  110. commandString = sc.nextLine();
  111. } while (!thisC.checkCommand(commandString));
  112. ClientMain.SocketThread a = thisC.new SocketThread();
  113. a.start();
  114. }
  115. }
分享到： 利用NIO建立Socket服务器 利用NIO建立Socket服务器
评论
2 楼 lishuoying 2013-10-08  
读取命令有小小问题
  1. int count = channel.read(clientBuffer);
  2. if (count > 0) {
  3. //命令解释
  4. }

channel.read(clientBuffer) 有可能接收到命令字节的一部分，命令解释会出问题

1 楼 chinesejie 2012-11-07  
同学，您的程序有bug，我帮你调了几天了，终于搞定。
用nio写的server这段。
while (passlen>=0) {     
                        clientBuffer.flip();    
                        fileOut.write(clientBuffer.array(), 0, passlen);   
                        passlen = channel.read(clientBuffer);  
                    }

应嘎是这样的逻辑：
while (passlen>=0) {   
if（passlen!=0） {  
                        clientBuffer.flip(); 
}   
                        fileOut.write(clientBuffer.array(), 0, passlen);   
                        passlen = channel.read(clientBuffer);  
                    }    
这样传大文件也不都会错了。

上一篇：通过监控线程状态来保证socket服务器的稳定运行


下一篇：经过一年时间的沉淀 再次回首 TCP Socket服务器编程--转