安卓手机建站dede 网站标题

当前位置： 首页 > news >正文

安卓手机建站,dede 网站标题,荣耀商城官网网站,wordpress没有加载图片作者简介#xff1a;☕️大家好#xff0c;我是Aomsir#xff0c;一个爱折腾的开发者#xff01; 个人主页#xff1a;Aomsir_Spring5应用专栏,Netty应用专栏,RPC应用专栏-CSDN博客 当前专栏#xff1a;Netty应用专栏_Aomsir的博客-CSDN博客 文章目录 参考文献前言单线程… 作者简介☕️大家好我是Aomsir一个爱折腾的开发者 个人主页Aomsir_Spring5应用专栏,Netty应用专栏,RPC应用专栏-CSDN博客 当前专栏Netty应用专栏_Aomsir的博客-CSDN博客 文章目录 参考文献前言单线程Reactor模型主从式Reactor模型多线程知识扫盲Worker线程Boss线程客户端 总结 参考文献 孙哥suns说NettyNetty官方文档 前言 在我们之前的文章中我们详细地探讨了Java NIO和Selector的相关内容这为我们进一步的学习打下了坚实的基础。从本篇文章开始我们将深入学习并理解Reactor模型 单线程Reactor模型 在前两篇文章我们使用Selector去监控Channel的ACCEPT事件、WRITE事件、READ事件等等监听到以后就在当前线程进行处理这已经是一个单线程的Reactor的模型Selector来进行分发起到一个多路复用器的作用但是这个还远远不够怎么能只让一个线程来同时处理ACCEPT、WRITE和READ所以就有了我们后面的主从式
主从式Reactor模型 谈起主从架构也就是master-slave它是主节点做一部分内容从节点做另外一部分的内容主从都干活但是干的活内容不一样比如我们常见的MySQL和Redis它们的读写分离就是主从式架构。 还有一种架构是主备架构(Master-Backup)这种架构就是主挂了以后从起作用干的活是一样的比如Redis中的哨兵机制。 结合如下图例更为详细的理解主从式Reactor模型我们的Boss和Worker都是不同的线程甚至在实战过程中会是不同的服务器。Boss线程主要用于接收Accept请求去与客户端建立SocketChannel连接Worker线程主要去处理实际的读写操作。我们需要把单线程Reactor模型中的sc.register(selector, SelectionKey.OP_READ)转移到Worker线程中去。
多线程知识扫盲 在接下来的学习中我们将使用NIO和Selector来实现一个主从Reactor模型。这需要我们具备一定的多线程知识因此这里我会为你简单介绍一下Java中的多线程。 在Java中我们通常通过Thread类来创建和管理新的线程。在实际开发中我们可以创建一个新的类让它继承Thread类并重写其run方法。在这个run方法中我们可以编写自己的多线程任务逻辑。但是Java的类只能单继承这在某些情况下可能会对我们的系统设计造成限制。 因此Java还为我们提供了另一种创建线程的方式即通过实现Runnable接口。我们可以自定义一个类让它实现Runnable接口并重写其run方法在这个方法中编写我们的多线程任务逻辑。然后我们可以将这个Runnable实现类的对象传递给Thread类的构造方法从而创建Thread类的对象。这种方式的优点是我们不再需要直接继承Thread类来实现多线程任务而是可以将任务逻辑封装在实现了Runnable接口的类中。这样我们的类就可以在保持多线程功能的同时也能继承其他类从而提供更大的设计灵活性。 由于Java的Thread类实现了Runnable接口我们可以在设计系统时采用以下策略在Runnable的实现类中添加一个Thread类型的属性并提供一个register方法。在这个register方法中我们可以初始化Thread属性直接将当前类对象(Runnable实现类)传入Thread构造方法进行初始化然后启动线程。这样我们就可以直接在Runnable内部直接进行线程任务逻辑等而外部只需要提供一个Runnable接口实现类线程的创建和启动等都在Runnable接口内部进行操作封装度更高也更灵活 ⚠️注意 启动多线程任务是通过Thread#start()方法而不是通过Thread#run()方法调用start方法以后CPU的时间片也不会立马分配给这个线程除了Thread#run()和Runnable#run()方法内的代码是属于多线程的其余的都是main线程包括Runnable实现类中的自定义方法CPU时间片不一定会等待主线程某个方法完全执行才切换给别的线程但它一定会等一个代码块执行完比如if public class MyThread extends Thread{Overridepublic void run() {for (int i 0; i 100; i) {System.out.println(线程任务逻辑 i);}} }public class MyRunnable implements Runnable{Overridepublic void run() {for (int i 0; i 100; i) {System.out.println(线程任务逻辑 i);}} }public class RunnableTest {public static void main(String[] args) {// 创建任务对象MyRunnable myRunnable new MyRunnable();// 创建线程对象,并将任务传递进去Thread t1 new Thread(myRunnable);// 启动线程t1.start();for (int i 0; i 100; i) {System.out.println(main线程 i);}} }Worker线程 这是我们的Worker线程用于处理客户端与服务端的读写在我们这个案例中所有的读写都交给这些worker线程主线程不管具体的写。如果是单核CPU那么时间片会不停的在这些时间片时间轮转而如果是多核CPU那直接主线程用于处理连接多个worker线程用于处理具体的读写 在下面这个Worker模型中我们Worker是一个Runnable实现类其中包含一个Thread类型的属性在register方法中对它进行初始化将实现类本类对象传入代表后面这个thread对象调用的任务是实现类中重写的run方法的逻辑。 ⚠️注意 主线程和Worker线程维护不同的selector以免出现污染主线程的selector监控ServerSocketChannel的ACCEPT事件Worker线程的selector监控注册在对应线程上的SocketChannel的READ/WRITE事件register方法属于主线程如果等初始化完还没将SocketChannel注册到这个线程的Selector上就去执行Worker线程的run方法那selector就会成为阻塞状态当CPU时间片切换回主线程就会注册不上成为一个死锁状态。为了解决上面这个问题我们需要将注册这部分的代码放在任务队列里进行传递但是阻塞问题还是存在所以我们将selector唤醒不让其阻塞。当时间片切换到Worker线程select方法就不会阻塞运行循环下面的代码将注册的代码取出来运行然后处理读写 ☹️难点 时间片在sc未注册到selector时就切换给worker线程导致selector阻塞然后导致阻塞使用任务队列传递代码同时需要唤醒selector思路的转变 public class Worker implements Runnable {private static final Logger log LoggerFactory.getLogger(Worker.class);// 一个线程对应一个selector,以免污染private Selector selector;// 线程Thread对象private Thread thread;// 线程名private String name;// 通过volatile进行线程同步private volatile boolean isCreated;// 任务队列private ConcurrentLinkedQueueRunnable runnables new ConcurrentLinkedQueue();// 构造器public Worker(String name) {this.name name;}// 线程任务(此段还属于主线程!!!)public void register(SocketChannel sc) throws IOException,InterruptedException {log.debug(worker register invoke…);// 对于一个Runnable对象,被调用register后// isCreated标志位就会被置为true// 注意CPU等这个if代码块执行结束才有可能被调度到worker线程if (!isCreated) {thread new Thread(this, name);// 调了start,不会立马分配资源(除非抢夺)thread.start();selector Selector.open();isCreated true;}// 模拟此处时间片分给worker线程// worker线程进入run方法,开始阻塞监听// selector就会一直阻塞在select方法上,时间片切换回主线程也无法注册// Thread.sleep(1000);// 任务队列将main线程中注册的代码传递给worker线程runnables.add(() - {try {sc.register(selector, SelectionKey.OP_READ);} catch (ClosedChannelException e) {throw new RuntimeException(e);}});// 唤醒阻塞在select方法上的worker线程// 这样时间片切换到worker线程就直接跳过select方法selector.wakeup();}/*** 线程任务实际处理读写操作*/Overridepublic void run() {while (true) {log.debug(worker run method invoke…);try {// 阻塞监听SocketChannel的OP_READselector.select();// 从任务队列中取出任务执行Runnable poll runnables.poll();if (poll ! null) {// 执行注册的步骤poll.run();}IteratorSelectionKey iterator selector.selectedKeys().iterator();while (iterator.hasNext()) {SelectionKey scKey iterator.next();iterator.remove();if (scKey.isReadable()) {SocketChannel sc (SocketChannel) scKey.channel();ByteBuffer buffer ByteBuffer.allocate(30);int read sc.read(buffer);if (read -1) {scKey.cancel();break;}buffer.flip();String result Charset.defaultCharset().decode(buffer).toString();System.out.println(result result);}}} catch (IOException e) {throw new RuntimeException(e);}}} }Boss线程 Boss线程用来监听ServerSocketChannel的ACCEPT事件监听到了以后将其传递给Worker线程去注册和监听处理注意线程池只有两个Worker线程为了保证每一个新进来的SocketChannel都被注册到与前一个线程不同的线程上这里使用AtomicInteger原子操作类来处理 public class ReactorBossServer {private static final Logger log LoggerFactory.getLogger(ReactorBossServer.class);public static void main(String[] args) throws Exception{log.debug(boss thread start…);ServerSocketChannel ssc ServerSocketChannel.open();ssc.configureBlocking(false);ssc.bind(new InetSocketAddress(8000));Selector selector Selector.open();ssc.register(selector, SelectionKey.OP_ACCEPT);// 模拟任务池,将任务对象进行创建Worker[] workers new Worker[2];for (int i 0; i workers.length; i) {// Worker worker new Worker(worker1);workers[i] new Worker(worker - i);}// 原子操作类AtomicInteger index new AtomicInteger();while (true) {// 阻塞等待Channel的事件的触发selector.select();IteratorSelectionKey iterator selector.selectedKeys().iterator();while (iterator.hasNext()) {SelectionKey sscSelectionKey iterator.next();iterator.remove();// 如果是ACCEPT请求则进行处理,交给worker线程处理if (sscSelectionKey.isAcceptable()) {SocketChannel sc ssc.accept();sc.configureBlocking(false);log.debug(boss invoke worker register…);// hash取模 x%2 0|1// 通过原子类确保sc每次进来注册给不同的workerworkers[index.getAndIncrement() % workers.length].register(sc);log.debug(boss invoke worker register…);}}}} }客户端 public class MyClient {private static final Logger log LoggerFactory.getLogger(MyClient.class);public static void main(String[] args) throws Exception{// 1、创建客户端channel,并连接服务端SocketChannel socketChannel SocketChannel.open();socketChannel.connect(new InetSocketAddress(8000));socketChannel.write(Charset.defaultCharset().encode(hello\n));System.out.println(————————————-);} }总结 这篇文章的阅读绝对值得你的精心研读。Netty的基础建立在NIO之上如果单纯的学习Netty你可能只会看到一堆的API而无法深入理解其背后的设计原则和工作机制。 然而本篇文章从设计理念到编码实践详细剖析了Reactor模型为Netty的学习铺平了道路。这不仅帮助你理解Netty的运作方式更能让你洞察其背后的设计哲学使你在学习时不仅知其然更能知其所以然。因此这篇文章对于深化你对Netty的理解研究其内部工作原理无疑具有极大的价值