首页 > 其他分享 >8.NIO-多线程优化

8.NIO-多线程优化

时间:2022-10-12 13:45:49浏览次数:58  
标签:workSelector NIO register iter sc new 多线程 优化 public

1.4.4、多线程优化

设计思路:分两组选择器

  • 单线程配一个选择器,专门处理accpet事件(建立连接) BOSS
  • 创建多线程,每个线程一个选择器,专门处理read事件 WORK

服务端

@Slf4j
public class ThreadServer {
    public static void main(String[] args) throws IOException, InterruptedException {
        Thread.currentThread().setName("boss thread");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);

        Selector bossSelector = Selector.open();
        ssc.register(bossSelector, SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress(50001));

        Worker worker = new Worker("work thread");
        worker.register();

        while (true) {
            bossSelector.select();
            Iterator<SelectionKey> iter = bossSelector.selectedKeys().iterator();
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                if (key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    log.debug("connected...{}", sc.getRemoteAddress());
                    //关联selector
                    log.debug("before register...{}", sc.getRemoteAddress());

                    sc.register(worker.workSelector, SelectionKey.OP_READ, null);
                    log.debug("after register...{}", sc.getRemoteAddress());
                }
            }
        }
    }

    @Data
    static class Worker implements Runnable {
        private String name;
        private Selector workSelector;
        private volatile boolean flag = false;

        public Worker(String name) {
            this.name = name;
        }

        //初始化线程和selector

        public void register() throws IOException {
            if (!flag) {
                workSelector = Selector.open();
                new Thread(this, name).start();
                flag = true;
            }

        }

        @Override
        public void run() {
            while (true) {
                try {
                    workSelector.select();
                    Iterator<SelectionKey> iter = workSelector.selectedKeys().iterator();
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();
                        iter.remove();
                        if (key.isReadable()) {
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            SocketChannel sc = (SocketChannel) key.channel();
                            log.debug("read...{}", sc.getRemoteAddress());
                            sc.read(buffer);
                            buffer.flip();
                            ByteBufferUtil.debugAll(buffer);
                        }
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}

客户端

public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 50001));
        SocketAddress address = sc.getLocalAddress();
        System.out.println(address.toString());
        sc.write(Charset.defaultCharset().encode("123456"));
        System.in.read();
    }
}

问题:只打印了before信息,没有打印after和read信息,因为worker线程start后,run方法中workSelector.select();会阻塞,在主线程boss中,sc.register(workerSelecotr..)就不会执行.

run方法是在work线程中执行,sc.register是在boss线程中执行,只要run运行在register之前就会阻塞、

解决:把worker.register();放到sc.register之前,这是sc.register就可能发生在work.register之前,selector.select()后执行,就没有问题了。

新问题:再来一个客户端,之前的select()方法已经执行阻塞了,新的就register不了

解决:把worker.register();和sc.register(worker.workSelector, SelectionKey.OP_READ, null);放早同一个线程里,就可以控制顺序

public void register(SocketChannel sc) throws IOException {
    if (!flag) {
        workSelector = Selector.open();
        new Thread(this, name).start();
        flag = true;
    }

    sc.register(workSelector, SelectionKey.OP_READ, null);
}

但是register还在在boss线程中被调用,只有run方法才是在work线程中执行。

  • 方案1:采用一个队列,将sc.register(workSelector, SelectionKey.OP_READ, null)放入队列中,然后在run方法中从队列中拿到任务,执行,还要调用warkUp唤醒select的阻塞
@Slf4j
public class ThreadServer {
    public static void main(String[] args) throws IOException, InterruptedException {
        Thread.currentThread().setName("boss thread");
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);

        Selector bossSelector = Selector.open();
        ssc.register(bossSelector, SelectionKey.OP_ACCEPT);
        ssc.bind(new InetSocketAddress(50001));

        Worker worker = new Worker("work thread");

