前面 https://blog.csdn.net/qq_32459653/article/details/81537669中分析了channelPIpeline 在中执行的流程netty的流程,理解这个是我们理解netty源码的基础,今天我们在做一件事,模拟netty实现一个简易的NIO框架,事实上,这也是netty本质的模型,必要的时候我会添加一些代码注释帮助你们理解
先定义一个boss 接口和worker接口
package com.lin.NIONetty.pool; import java.nio.channels.ServerSocketChannel; /** * boss接口 * * */ public interface Boss { /** * 加入一个新的ServerSocket * @param serverChannel */ public void registerAcceptChannelTask(ServerSocketChannel serverChannel); }
package com.lin.NIONetty.pool; import java.nio.channels.SocketChannel; /** * worker接口 * * */ public interface Worker { /** * 加入一个新的客户端会话 * @param channel */ public void registerNewChannelTask(SocketChannel channel); }
这里抽象出一个公共父类
package com.lin.NIONetty; import com.lin.NIONetty.pool.NioSelectorRunnablePool; import java.io.IOException; import java.nio.channels.Selector; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicBoolean; /** * 抽象selector线程类 * * - * */ public abstract class AbstractNioSelector implements Runnable { /** * 线程池 */ private final Executor executor; /** * 选择器 */ protected Selector selector; /** * 选择器wakenUp状态标记 */ protected final AtomicBoolean wakenUp = new AtomicBoolean(); /** * 任务队列 */ private final Queue<Runnable> taskQueue = new ConcurrentLinkedQueue<Runnable>(); /** * 线程名称 */ private String threadName; /** * 线程管理对象 */ protected NioSelectorRunnablePool selectorRunnablePool; AbstractNioSelector(Executor executor, String threadName, NioSelectorRunnablePool selectorRunnablePool) { this.executor = executor; this.threadName = threadName; this.selectorRunnablePool = selectorRunnablePool; openSelector(); } /** * 获取selector并启动线程 */ private void openSelector() { try { this.selector = Selector.open(); } catch (IOException e) { throw new RuntimeException("Failed to create a selector."); } executor.execute(this); } public void run() { Thread.currentThread().setName(this.threadName); while (true) { try { wakenUp.set(false); select(selector); processTaskQueue(); process(selector); } catch (Exception e) { // ignore } } } /** * 注册一个任务并激活selector * * @param task */ protected final void registerTask(Runnable task) { taskQueue.add(task); Selector selector = this.selector; if (selector != null) { if (wakenUp.compareAndSet(false, true)) { selector.wakeup(); //在这里唤醒 } } else { taskQueue.remove(task); } } /** * 执行队列里的任务 */ private void processTaskQueue() { for (;;) { final Runnable task = taskQueue.poll(); if (task == null) { break; } task.run(); } } /** * 获取线程管理对象 * @return */ public NioSelectorRunnablePool getSelectorRunnablePool() { return selectorRunnablePool; } /** * select抽象方法 * * @param selector * @return * @throws IOException */ protected abstract int select(Selector selector) throws IOException; /** * selector的业务处理 * * @param selector * @throws IOException */ protected abstract void process(Selector selector) throws IOException; }
package com.lin.NIONetty; import com.lin.NIONetty.pool.Boss; import com.lin.NIONetty.pool.NioSelectorRunnablePool; import com.lin.NIONetty.pool.Worker; import org.jboss.netty.handler.codec.spdy.SpdySynReplyFrame; import java.io.IOException; import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; import java.util.concurrent.Executor; /** * boss实现类 * * */ public class NioServerBoss extends AbstractNioSelector implements Boss { public NioServerBoss(Executor executor, String threadName, NioSelectorRunnablePool selectorRunnablePool) { super(executor, threadName, selectorRunnablePool); } @Override protected void process(Selector selector) throws IOException { Set<SelectionKey> selectedKeys = selector.selectedKeys(); if (selectedKeys.isEmpty()) { return; } for (Iterator<SelectionKey> i = selectedKeys.iterator(); i.hasNext();) { SelectionKey key = i.next(); i.remove(); ServerSocketChannel server = (ServerSocketChannel) key.channel(); // 新客户端 SocketChannel channel = server.accept(); // 设置为非阻塞 channel.configureBlocking(false); // 获取一个worker Worker nextworker = getSelectorRunnablePool().nextWorker(); // 注册新客户端接入任务 nextworker.registerNewChannelTask(channel); System.out.println("新客户端链接"); } } public void registerAcceptChannelTask(final ServerSocketChannel serverChannel){ final Selector selector = this.selector; registerTask(new Runnable() { public void run() { try { //注册serverChannel到selector serverChannel.register(selector, SelectionKey.OP_ACCEPT); } catch (ClosedChannelException e) { e.printStackTrace(); } } }); } @Override protected int select(Selector selector) throws IOException { int size= selector.selectedKeys().size(); return selector.select(); } }
package com.lin.NIONetty; import com.lin.NIONetty.pool.NioSelectorRunnablePool; import com.lin.NIONetty.pool.Worker; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; import java.util.concurrent.Executor; /** * worker实现类 * * */ public class NioServerWorker extends AbstractNioSelector implements Worker { public NioServerWorker(Executor executor, String threadName, NioSelectorRunnablePool selectorRunnablePool) { super(executor, threadName, selectorRunnablePool); } @Override protected void process(Selector selector) throws IOException { Set<SelectionKey> selectedKeys = selector.selectedKeys(); if (selectedKeys.isEmpty()) { return; } Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator(); while (ite.hasNext()) { SelectionKey key = (SelectionKey) ite.next(); // 移除,防止重复处理 ite.remove(); // 得到事件发生的Socket通道 SocketChannel channel = (SocketChannel) key.channel(); // 数据总长度 int ret = 0; boolean failure = true; ByteBuffer buffer = ByteBuffer.allocate(1024); //读取数据 try { ret = channel.read(buffer); failure = false; } catch (Exception e) { // ignore } //判断是否连接已断开 if (ret <= 0 || failure) { key.cancel(); System.out.println("客户端断开连接"); }else{ System.out.println("收到数据:" + new String(buffer.array())); //回写数据 ByteBuffer outBuffer = ByteBuffer.wrap("收到\n".getBytes()); channel.write(outBuffer);// 将消息回送给客户端 } } } /** * 加入一个新的socket客户端 */ public void registerNewChannelTask(final SocketChannel channel){ final Selector selector = this.selector; registerTask(new Runnable() { public void run() { try { //将客户端注册到selector中 channel.register(selector, SelectionKey.OP_READ); } catch (ClosedChannelException e) { e.printStackTrace(); } } }); } @Override protected int select(Selector selector) throws IOException { return selector.select(500); } }
接着定义一个管理类,用于初始化worker和boss
package com.lin.NIONetty.pool; import com.lin.NIONetty.NioServerBoss; import com.lin.NIONetty.NioServerWorker; import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicInteger; /** * selector线程管理者 * * */ public class NioSelectorRunnablePool { /** * boss线程数组 */ private final AtomicInteger bossIndex = new AtomicInteger(); private Boss[] bosses; /** * worker线程数组 */ private final AtomicInteger workerIndex = new AtomicInteger(); private Worker[] workeres; public NioSelectorRunnablePool(Executor boss, Executor worker) { initBoss(boss, 1); initWorker(worker, 1); } /** * 初始化boss线程 * @param boss * @param count */ private void initBoss(Executor boss, int count) { this.bosses = new NioServerBoss[count]; for (int i = 0; i < bosses.length; i++) { bosses[i] = new NioServerBoss(boss, "boss thread " + (i+1), this); } } /** * 初始化worker线程 * @param worker * @param count */ private void initWorker(Executor worker, int count) { this.workeres = new NioServerWorker[count]; for (int i = 0; i < workeres.length; i++) { workeres[i] = new NioServerWorker(worker, "worker thread " + (i+1), this); } } /** * 获取一个worker * @return */ public Worker nextWorker() { return workeres[Math.abs(workerIndex.getAndIncrement() % workeres.length)]; } /** * 获取一个boss * @return */ public Boss nextBoss() { return bosses[Math.abs(bossIndex.getAndIncrement() % bosses.length)]; } }
package com.lin.NIONetty; import com.lin.NIONetty.pool.Boss; import com.lin.NIONetty.pool.NioSelectorRunnablePool; import java.net.SocketAddress; import java.nio.channels.ServerSocketChannel; /** * 服务类 * * */ public class ServerBootstrap { private NioSelectorRunnablePool selectorRunnablePool; public ServerBootstrap(NioSelectorRunnablePool selectorRunnablePool) { this.selectorRunnablePool = selectorRunnablePool; } /** * 绑定端口 * @param localAddress */ public void bind(final SocketAddress localAddress){ try { // 获得一个ServerSocket通道 ServerSocketChannel serverChannel = ServerSocketChannel.open(); // 设置通道为非阻塞 serverChannel.configureBlocking(false); // 将该通道对应的ServerSocket绑定到port端口 serverChannel.socket().bind(localAddress); //获取一个boss线程 Boss nextBoss = selectorRunnablePool.nextBoss(); //向boss注册一个ServerSocket通道 nextBoss.registerAcceptChannelTask(serverChannel); } catch (Exception e) { e.printStackTrace(); } } }
测试类
package com.lin.NIONetty; import com.lin.NIONetty.pool.NioSelectorRunnablePool; import java.net.InetSocketAddress; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** * 启动函数 * * */ public class Start { public static void main(String[] args) { ExecutorService boss = Executors.newCachedThreadPool(); ExecutorService workers = Executors.newCachedThreadPool(); //初始化线程 NioSelectorRunnablePool nioSelectorRunnablePool = new NioSelectorRunnablePool(boss, workers); //获取服务类 ServerBootstrap bootstrap = new ServerBootstrap(nioSelectorRunnablePool); //绑定端口 bootstrap.bind(new InetSocketAddress(10101)); //该方法内,会唤醒阻塞的work线程 System.out.println("start"); } } 以上也就是netty最精简的代码实现,弄懂了上面的流程及其实现,你就理解了netty最基本的原理,当然不明白的也可以私信我,欢迎大家提问,当然我也是网上收集了许多资料才搞明白的,看的时候要有耐心,一步一步debug(新手记得设置成degbug线程模式),最终你会看明白的,下一篇直接带大家看netty3.0的源码,搞懂了这个看netty源码就会轻松多的