java 简单聊天室

1330阅读　0评论2018-05-07　nuoyazhou110
分类：Java

使用同步非阻塞模型NIO.
主线程负责监听端口、accept连接、read操作.
启动另外一个线程负责消息的write操作.

采用包长+包体组装协议.
LinkedBlockingQueue 存储待广播的消息内容.


服务端代码：

点击(此处)折叠或打开

  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.ServerSocketChannel;
  7. import java.nio.channels.SocketChannel;
  8. import java.util.Iterator;
  9. import java.util.LinkedList;
  10. import java.util.List;
  11. import java.util.Set;
  12. import java.util.concurrent.BlockingQueue;
  13. import java.util.concurrent.LinkedBlockingQueue;


  16. public class ServerSocketDemo2 {
  17.     
  18.     private static BlockingQueue<String> bq = new LinkedBlockingQueue<>();
  19.     
  20.     private static List<SocketChannel> listSc = new LinkedList<SocketChannel>();

  22.     public static void main(String[] args) throws Exception {
  23.         
  24.         //1.该线程负责 写socket
  25.         //2.主线程维连接和读socket
  26.         new Thread(new Runnable() {
  27.             @Override
  28.             public void run() {
  29.                 System.out.println("该线程负责写入socket");
  30.                 //该线程负责写操作
  31.                 String msg = null;
  32.                 while (true ) {
  33.                     try {
  34.                         //此处会阻塞
  35.                         msg = bq.take();
  36.                     } catch (InterruptedException e1) {
  37.                         e1.printStackTrace();
  38.                     }
  39.                     System.out.println(System.currentTimeMillis());
  40.                     System.out.println("接收到数据"+msg);
  41.                 
  42.                     Iterator<SocketChannel> it = listSc.iterator();
  43.                     while(it.hasNext()) {
  44.                      SocketChannel sc = it.next();
  45.                         if(!sc.isConnected()) {
  46.                             continue;
  47.                         }
  48.                         
  49.                         //先写入包长度
  50.                         ByteBuffer headwb = ByteBuffer.allocate(4);
  51.                         headwb.putInt(msg.getBytes().length);
  52.                         //重绕缓冲区 否则下面无法进行write操作
  53.                         headwb.flip();
  54.                         while(headwb.hasRemaining()) {
  55.                             try {
  56.                                 sc.write(headwb);
  57.                             } catch (IOException e) {
  58.                                 e.printStackTrace();
  59.                             }
  60.                         }
  61.                         //再写入包内容
  62.                         ByteBuffer bodywb = ByteBuffer.wrap(msg.getBytes());
  63.                         while(bodywb.hasRemaining()) {
  64.                             try {
  65.                                 sc.write(bodywb);
  66.                                 System.out.println("发送完成");
  67.                             } catch (IOException e) {
  68.                                 e.printStackTrace();
  69.                             }
  70.                         }
  71.                     }
  72.                 }
  73.             }
  74.         }).start();
  75.         //1. 创建一个选择器
  76.         Selector s = Selector.open();
  77.         //2.创建一个服务端channel
  78.         ServerSocketChannel ssc = ServerSocketChannel.open();
  79.         //3.置为非阻塞IO
  80.         ssc.configureBlocking(false);
  81.         ssc.bind(new InetSocketAddress("127.0.0.1", 44446));
  82.         
  83.         //4.将当前channel 的accept read write操作注册到 选择器中
  84.         ssc.register(s , SelectionKey.OP_ACCEPT);
  85.         //5循环接收事件
  86.         while(true) {
  87.             s.select();//该方法为阻塞方法，如果没有事件唤醒就不会执行下边的代码
  88.             //遍历已经唤醒的事件列表
  89.             Set<SelectionKey> list = s.selectedKeys();
  90.             Iterator<SelectionKey> it = list.iterator();
  91.             while (it.hasNext()) {
  92.                 SelectionKey sk = it.next();
  93.                 if( sk.isAcceptable() == true ) {
  94.                     //获取channel 强制转换
  95.                     ServerSocketChannel wsc = (ServerSocketChannel)sk.channel();
  96.                     //设置为非阻塞模式 所以accepc操作不一定有返回值 需循环操作只到有返回值为止
  97.                     SocketChannel sc = null;
  98.                     while ( sc == null ) {
  99.                         sc = wsc.accept();
  100.                     }
  101.                     //将该连接socket置为非阻塞状态
  102.                     sc.configureBlocking(false);
  103.                     //当前socket同时注册 读和写监控
  104.                     sc.register(s,SelectionKey.OP_READ );
  105.                     

  107.                     listSc.add(sc);
  108.                     
  109.                 } else if(sk.isReadable() == true) {
  110.                     SocketChannel sc = (SocketChannel)sk.channel();
  111.                     //从socket读取数据写入缓冲区 先读整型的包长
  112.                     ByteBuffer ib = ByteBuffer.allocate(4);
  113.                     while(ib.hasRemaining()) {
  114.                         sc.read(ib);
  115.                     }
  116.                     //重绕缓冲区 否则 下面的getInt不能获取准确的长度
  117.                     ib.flip();
  118.                     int len = ib.getInt();
  119.                     ByteBuffer contentBuffer = ByteBuffer.allocate(len);
  120.                     while(contentBuffer.hasRemaining()) {
  121.                         sc.read(contentBuffer);
  122.                     }
  123.                     
  124.                     //写入队列
  125.                     bq.put(new String(contentBuffer.array()));
  126.                     
  127.                 } else {
  128.                     throw new Exception("未知操作类型！");
  129.                 }
  130.                 it.remove();
  131.             }
  132.         }
  133.     }

  135. }