        while (true) {
            bossSelector.select();
            Iterator<SelectionKey> iter = bossSelector.selectedKeys().iterator();
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                if (key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    log.debug("connected...{}", sc.getRemoteAddress());
                    //关联selector
                    log.debug("before register...{}", sc.getRemoteAddress());
                    worker.register(sc);

                    log.debug("after register...{}", sc.getRemoteAddress());
                }
            }
        }
    }

    @Data
    static class Worker implements Runnable {
        private String name;
        private Selector workSelector;
        private volatile boolean flag = false;

        private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

        public Worker(String name) {
            this.name = name;
        }

        //初始化线程和selector

        public void register(SocketChannel sc) throws IOException {
            if (!flag) {
                workSelector = Selector.open();
                new Thread(this, name).start();
                flag = true;
            }
            queue.add(()->{
                try {
                    sc.register(workSelector, SelectionKey.OP_READ, null);
                } catch (ClosedChannelException e) {
                    throw new RuntimeException(e);
                }
            });
            workSelector.wakeup();//唤醒select方法
        }

        @Override
        public void run() {
            while (true) {
                try {
                    workSelector.select();

                    //取出任务并执行
                    Runnable task = queue.poll();
                    if (task != null) {
                        task.run();
                    }

                    Iterator<SelectionKey> iter = workSelector.selectedKeys().iterator();
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();
                        iter.remove();
                        if (key.isReadable()) {
                            ByteBuffer buffer = ByteBuffer.allocate(16);
                            SocketChannel sc = (SocketChannel) key.channel();
                            log.debug("read...{}", sc.getRemoteAddress());
                            sc.read(buffer);
                            buffer.flip();
                            ByteBufferUtil.debugAll(buffer);
                        }
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}
  • 方案2:不用队列,23步都是在boss线程中执行,顺序固定

分析:wakeup可以在select前或后唤醒,相当于发一张票

  1. 123顺序,先阻塞,然后warkup唤醒,没问题
  2. 213,ok
  3. 231,和第2情况相同

标签:workSelector,NIO,register,iter,sc,new,多线程,优化,public
From: https://www.cnblogs.com/jpymll/p/16784231.html

相关文章

  • 9.NIO-多线程work轮询
    1.4.5、多线程多work轮询@Slf4jpublicclassThreadServerWorks{publicstaticvoidmain(String[]args)throwsIOException,InterruptedException{T......
  • 10.NIO-IO模型
    1.5、NIOBIO1.5.1、StreamChannelstream不会自动缓冲数据,channel会利用系统提供的发送缓冲区、接收缓冲区(更为底层)stream仅支持阻塞API,channel同时支持阻塞、非......
  • 数据库优化学习笔记_冷热分离
    冷热分离当数据库表数据体量大,即使是做了很多SQL层面的优化(索引、执行计划、优化语句、表结构设计)读写依然很慢可以考虑从冷热数据分离去提高速度热数据:对用户而言,是需......
  • 5.NIO-网络编程-阻塞VS非阻塞
    1.4、网络编程1.4.1阻塞VS非阻塞阻塞:ServerSocketChannel.accept()阻塞到客户端连接SocketChannel.read()阻塞到客户端发送数据//服务端@Slf4jpublicclassSo......
  • 6.NIO-selector详解
    1.4.2、selector单线程可以配合Selector完成对多个Channel可读写事件的监控,这称之为多路复用只能用于网络IO,文件IO不可用,因为FileChannel没有阻塞模式如果不用sele......
  • 7.NIO-selector-写入内容过多
    1.4.3、写入内容过多服务端publicclassWriteServer{publicstaticvoidmain(String[]args)throwsIOException{ServerSocketChannelssc=ServerS......
  • ModStartBlog v5.9.0 新增组件特性,基础布局优化
    系统介绍ModStart是一个基于Laravel模块化极速开发框架。模块市场拥有丰富的功能应用,支持后台一键快速安装,让开发者能快的实现业务功能开发。系统完全开源,基于Apache2.......
  • 1.1.NIO-三大组件
    1、NIO基础non-blockingio非阻塞io1.1、三大组件1.1.1、Channel&Bufferchannel类似于stream,它就是读写数据的双向通道,可以从channel将数据读入buffer,也可以将buffer......
  • #打卡不停更#在开鸿智谷NiobeU4移植lvgl并实现ADC按键状态显示
    本文主要分享在开鸿智谷NiobeU4开发板移植lvgl经验,并实现按键按下sw4显示SW4Pressed松开显示SW4Release,整理踩坑经验分享如下。1.移植准备开鸿智谷NiobeU4开发板Openhar......
  • arthas的调试优化代码运行速度
    $$java-jararthas-boot.jar*[1]:35542[2]:71560math-game.jar$tracedemo.MathGamerunPressQorCtrl+Ctoabort.Affect(class-cnt:1,method-cnt:1)......