在网络的初期网民很少服务器完全无压力那时的技术也没有现在先进通常用一个线程来全程跟踪处理一个请求。因为这样最简单。其实代码实现大家都知道就是服务器上有个ServerSocket在某个端口监听接收到客户端的连接后会创建一个Socket并把它交给一个线程进行后续处理。线程主要从Socket读取客户端传过来的数据然后进行业务处理并把结果再写入Socket传回客户端。由于网络的原因Socket创建后并不一定能立刻从它上面读取数据可能需要等一段时间此时线程也必须一直阻塞着。在向Socket写入数据时也可能会使线程阻塞。这里准备了一个示例主要逻辑如下客户端创建20个Socket并连接到服务器上再创建20个线程每个线程负责一个Socket。服务器端接收到这20个连接创建20个Socket接着创建20个线程每个线程负责一个Socket。为了模拟服务器端的Socket在创建后不能立马读取数据让客户端的20个线程分别休眠5-10之间的一个随机秒数。客户端的20个线程会在第5秒到第10秒这段时间内陆陆续续的向服务器端发送数据服务器端的20个线程也会陆陆续续接收到数据。/** * author lixinjie * since 2019-05-07 */ public class BioServer { static AtomicInteger counter new AtomicInteger(0); static SimpleDateFormat sdf new SimpleDateFormat(“HH:mm:ss”); public static void main(String[] args) { try { ServerSocket ss new ServerSocket(); ss.bind(new InetSocketAddress(“localhost”, 8080)); while (true) { Socket s ss.accept(); processWithNewThread(s); } } catch (Exception e) { e.printStackTrace(); } } static void processWithNewThread(Socket s) { Runnable run () - { InetSocketAddress rsa (InetSocketAddress)s.getRemoteSocketAddress(); System.out.println(time() “-” rsa.getHostName() “:” rsa.getPort() “-” Thread.currentThread().getId() “:” counter.incrementAndGet()); try { String result readBytes(s.getInputStream()); System.out.println(time() “-” result “-” Thread.currentThread().getId() “:” counter.getAndDecrement()); s.close(); } catch (Exception e) { e.printStackTrace(); } }; new Thread(run).start(); } static String readBytes(InputStream is) throws Exception { long start 0; int total 0; int count 0; byte[] bytes new byte[1024]; //开始读数据的时间 long begin System.currentTimeMillis(); while ((count is.read(bytes)) -1) { if (start 1) { //第一次读到数据的时间 start System.currentTimeMillis(); } total count; } //读完数据的时间 long end System.currentTimeMillis(); return “wait” (start - begin) “ms,read” (end - start) “ms,total” total “bs”; } static String time() { return sdf.format(new Date()); } }/** * author lixinjie * since 2019-05-07/ public class Client { public static void main(String[] args) { try { for (int i 0; i 20; i) { Socket s new Socket(); s.connect(new InetSocketAddress(“localhost”, 8080)); processWithNewThread(s, i); } } catch (IOException e) { e.printStackTrace(); } } static void processWithNewThread(Socket s, int i) { Runnable run () - { try { //睡眠随机的5-10秒模拟数据尚未就绪 Thread.sleep((new Random().nextInt(6) 5) * 1000); //写1M数据为了拉长服务器端读数据的过程 s.getOutputStream().write(prepareBytes()); //睡眠1秒让服务器端把数据读完 Thread.sleep(1000); s.close(); } catch (Exception e) { e.printStackTrace(); } }; new Thread(run).start(); } static byte[] prepareBytes() { byte[] bytes new byte[10241024*1]; for (int i 0; i bytes.length; i) { bytes[i] 1; } return bytes; } }执行结果如下时间-IP:Port-线程Id:当前线程数 15:11:52-127.0.0.1:55201-10:1 15:11:52-127.0.0.1:55203-12:2 15:11:52-127.0.0.1:55204-13:3 15:11:52-127.0.0.1:55207-16:4 15:11:52-127.0.0.1:55208-17:5 15:11:52-127.0.0.1:55202-11:6 15:11:52-127.0.0.1:55205-14:7 15:11:52-127.0.0.1:55206-15:8 15:11:52-127.0.0.1:55209-18:9 15:11:52-127.0.0.1:55210-19:10 15:11:52-127.0.0.1:55213-22:11 15:11:52-127.0.0.1:55214-23:12 15:11:52-127.0.0.1:55217-26:13 15:11:52-127.0.0.1:55211-20:14 15:11:52-127.0.0.1:55218-27:15 15:11:52-127.0.0.1:55212-21:16 15:11:52-127.0.0.1:55215-24:17 15:11:52-127.0.0.1:55216-25:18 15:11:52-127.0.0.1:55219-28:19 15:11:52-127.0.0.1:55220-29:20 时间-等待数据的时间,读取数据的时间,总共读取的字节数-线程Id:当前线程数 15:11:58-wait5012ms,read1022ms,total1048576bs-17:20 15:11:58-wait5021ms,read1022ms,total1048576bs-13:19 15:11:58-wait5034ms,read1008ms,total1048576bs-11:18 15:11:58-wait5046ms,read1003ms,total1048576bs-12:17 15:11:58-wait5038ms,read1005ms,total1048576bs-23:16 15:11:58-wait5037ms,read1010ms,total1048576bs-22:15 15:11:59-wait6001ms,read1017ms,total1048576bs-15:14 15:11:59-wait6016ms,read1013ms,total1048576bs-27:13 15:11:59-wait6011ms,read1018ms,total1048576bs-24:12 15:12:00-wait7005ms,read1008ms,total1048576bs-20:11 15:12:00-wait6999ms,read1020ms,total1048576bs-14:10 15:12:00-wait7019ms,read1007ms,total1048576bs-26:9 15:12:00-wait7012ms,read1015ms,total1048576bs-21:8 15:12:00-wait7023ms,read1008ms,total1048576bs-25:7 15:12:01-wait7999ms,read1011ms,total1048576bs-18:6 15:12:02-wait9026ms,read1014ms,total1048576bs-10:5 15:12:02-wait9005ms,read1031ms,total1048576bs-19:4 15:12:03-wait10007ms,read1011ms,total1048576bs-16:3 15:12:03-wait10006ms,read1017ms,total1048576bs-29:2 15:12:03-wait10010ms,read1022ms,total1048576bs-28:1可以看到服务器端确实为每个连接创建一个线程共创建了20个线程。客户端进入休眠约5-10秒模拟连接上数据不就绪服务器端线程在等待等待时间约5-10秒。客户端陆续结束休眠往连接上写入1M数据服务器端开始读取数据整个读取过程约1秒。可以看到服务器端的工作线程会把时间花在“等待数据”和“读取数据”这两个过程上。这有两个不好的地方一是有很多客户端同时发起请求的话服务器端要创建很多的线程可能会因为超过了上限而造成崩溃。二是每个线程的大部分时光中都是在阻塞着无事可干造成极大的资源浪费。开头已经说了那个年代网民很少所以不可能会有大量请求同时过来。至于资源浪费就浪费吧反正闲着也是闲着。来个简单的小例子饭店共有10张桌子且配备了10位服务员。只要有客人来了大堂经理就把客人带到一张桌子并安排一位服务员全程陪同。即使客人暂时不需要服务服务员也一直在旁边站着。可能觉着是一种浪费其实非也这就是尊贵的VIP服务。其实VIP映射的是一对一的模型主要体现在“专用”上或“私有”上。真正的多路复用技术多路复用技术原本指的是在通信方面多种信号或数据从宏观上看交织在一起使用同一条传输通道进行传输。这样做的目的一方面可以充分利用通道的传输能力另一方面自然是省时省力省钱啦。其实这个概念非常的“生活化”随手就可以举个例子一条小水渠里水在流在一端往里倒入大量乒乓球在另一端用网进行过滤把乒乓球和水流分开。这就是一个比较“土”的多路复用首先在发射端把多种信号或数据进行“混合”接着是在通道上进行传输最后在接收端“分离”出自己需要的信号或数据。相信大家都看出来了这里的重点其实就是处理好“混合”和“分离”对于不同的信号或数据有不同的处理方法。比如以前的有线电视是模拟信号即电磁波。一家一般只有一根信号线但可以同时接多个电视每个电视任意换台互不影响。这是由于不同频率的波可以混合和分离。当然可能不是十分准确明白意思就行了。再比如城市的高铁站一般都有数个站台供高铁同时停靠但城市间的高铁轨道单方向只有一条如何保证那么多趟高铁安全运行呢很明显是分时使用每趟高铁都有自己的时刻。多趟高铁按不同的时刻出站相当于混合按不同的时刻进站相当于分离。总结一下多路指的是多种不同的信号或数据或其它事物复用指的是共用同一个物理链路或通道或载体。可见多路复用技术是一种一对多的模型“多”的这一方复用了“一”的这一方。其实一对多的模型主要体现在“公用”上或“共享”上。您先看着我一会再过来一对一服务是典型的有钱任性虽然响应及时、服务周到但不是每个人都能享受的毕竟还是“屌丝”多嘛那就来个共享服务吧。所以实际当中更多的情况是客人坐下后会给他一个菜单让他先看着反正也不可能立马点餐服务员就去忙别的了。可能不时的会有服务员从客人身旁经过发现客人还没有点餐就会主动去询问现在需要点餐吗如果需要服务员就给你写菜单如果不需要服务员就继续往前走了。这种情况饭店整体运行的也很好但是服务员人数少多了。现在服务10桌客人4个服务员绰绰有余。这节省的可都是纯利润呀。)因为10桌客人同时需要服务的情况几乎是不会发生的绝大部分情况都是错开的。如果真有的话那就等会好了又不是120/119人命关天的。回到代码里情况与之非常相似完全可以采用相同的理论去处理。连接建立后找个地方把它放到那里可以暂时先不管它反正此时也没有数据可读。但是数据早晚会到来的所以要不时的去询问每个连接有数据没有有的话就读取数据没有的话就继续不管它。其实这个模式在Java里早就有了就是Java NIO这里的大写字母“N”是单词“New”即“新”的意思主要是为了和上面的“一对一”进行区分。先铺垫一下吧现在需要把Socket交互的过程再稍微细化一些。客户端先请求连接connect服务器端然后接受连接accept然后客户端再向连接写入数据write接着服务器端从连接上读出数据read。和打电话的场景一样主叫拨号connect被叫接听accept主叫说话speak被叫聆听listen。主叫给被叫打电话说明主叫找被叫有事所以被叫关注的是接通电话听对方说。客户端主动向服务器端发起请求说明客户端找服务器端有事所以服务器端关注的是接受请求读取对方传来的数据。这里把接受请求读取数据称为服务器端感兴趣的操作。在Java NIO中接受请求的操作用OP_ACCEPT表示读取数据的操作用OP_READ表示。我决定先过一遍饭店的场景让首次接触Java NIO的同学不那么迷茫。就是把常规的场景进行了定向整理稍微有点刻意明白意思就行了。1、专门设立一个“跑腿”服务员工作职责单一就是问问客人是否需要服务。2、站在门口接待客人本来是大堂经理的工作但是他不愿意在门口盯着于是就委托给跑腿服务员你帮我盯着有人来了告诉我。于是跑腿服务员就有了一个任务替大堂经理盯梢。终于来客人了跑腿服务员赶紧告诉了大堂经理。3、大堂经理把客人带到座位上对跑腿服务员说客人接下来肯定是要点餐的但是现在在看菜单不知道什么时候能看好所以你不时的过来问问看需不需要点餐需要的话就再喊来一个“点餐”服务员给客人写菜单。于是跑腿服务员就又多了一个任务就是盯着这桌客人不时来问问如果需要服务的话就叫点餐服务员过来服务。4、跑腿服务员在某次询问中客人终于决定点餐了跑题服务员赶紧找来一个点餐服务员为客人写菜单。5、就这样跑腿服务员既要盯着门外新过来的客人也要盯着门内已经就坐的客人。新客人来了通知大堂经理去接待。就坐的客人决定点餐了通知点餐服务员去写菜单。事情就这样一直循环的持续下去一切都挺好。角色明确职责单一配合很好。大堂经理和点餐服务员是需求的提供者或实现者跑腿服务员是需求的发现者并识别出需求的种类需要接待的交给大堂经理需要点餐的交给点餐服务员。哈哈Java NIO来啦代码的写法非常的固定可以配合着后面的解说来看这样就好理解了如下/** * author lixinjie * since 2019-05-07 */ public class NioServer { static int clientCount 0; static AtomicInteger counter new AtomicInteger(0); static SimpleDateFormat sdf new SimpleDateFormat(“HH:mm:ss”); public static void main(String[] args) { try { Selector selector Selector.open(); ServerSocketChannel ssc ServerSocketChannel.open(); ssc.configureBlocking(false); ssc.register(selector, SelectionKey.OP_ACCEPT); ssc.bind(new InetSocketAddress(“localhost”, 8080)); while (true) { selector.select(); Set keys selector.selectedKeys(); Iterator iterator keys.iterator(); while (iterator.hasNext()) { SelectionKey key iterator.next(); iterator.remove(); if (key.isAcceptable()) { ServerSocketChannel ssc1 (ServerSocketChannel)key.channel(); SocketChannel sc null; while ((sc ssc1.accept()) ! null) { sc.configureBlocking(false); sc.register(selector, SelectionKey.OP_READ); InetSocketAddress rsa (InetSocketAddress)sc.socket().getRemoteSocketAddress(); System.out.println(time() “-” rsa.getHostName() “:” rsa.getPort() “-” Thread.currentThread().getId() “:” (clientCount)); } } else if (key.isReadable()) { //先将“读”从感兴趣操作移出待把数据从通道中读完后再把“读”添加到感兴趣操作中 //否则该通道会一直被选出来 key.interestOps(key.interestOps() (~ SelectionKey.OP_READ)); processWithNewThread((SocketChannel)key.channel(), key); } } } } catch (Exception e) { e.printStackTrace(); } } static void processWithNewThread(SocketChannel sc, SelectionKey key) { Runnable run () - { counter.incrementAndGet(); try { String result readBytes(sc); //把“读”加进去 key.interestOps(key.interestOps() | SelectionKey.OP_READ); System.out.println(time() “-” result “-” Thread.currentThread().getId() “:” counter.get()); sc.close(); } catch (Exception e) { e.printStackTrace(); } counter.decrementAndGet(); }; new Thread(run).start(); } static String readBytes(SocketChannel sc) throws Exception { long start 0; int total 0; int count 0; ByteBuffer bb ByteBuffer.allocate(1024); //开始读数据的时间 long begin System.currentTimeMillis(); while ((count sc.read(bb)) -1) { if (start 1) { //第一次读到数据的时间 start System.currentTimeMillis(); } total count; bb.clear(); } //读完数据的时间 long end System.currentTimeMillis(); return “wait” (start - begin) “ms,read” (end - start) “ms,total” total “bs”; } static String time() { return sdf.format(new Date()); } }
 推流例程 — 技术架构文档)
