正文

网络编程--rpc实现原理--nio多线程--迭代版本v2

limeOracle  limeOracle  2022-09-05  224

关键词：

　　啦啦啦

V2——增加WriteQueue队列，存放selectionKey。addWriteEventToQueue()添加selectionKey并唤醒阻塞的selector。等selector唤醒之后再注册OP_WRITE事件。

（

selectionKey.cancel();清除key对应事件之后，由于多线程 main线程和对应的IO线程会抢夺selector资源。

在selector.select()和sc.register(selectionKey.selector(), SelectionKey.OP_WRITE);处会发生死锁，因此将需要注册IO操作的selectionKey放入队列并唤醒selector，之后在注册IO操作。

）

NIO 多线程

1.打开Selector
2.打开ServerSocketChannel
3.获取与此Channel关联的ServerSocket并绑定地址
4.设置Channel为非阻塞
5.将Channel注册到Selector并指定操作位

6.阻塞select()返回具有操作的通道个数。如果唤醒之后就绪通道数小于1，则将WriteQueue、ReadQueue队列中的所有sc注册到selector中并分别设置写操作位、读操作位。
7.获取SelectionKeys遍历，最后移除
8.根据key判断操作：isAcceptable、isConnectable、isReadable、isWritable
9.1.isAcceptable：根据key获取Channel，accept()获取SocketChannel并设置为非阻塞模式，注册到Selector并指定读操作位和缓冲区。
9.2.isConnectable：仅仅就是连接状态
9.3.isReadable：取消Read事件，启动Read线程处理IO读操作。添加selectionKey到WriteQueue中并唤醒selector。
9.4.isWritable：取消Write事件，启动Write线程处理IO写操作。添加selectionKey到ReadQueue中并唤醒selector。

    private static Queue<SelectionKey> writeQueue = new LinkedBlockingQueue<SelectionKey>();
    public static void addWriteEventToQueue(SelectionKey selectionKey) {
        System.out.println(Thread.currentThread() + " -- -- -- 添加selectionKey到队列，并唤醒selector");
        writeQueue.add(selectionKey);
        selectionKey.selector().wakeup();
    }

　　　　　　　int n = selector.select();
            System.out.println(Thread.currentThread() + " 事件就绪通道个数 ： " + n);
            if(n < 1){
                System.out.println(Thread.currentThread() + " -- -- -- selector被唤醒，注册队列中所有socketChannel为OP_WRITE操作");
                Iterator<SelectionKey> iterator = writeQueue.iterator();
                while(iterator.hasNext()){
                    SelectionKey selectionKey = iterator.next();
                    SocketChannel sc = (SocketChannel) selectionKey.channel();
                    sc.register(selectionKey.selector(), SelectionKey.OP_WRITE);
                }
            }

　　Class ： Service

package lime.pri.limeNio.optimize.socket4;

import java.io.IOException;
import java.net.InetSocketAddress;
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.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class Service {

    public static void main(String[] args) throws IOException, InterruptedException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(9999));
        serverSocketChannel.configureBlocking(false);
        Selector selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        while (true) {
            System.out.println(Thread.currentThread() + " 监听端口@9999,等待客户端连接...");
            int n = selector.select();
            System.out.println(Thread.currentThread() + " 事件就绪通道个数 ： " + n);
            if(n < 1){
                System.out.println(Thread.currentThread() + " -- -- -- selector被唤醒，注册队列中所有socketChannel为OP_WRITE操作");
                Iterator<SelectionKey> iterator = writeQueue.iterator();
                while(iterator.hasNext()){
                    SelectionKey selectionKey = iterator.next();
                    SocketChannel sc = (SocketChannel) selectionKey.channel();
                    sc.register(selectionKey.selector(), SelectionKey.OP_WRITE);
                }
            }
            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectedKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey selectionKey = iterator.next();
                iterator.remove();
                if (selectionKey.isValid() && selectionKey.isAcceptable()) {
                    System.out.println(Thread.currentThread() + " -- -- -- 处理Acceptable事件");
                    ServerSocketChannel ssc = (ServerSocketChannel) selectionKey.channel();
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    sc.register(selector, SelectionKey.OP_READ);
                }
                if (selectionKey.isValid() && selectionKey.isReadable()) {
                    System.out.println(Thread.currentThread() + " -- -- -- 处理Readable事件");
                    selectionKey.cancel();
                    RequestProcessor.proceess(selectionKey, selector);
                }
                if (selectionKey.isValid() && selectionKey.isWritable()) {
                    System.out.println(Thread.currentThread() + " -- -- -- 处理Writable事件");
                    selectionKey.cancel();
                    ResponseProcessor.proceess(selectionKey, selector);
                }
            }
        }
    }

    private static Queue<SelectionKey> writeQueue = new LinkedBlockingQueue<SelectionKey>();
    public static void addWriteEventToQueue(SelectionKey selectionKey) {
        System.out.println(Thread.currentThread() + " -- -- -- 添加selectionKey到队列，并唤醒selector");
        writeQueue.add(selectionKey);
        selectionKey.selector().wakeup();
    }
}

