Java网络编程——使用NIO实现非阻塞Socket通信



转自:https://blog.csdn.net/yanmei_yao/article/details/8586199

除了普通的Socket与ServerSocket实现的阻塞式通信外,java提供了非阻塞式通信的NIO API。先看一下NIO的实现原理。

 

从图中可以看出,服务器上所有Channel(包括ServerSocketChannel和SocketChannel)都需要向Selector注册,而该Selector则负责监视这些Socket的IO状态,当其中任意一个或者多个Channel具有可用的IO操作时,该Selector的select()方法将会返回大于0的整数,该整数值就表示该Selector上有多少个Channel具有可用的IO操作,并提供了selectedKeys()方法来返回这些Channel对应的SelectionKey集合。正是通过Selector,使得服务器端只需要不断地调用Selector实例的select()方法即可知道当前所有Channel是否有需要处理的IO操作。

看个demo


NClient.java

 

  1. import java.io.IOException;
  2. import java.net.InetSocketAddress;
  3. import java.nio.ByteBuffer;
  4. import java.nio.channels.Channel;
  5. import java.nio.channels.SelectionKey;
  6. import java.nio.channels.Selector;
  7. import java.nio.channels.ServerSocketChannel;
  8. import java.nio.channels.SocketChannel;
  9. import java.nio.charset.Charset;
  10. public class Server {
  11. public static void main(String[] args) {
  12. int port = 8082;
  13. int backlog = 3000;
  14. Charset charSet = Charset.forName(“UTF-8″);
  15. try {
  16. Selector selector = Selector.open();
  17. ServerSocketChannel channel = ServerSocketChannel.open();
  18. channel.bind(new InetSocketAddress(“localhost”,port), backlog);
  19. channel.configureBlocking(false);
  20. channel.register(selector, SelectionKey.OP_ACCEPT);
  21. while(selector.select()>0){
  22. for(SelectionKey sk:selector.selectedKeys()){
  23. //移除,TODO
  24. selector.selectedKeys().remove(sk);
  25. if(sk.isAcceptable()){
  26. ServerSocketChannel ssc = (ServerSocketChannel) sk.channel();
  27. //SocketChannel sc = channel.accept();
  28. SocketChannel sc = ssc.accept();
  29. sc.configureBlocking(false);
  30. sc.register(selector, SelectionKey.OP_READ);
  31. sk.interestOps(SelectionKey.OP_ACCEPT);
  32. }
  33. if(sk.isReadable()){
  34. SocketChannel saChannel = (SocketChannel) sk.channel();
  35. ByteBuffer buffer = ByteBuffer.allocate(1024);
  36. String content = “”;
  37. while(saChannel.read(buffer)>0){
  38. buffer.flip();
  39. content = content + charSet.decode(buffer);
  40. }
  41. System.out.println(“读取到的数据为:”+content);
  42. sk.interestOps(SelectionKey.OP_READ);
  43. if(content!=null&&!”".equals(content)){
  44. for(SelectionKey sko:selector.keys()){
  45. Channel targetChannel = sko.channel();
  46. if(targetChannel instanceof SocketChannel){
  47. SocketChannel tc = (SocketChannel) targetChannel;
  48. tc.write(charSet.encode(content));
  49. }
  50. }
  51. }
  52. }
  53. }
  54. }
  55. } catch (IOException e) {
  56. e.printStackTrace();
  57. }
  58. }
  59. }
  1. import java.io.IOException;
  2. import java.net.InetSocketAddress;
  3. import java.nio.ByteBuffer;
  4. import java.nio.channels.SelectionKey;
  5. import java.nio.channels.Selector;
  6. import java.nio.channels.SocketChannel;
  7. import java.nio.charset.Charset;
  8. import java.util.Scanner;
  9. public class NClient {
  10. //定义检测SocketChannel的Selector对象
  11. private Selector selector=null;
  12. //定义处理编码和解码的字符集
  13. private Charset charset=Charset.forName("UTF-8");
  14. //客户端SocketChannel
  15. private SocketChannel sc=null;
  16. public void init() throws IOException{
  17. selector=Selector.open();
  18. InetSocketAddress isa=new InetSocketAddress("127.0.0.1",30000);
  19. //调用open静态方法创建连接到指定主机的SocketChannel
  20. sc=SocketChannel.open(isa);
  21. //设置该sc以非阻塞方式工作
  22. sc.configureBlocking(false);
  23. //将Socketchannel对象注册到指定Selector
  24. sc.register(selector, SelectionKey.OP_READ);
  25. //启动读取服务器端数据的线程
  26. new ClientThread().start();
  27. //创建键盘输入流
  28. Scanner scan=new Scanner(System.in);
  29. while(scan.hasNextLine()){
  30. //读取键盘输入
  31. String line=scan.nextLine();
  32. //将键盘输入的内容输出到SocketChannel中
  33. sc.write(charset.encode(line));
  34. }
  35. }
  36. //定义读取服务器数据的线程
  37. private class ClientThread extends Thread{
  38. public void run(){
  39. try{
  40. while(selector.select()>0){
  41. //遍历每个有可用IO操作Channel对应的SelectionKey
  42. for(SelectionKey sk:selector.selectedKeys()){
  43. //删除正在处理的SelectionKey
  44. selector.selectedKeys().remove(sk);
  45. //如果该SelectionKey对应的Channel中有可读的数据
  46. if(sk.isReadable()){
  47. //使用NIO读取channel中的数据
  48. SocketChannel sc=(SocketChannel) sk.channel();
  49. ByteBuffer buff=ByteBuffer.allocate(1024);
  50. String content="";
  51. while(sc.read(buff)>0){
  52. //sc.read(buff);
  53. buff.flip();
  54. content+=charset.decode(buff);
  55. }
  56. //打印输出读取的内容
  57. System.out.println("聊天信息"+content);
  58. //为下一次读取做准备
  59. sk.interestOps(SelectionKey.OP_READ);
  60. }
  61. }
  62. }
  63. }catch(IOException ex){
  64. ex.printStackTrace();
  65. }
  66. }
  67. }
  68. public static void main(String[]args) throws IOException{
  69. new NClient().init();
  70. }
  71. }

 

