NIO selector 多路复用reactor线程模型
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了NIO selector 多路复用reactor线程模型,小编现在分享给大家,供广大互联网技能从业者学习和参考。文章包含5679字,纯文字阅读大概需要9分钟。
内容图文
NIO selector 多路复用reactor线程模型
package com.study.hc.net.nio; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.*; import java.util.Iterator; import java.util.Random; import java.util.Set; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.FutureTask; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicInteger; /** * NIO selector 多路复用reactor线程模型 */publicclass NIOServerV3 { /** 处理业务操作的线程 */privatestatic ExecutorService workPool = Executors.newCachedThreadPool(); /** * 封装了selector.select()等事件轮询的代码 */abstractclass ReactorThread extends Thread { Selector selector; LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>(); /** * Selector监听到有事件后,调用这个方法 */publicabstractvoid handler(SelectableChannel channel) throws Exception; private ReactorThread() throws IOException { selector = Selector.open(); } volatileboolean running = false; @Override publicvoid run() { // 轮询Selector事件while (running) { try { // 执行队列中的任务 Runnable task; while ((task = taskQueue.poll()) != null) { task.run(); } selector.select(1000); // 获取查询结果 Set<SelectionKey> selected = selector.selectedKeys(); // 遍历查询结果 Iterator<SelectionKey> iter = selected.iterator(); while (iter.hasNext()) { // 被封装的查询结果 SelectionKey key = iter.next(); iter.remove(); int readyOps = key.readyOps(); // 关注 Read 和 Accept两个事件if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { try { SelectableChannel channel = (SelectableChannel) key.attachment(); channel.configureBlocking(false); handler(channel); if (!channel.isOpen()) { key.cancel(); // 如果关闭了,就取消这个KEY的订阅 } } catch (Exception ex) { key.cancel(); // 如果有异常,就取消这个KEY的订阅 } } } selector.selectNow(); } catch (IOException e) { e.printStackTrace(); } } } private SelectionKey register(SelectableChannel channel) throws Exception { // 为什么register要以任务提交的形式,让reactor线程去处理? // 因为线程在执行channel注册到selector的过程中,会和调用selector.select()方法的线程争用同一把锁 // 而select()方法实在eventLoop中通过while循环调用的,争抢的可能性很高,为了让register能更快的执行,就放到同一个线程来处理 FutureTask<SelectionKey> futureTask = new FutureTask<>(() -> channel.register(selector, 0, channel)); taskQueue.add(futureTask); return futureTask.get(); } privatevoid doStart() { if (!running) { running = true; start(); } } } private ServerSocketChannel serverSocketChannel; // 1、创建多个线程 - accept处理reactor线程 (accept线程)private ReactorThread[] mainReactorThreads = new ReactorThread[1]; // 2、创建多个线程 - io处理reactor线程 (I/O线程)private ReactorThread[] subReactorThreads = new ReactorThread[8]; /** * 初始化线程组 */privatevoid newGroup() throws IOException { // 创建IO线程,负责处理客户端连接以后socketChannel的IO读写for (int i = 0; i < subReactorThreads.length; i++) { subReactorThreads[i] = new ReactorThread() { @Override publicvoid handler(SelectableChannel channel) throws IOException { // work线程只负责处理IO处理,不处理accept事件 SocketChannel ch = (SocketChannel) channel; ByteBuffer requestBuffer = ByteBuffer.allocate(1024); while (ch.isOpen() && ch.read(requestBuffer) != -1) { // 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)if (requestBuffer.position() > 0) break; } if (requestBuffer.position() == 0) return; // 如果没数据了, 则不继续后面的处理 requestBuffer.flip(); byte[] content = newbyte[requestBuffer.limit()]; requestBuffer.get(content); System.out.println(new String(content)); System.out.println(Thread.currentThread().getName() + "收到数据,来自:" + ch.getRemoteAddress()); // TODO 业务操作 数据库、接口... workPool.submit(() -> { }); // 响应结果 200 String response = "HTTP/1.1 200 OK\r\n" + "Content-Length: 11\r\n\r\n" + "Hello World"; ByteBuffer buffer = ByteBuffer.wrap(response.getBytes()); while (buffer.hasRemaining()) { ch.write(buffer); } } }; } // 创建mainReactor线程, 只负责处理serverSocketChannelfor (int i = 0; i < mainReactorThreads.length; i++) { mainReactorThreads[i] = new ReactorThread() { AtomicInteger incr = new AtomicInteger(0); @Override publicvoid handler(SelectableChannel channel) throws Exception { // 只做请求分发,不做具体的数据读取 ServerSocketChannel ch = (ServerSocketChannel) channel; SocketChannel socketChannel = ch.accept(); socketChannel.configureBlocking(false); // 收到连接建立的通知之后,分发给I/O线程继续去读取数据int index = incr.getAndIncrement() % subReactorThreads.length; ReactorThread workEventLoop = subReactorThreads[index]; workEventLoop.doStart(); SelectionKey selectionKey = workEventLoop.register(socketChannel); selectionKey.interestOps(SelectionKey.OP_READ); System.out.println(Thread.currentThread().getName() + "收到新连接 : " + socketChannel.getRemoteAddress()); } }; } } /** * 初始化channel,并且绑定一个eventLoop线程 * * @throws IOException IO异常 */privatevoid initAndRegister() throws Exception { // 1、 创建ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.configureBlocking(false); // 2、 将serverSocketChannel注册到selectorint index = new Random().nextInt(mainReactorThreads.length); mainReactorThreads[index].doStart(); SelectionKey selectionKey = mainReactorThreads[index].register(serverSocketChannel); selectionKey.interestOps(SelectionKey.OP_ACCEPT); } /** * 绑定端口 * * @throws IOException IO异常 */privatevoid bind() throws IOException { // 1、 正式绑定端口,对外服务 serverSocketChannel.bind(new InetSocketAddress(8080)); System.out.println("启动完成,端口8080"); } publicstaticvoid main(String[] args) throws Exception { NIOServerV3 nioServerV3 = new NIOServerV3(); nioServerV3.newGroup(); // 1、 创建main和sub两组线程 nioServerV3.initAndRegister(); // 2、 创建serverSocketChannel,注册到mainReactor线程上的selector上 nioServerV3.bind(); // 3、 为serverSocketChannel绑定端口 } }
原文:https://www.cnblogs.com/itbac/p/12046893.html
内容总结
以上是互联网集市为您收集整理的NIO selector 多路复用reactor线程模型全部内容,希望文章能够帮你解决NIO selector 多路复用reactor线程模型所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错,欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 gblab@vip.qq.com 举报,一经查实,本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。