　　Class ： RequestProcessor

package lime.pri.limeNio.optimize.socket4;

import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RequestProcessor {

    private static ExecutorService exec = Executors.newFixedThreadPool(2);

    public static void proceess(final SelectionKey selectionKey, final Selector selector) {
        exec.submit(new Runnable() {

            public void run() {
                try {
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                    byteBuffer.clear();
                    SocketChannel sc = (SocketChannel) selectionKey.channel();
                    ByteArrayOutputStream bos = new ByteArrayOutputStream();
                    while(sc.read(byteBuffer) != -1){
                        bos.write(byteBuffer.array());
                        byteBuffer.clear();
                    }
                    System.out.println(Thread.currentThread() + " 客户端( " + sc.getRemoteAddress() + " ) 请求 ： " + bos.toString());
                    Service.addWriteEventToQueue(selectionKey);
                } catch (Exception e) {
                }
            }
        });
    }

}

　　Class ： ResponseProcessor

package lime.pri.limeNio.optimize.socket4;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ResponseProcessor {

    private static ExecutorService exec = Executors.newFixedThreadPool(2);

    public static void proceess(final SelectionKey selectionKey, final Selector selector) {
        exec.submit(new Runnable() {

            public void run() {
                try {
                    String response = "服务端响应 ： " + new Date().toString();
                    ByteBuffer byteBuffer = ByteBuffer.wrap(response.getBytes());
                    SocketChannel sc = (SocketChannel) selectionKey.channel();
                    sc.write(byteBuffer);
                    sc.close();
                } catch (IOException e) {
                }
            }
        });
    }

}

　　Class ： Client

package lime.pri.limeNio.optimize.socket4;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

public class Client {

    public static void main(String[] args) throws IOException {
        for (int i = 0; i < 10; i++) {
            new Thread() {
                @Override
                public void run() {
                    try {
                        SocketChannel socketChannel = SocketChannel.open();
                        socketChannel.connect(new InetSocketAddress("127.0.0.1", 9999));

                        socketChannel.write(ByteBuffer.wrap("Query Date".getBytes()));
                        socketChannel.shutdownOutput();

                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        socketChannel.read(byteBuffer);
                        System.out.println(new String(byteBuffer.array()));
                        socketChannel.close();
                    } catch (IOException e) {
                    }
                }

            }.start();
        }
    }
}