NServer.java

  1. import java.io.IOException;
  2. import java.net.InetSocketAddress;
  3. import java.nio.ByteBuffer;
  4. import java.nio.channels.SelectionKey;
  5. import java.nio.channels.Selector;
  6. import java.nio.channels.SocketChannel;
  7. import java.nio.charset.Charset;
  8. import java.util.Scanner;
  9. public class Client {
  10. private static Charset charSet = Charset.forName(“UTF-8″);
  11. public static void main(String[] args) {
  12. int port = 8082;
  13. try {
  14. Selector selector = Selector.open();
  15. SocketChannel socketChannel = SocketChannel.open();
  16. socketChannel.connect(new InetSocketAddress(“localhost”, port));
  17. socketChannel.configureBlocking(false);
  18. socketChannel.register(selector, SelectionKey.OP_READ);
  19. Scanner scanner = new Scanner(System.in);
  20. new ReadMsg(selector).start();
  21. while(scanner.hasNextLine()){
  22. String line = scanner.nextLine();
  23. socketChannel.write(charSet.encode(line));
  24. }
  25. } catch (IOException e) {
  26. e.printStackTrace();
  27. }
  28. }
  29. }
  30. class ReadMsg extends Thread{
  31. private static Charset charSet = Charset.forName(“UTF-8″);
  32. private Selector selector;
  33. public ReadMsg(Selector selector){
  34. this.selector = selector;
  35. }
  36. @Override
  37. public void run() {
  38. try {
  39. while(selector.select()>0){
  40. for(SelectionKey sk:selector.selectedKeys()){
  41. selector.selectedKeys().remove(sk);
  42. if(sk.isConnectable()){
  43. SocketChannel channel = (SocketChannel) sk.channel();
  44. if(channel.isConnectionPending()){
  45. channel.finishConnect();
  46. }
  47. channel.configureBlocking(false);
  48. channel.register(selector, SelectionKey.OP_READ);
  49. sk.interestOps(SelectionKey.OP_READ);
  50. }
  51. if(sk.isReadable()){
  52. String content = “”;
  53. ByteBuffer buffer = ByteBuffer.allocate(1024);
  54. SocketChannel channel = (SocketChannel) sk.channel();
  55. while(channel.read(buffer)>0){
  56. buffer.flip();
  57. content = content+charSet.decode(buffer);
  58. }
  59. System.out.println(“接收到的信息:”+content);
  60. sk.interestOps(SelectionKey.OP_READ);
  61. }
  62. }
  63. }
  64. } catch (IOException e) {
  65. e.printStackTrace();
  66. }
  67. }
  68. }
  1. import java.io.IOException;
  2. import java.net.InetSocketAddress;
  3. import java.nio.ByteBuffer;
  4. import java.nio.channels.Channel;
  5. import java.nio.channels.SelectionKey;
  6. import java.nio.channels.Selector;
  7. import java.nio.channels.ServerSocketChannel;
  8. import java.nio.channels.SocketChannel;
  9. import java.nio.charset.Charset;
  10. public class NServer {
  11. //用于检测所有Channel状态的Selector
  12. private Selector selector=null;
  13. //定义实现编码、解码的字符集对象
  14. private Charset charset=Charset.forName("UTF-8");
  15. public void init() throws IOException{
  16. selector=Selector.open();
  17. //通过open方法来打开一个未绑定的ServerSocketChannel实例
  18. ServerSocketChannel server=ServerSocketChannel.open();
  19. InetSocketAddress isa=new InetSocketAddress("127.0.0.1",30000);
  20. //将该ServerSocketChannel绑定到指定ip地址
  21. server.socket().bind(isa);
  22. //设置ServerSocket以非阻塞方式工作
  23. server.configureBlocking(false);
  24. //将server注册到指定Selector对象
  25. server.register(selector, SelectionKey.OP_ACCEPT);
  26. while(selector.select()>0){
  27. //依次处理selector上的每个已选择的SelectionKey
  28. for(SelectionKey sk:selector.selectedKeys()){
  29. //从selector上的已选择Key集中删除正在处理的SelectionKey
  30. selector.selectedKeys().remove(sk);
  31. //如果sk对应的通信包含客户端的连接请求
  32. if(sk.isAcceptable()){
  33. //调用accept方法接受连接,产生服务器端对应的SocketChannel
  34. SocketChannel sc=server.accept();
  35. //设置采用非阻塞模式
  36. sc.configureBlocking(false);
  37. sc.register(selector, SelectionKey.OP_READ);
  38. //将sk对应的Channel设置成准备接受其他请求
  39. sk.interestOps(SelectionKey.OP_ACCEPT);
  40. }
  41. //如果sk对应的通道有数据需要读取
  42. if(sk.isReadable()){
  43. //获取该SelectionKey对应的Channel,该Channel中有可读的数据
  44. SocketChannel sc=(SocketChannel) sk.channel();
  45. //定义准备之星读取数据的ByteBuffer
  46. ByteBuffer buff=ByteBuffer.allocate(1024);
  47. String content="";
  48. //开始读取数据
  49. try{
  50. while(sc.read(buff)>0){
  51. buff.flip();
  52. content+=charset.decode(buff);
  53. }
  54. //打印从该sk对应的Channel里读到的数据
  55. System.out.println("=========="+content);
  56. //将sk对应的Channel设置成准备下一次读取
  57. sk.interestOps(SelectionKey.OP_READ);
  58. //如果捕捉到该sk对应的channel出现异常,即表明该channel对应的client出现了
  59. //异常,所以从selector中取消sk的注册
  60. }catch(IOException e){
  61. //从Selector中删除指定的SelectionKey
  62. sk.cancel();
  63. if(sk.channel()!=null){
  64. sk.channel().close();
  65. }
  66. }
  67. //如果content的长度大于0,即聊天信息不为空
  68. if(content.length()>0){
  69. //遍历该selector里注册的所有SelectKey
  70. for(SelectionKey key:selector.keys()){
  71. //选取该key对应的Channel
  72. Channel targetChannel=key.channel();
  73. //如果该channel是SocketChannel对象
  74. if(targetChannel instanceof SocketChannel){
  75. //将独到的内容写入该Channel中
  76. SocketChannel dest=(SocketChannel) targetChannel;
  77. dest.write(charset.encode(content));
  78. }
  79. }
  80. }
  81. }
  82. }
  83. }
  84. }
  85. public static void main(String[]args) throws IOException{
  86. new NServer().init();
  87. }
  88. }

通过java提供的NIO实现非阻塞Socket通信,大大提高了网络服务器的性能。