客户端代码：

点击(此处)折叠或打开

  1. import java.net.InetSocketAddress;
  2. import java.nio.ByteBuffer;
  3. import java.nio.channels.SelectionKey;
  4. import java.nio.channels.Selector;
  5. import java.nio.channels.SocketChannel;
  6. import java.util.Iterator;
  7. import java.util.Set;

  9. public class ClientSocketDemo {

  11.     public static void main(String[] args) throws Exception {
  12.         //1. 创建一个选择器
  13.         Selector s = Selector.open();
  14.         //2 创建channel
  15.         SocketChannel sc = SocketChannel.open();
  16.         sc.configureBlocking(false);
  17.         sc.connect(new InetSocketAddress("127.0.0.1",44446));
  18.         sc.register(s, SelectionKey.OP_CONNECT);
  19.         while( true ) {
  20.             s.select();
  21.             //遍历已经唤醒的事件列表
  22.             Set<SelectionKey> list = s.selectedKeys();
  23.             Iterator<SelectionKey> it = list.iterator();
  24.             while (it.hasNext()) {
  25.                 SelectionKey sk = it.next();
  26.                 if(sk.isReadable() == true) {
  27.                     //获取channel
  28.                     SocketChannel readsc = (SocketChannel)sk.channel();
  29.                     //从socket读取数据写入缓冲区 先读整型的包长
  30.                     ByteBuffer ib = ByteBuffer.allocate(4);
  31.                     while(ib.hasRemaining()) {
  32.                         readsc.read(ib);
  33.                     }
  34.                     //重绕缓冲区 否则 下面的getInt不能获取准确的长度
  35.                     ib.flip();
  36.                     int len = ib.getInt();
  37.                     ByteBuffer contentBuffer = ByteBuffer.allocate(len);
  38.                     while(contentBuffer.hasRemaining()) {
  39.                         readsc.read(contentBuffer);
  40.                     }
  41.                     System.out.println(new String(contentBuffer.array()));
  42.                 } else if(sk.isWritable()) {
  43.                     //获取channel
  44.                     SocketChannel writesc = (SocketChannel)sk.channel();
  45.                     
  46.                     //返回字符串
  47.                     String rstr = "hellow world";
  48.                     //先写入包长度
  49.                     ByteBuffer headwb = ByteBuffer.allocate(4);
  50.                     headwb.putInt(rstr.getBytes().length);
  51.                     //重绕缓冲区 否则下面无法进行write操作
  52.                     headwb.flip();
  53.                     while(headwb.hasRemaining()) {
  54.                         writesc.write(headwb);
  55.                     }
  56.                     //再写入包内容
  57.                     ByteBuffer bodywb = ByteBuffer.wrap(rstr.getBytes());
  58.                     while(bodywb.hasRemaining()) {
  59.                         writesc.write(bodywb);
  60.                     }
  61.                     //注册当前通道上的读事件 并取消掉写事件
  62.                     writesc.register(s,sk.interestOps() & ~ SelectionKey.OP_WRITE );
  63.                 } else if(sk.isConnectable()) {
  64.                     //获取channel
  65.                     SocketChannel sc1 = (SocketChannel)sk.channel();
  66.                     while(!sc1.isConnected()) {
  67.                         sc1.finishConnect();
  68.                     }
  69.                     sc1.register(s, SelectionKey.OP_WRITE | SelectionKey.OP_READ);
  70.                 } else {
  71.                     throw new Exception("未知操作类型！");
  72.                 }
  73.                 it.remove();
  74.             }
  75.         }
  76.     }

  78. }


上一篇：JAVA多线程文件拷贝
下一篇：zookeeper安装及命令行操作