　　Console： Server

Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 1
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 1
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 3
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 2
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 2
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[pool-1-thread-2,5,main] 客户端( /127.0.0.1:4643 ) 请求 ： Query Date
Thread[pool-1-thread-2,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 2
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 2
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 2
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[pool-1-thread-2,5,main] 客户端( /127.0.0.1:4645 ) 请求 ： Query Date
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[pool-1-thread-1,5,main] 客户端( /127.0.0.1:4644 ) 请求 ： Query Date
Thread[pool-1-thread-1,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[pool-1-thread-2,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 0
Thread[pool-1-thread-2,5,main] 客户端( /127.0.0.1:4646 ) 请求 ： Query Date
Thread[main,5,main] -- -- -- selector被唤醒，注册队列中所有socketChannel为OP_WRITE操作
Thread[pool-1-thread-2,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[pool-1-thread-1,5,main] 客户端( /127.0.0.1:4642 ) 请求 ： Query Date
Thread[pool-1-thread-1,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[pool-1-thread-2,5,main] 客户端( /127.0.0.1:4647 ) 请求 ： Query Date
Thread[pool-1-thread-2,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[pool-1-thread-1,5,main] 客户端( /127.0.0.1:4648 ) 请求 ： Query Date
Thread[main,5,main] 事件就绪通道个数 ： 0
Thread[main,5,main] -- -- -- selector被唤醒，注册队列中所有socketChannel为OP_WRITE操作
Thread[pool-1-thread-1,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 0
Thread[main,5,main] -- -- -- selector被唤醒，注册队列中所有socketChannel为OP_WRITE操作
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 9
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[pool-1-thread-2,5,main] 客户端( /127.0.0.1:4649 ) 请求 ： Query Date
Thread[pool-1-thread-2,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 0
Thread[main,5,main] -- -- -- selector被唤醒，注册队列中所有socketChannel为OP_WRITE操作
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 10
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Acceptable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 1
Thread[main,5,main] -- -- -- 处理Readable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[pool-1-thread-1,5,main] 客户端( /127.0.0.1:4650 ) 请求 ： Query Date
Thread[pool-1-thread-1,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[pool-1-thread-2,5,main] 客户端( /127.0.0.1:4651 ) 请求 ： Query Date
Thread[main,5,main] 事件就绪通道个数 ： 0
Thread[main,5,main] -- -- -- selector被唤醒，注册队列中所有socketChannel为OP_WRITE操作
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[pool-1-thread-2,5,main] -- -- -- 添加selectionKey到队列，并唤醒selector
Thread[main,5,main] 事件就绪通道个数 ： 9
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 0
Thread[main,5,main] -- -- -- selector被唤醒，注册队列中所有socketChannel为OP_WRITE操作
Thread[main,5,main] 监听端口@9999,等待客户端连接...
Thread[main,5,main] 事件就绪通道个数 ： 10
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] -- -- -- 处理Writable事件
Thread[main,5,main] 监听端口@9999,等待客户端连接...

　　Class ： Client

Thread[Thread-7,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-0,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-3,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-9,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-1,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-4,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-6,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-5,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-8,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017
Thread[Thread-2,5,main] 服务端响应 ： Sat Jun 24 16:00:41 CST 2017

啦啦啦

网络编程--rpc实现原理--rpc--迭代版本v1--本地方法调用

　网络编程--RPC实现原理--目录　　啦啦啦V2——RPC--本地方法调用：不通过网络入门　　1. RPCObjectProxyrpcObjectProxy=newRPCObjectProxy(newLocalRPCClient());：绑定目标对象　　2. IUserServiceuserService=(IUserService)rpcObjectProxy.create 查看详情

netty_02_高性能的nio框架(代码片段)

文章目录一、前言二、从NIO到RPC(原理)2.1磁盘IO和网络IO2.2JavaIO的三个阶段(bio-nio-aio)同步阻塞IO(BIO)非阻塞IO（NIO）多路复用机制（select、poll、epoll）异步IO2.3Reactor三种模式单线程单Reactor模型多线程单Reactor模型多线... 查看详情

netty_02_高性能的nio框架(代码片段)

netty高性能原理剖析(代码片段)

...提供了SocketChannel和ServerSocketChannel两种不同的套接字通道实现。2.1多路复用通讯方式Netty架构按照Reactor模式设计和实现，它的服务端通信序列图如下:Netty的IO线程NioEventLoop由于聚合了多路复用器Selector，可以同时并发处理成百上千... 查看详情

netty原理-从nio开始

...非阻塞通信模型，Select（多路复用器）为此模型的核心，实现了多个连接一个线程当有客户端连接请求时此连接请求会被注册至select上，当select检测到此连接有I/O请求时才会打开一个线程去对此I/O请求进行处理-----单线程模型这... 查看详情

netty

...r的机制及子类基本介绍Selector选择器介绍和原理NIO非阻塞网络编程快速入门Netty概述Netty的线程模型概述**Netty服务模型**单Reactor单线程单Reactor多线程主从Reactor多线程Netty模型简单版进阶版详细版查看详情

day786.nio的优化实现原理-java性能调优实战(代码片段)

NIO的优化实现原理Hi，我是阿昌，今天学习记录的是关于NIO的优化实现原理。Tomcat中经常被提到的一个调优就是修改线程的I/O模型。Tomcat8.5版本之前，默认情况下使用的是BIO线程模型，如果在高负载、高并发的场... 查看详情

java网络编程和nio详解7：浅谈linux中nioselector的实现原理

Java网络编程和NIO详解7：浅谈Linux中NIOSelector的实现原理转自：https://www.jianshu.com/p/2b71ea919d49本系列文章首发于我的个人博客：https://h2pl.github.io/欢迎阅览我的CSDN专栏：Java网络编程和NIOhttps://blog.csdn.net/column/details/21963.html部分代码... 查看详情

java实现一个的基于nio的多线程web服务器(代码片段)

代码地址：https://github.com/iyuanyb/webserver实现了静态、动态资源获取；Cookie、Session、HTTP长连接，及Session和HTTP长连接的定时清除；类似SpringMVC的注解式编程，如 @RequestMapping @RequestParam 等，方法中可以根据参数名从前台... 查看详情

深入理解多线程——moniter的实现原理(代码片段)

深入理解多线程（四）——Moniter的实现原理收录于话题#和并发编程有关的那点事儿13个点击上方“Hollis”关注我，精彩内容第一时间呈现。全文字数：1200阅读时间：3分钟本文是《深入理解多线程系列文章》的第四篇。点击查看... 查看详情

python多线程原理与实现(代码片段)

Date:2019-06-04Author:SunPython多线程原理与实战目的：（1）了解python线程执行原理（2）掌握多线程编程与线程同步（3）了解线程池的使用1线程基本概念1.1线程是什么？线程是指进程内的一个执行单元,也是进程内的可调度实体.与进... 查看详情

java高性能编程基础-线程池的应用及实现原理

...池？线程池的原理线程池的接口定义和实现类APIExecutors工具类线程池应用示例任务execute过程各种线程池的使用为什么要用线程池？线程池是一种多线程处理形式，处理... 查看详情

rpc高性能框架总结3.nio示例代码编写和简析

...《NIO的原理以及与传统IO的对比》想要使用Java实现前面提到的高性能网络服务端，需要使用JDK1.4之后提供java.nio包下的ByteBuffer、SelectionKey、Selector、ServerSocketChannel以及SocketChannel;下面就是一个使用JDK提供的API编写的，... 查看详情

并发编程之美，带你深入理解java多线程原理

...统的效率。线程是在同一时间需要完成多项任务的时候被实现的。2.了解多线程了解多线程之前我们先搞清楚几个重要的概念！如上图所示：对我们的项目有一个主内存，这个主内存里面存放了我们的共享变量、方法区、堆中的... 查看详情

异步编程原理以及java实现(代码片段)

...;甚至有人称多线程为异步，实际上这两种技术背后的实现原理是不同的。假设您有2个任务，不涉及任何IO（在多处理器机器上）。在这种情况下，线程优于Async。因为像单线程程序一样的Async按顺序执行你的任... 查看详情

pythongui编程:音乐播放器(多线程爬虫进度条文件)(代码片段)

文章目录1.程序运行结果2.程序实现原理3.GUI布局4.功能介绍5.代码实现1.程序运行结果Python实现音乐播放器(爬虫、多线程、进度条、文件)2.程序实现原理本音乐播放器GUI方面运用Python的tkinter实现，播放的音乐来自网络爬虫和... 查看详情

spark通信框架sparknetworkcommon

参考技术A一直以来，基于Akka实现的RPC通信框架是Spark引以为豪的主要特性，也是与Hadoop等分布式计算框架对比过程中一大亮点，但是时代和技术都在演化，从Spark1.3.1版本开始，为了解决大块数据（如Shuffle）的传输问题，Spark引... 查看详情

java网络编程基础—基于tcp的nio简单聊天系统

在Java网络编程基础（四）中提到了基于Socket的TCP/IP简单聊天系统实现了一个多客户端之间护法消息的简单聊天系统。其服务端采用了多线程来处理多个客户端的消息发送，并转发给目的用户。但是由于它是基于Socket的，因此是... 查看详情