java多路复用器代码 http多路复用原理

阻塞、非阻塞、多路复用、同步、异步、BIO、NIO、AIO 一文搞定

关于IO会涉及到阻塞、非阻塞、多路复用、同步、异步、BIO、NIO、AIO等几个知识点。知识点虽然不难但平常经常容易搞混，特此Mark下，与君共勉。

网站制作、成都网站建设,成都做网站公司-创新互联建站已向上1000家企业提供了,网站设计,网站制作,网络营销等服务!设计与技术结合,多年网站推广经验,合理的价格为您打造企业品质网站。

阻塞IO情况下，当用户调用 read 后，用户线程会被阻塞，等内核数据准备好并且数据从内核缓冲区拷贝到用户态缓存区后 read 才会返回。可以看到是阻塞的两个部分。

非阻塞IO发出read请求后发现数据没准备好，会继续往下执行，此时应用程序会不断轮询polling内核询问数据是否准备好，当数据没有准备好时，内核立即返回EWOULDBLOCK错误。直到数据被拷贝到应用程序缓冲区，read请求才获取到结果。并且你要注意！这里最后一次 read 调用获取数据的过程，是一个同步的过程，是需要等待的过程。这里的同步指的是内核态的数据拷贝到用户程序的缓存区这个过程。

非阻塞情况下无可用数据时，应用程序每次轮询内核看数据是否准备好了也耗费CPU，能否不让它轮询，当内核缓冲区数据准备好了，以事件通知当机制告知应用进程数据准备好了呢？应用进程在没有收到数据准备好的事件通知信号时可以忙写其他的工作。此时 IO多路复用就派上用场了。

IO多路复用中文比较让人头大，IO多路复用的原文叫 I/O multiplexing，这里的 multiplexing 指的其实是在单个线程通过记录跟踪每一个Sock(I/O流)的状态来同时管理多个I/O流. 发明它的目的是尽量多的提高服务器的吞吐能力。实现一个线程监控多个IO请求，哪个IO有请求就把数据从内核拷贝到进程缓冲区，拷贝期间是阻塞的！现在已经可以通过采用mmap地址映射的方法，达到内存共享效果，避免真复制，提高效率。

像 select、poll、epoll 都是I/O多路复用的具体的实现。

select是第一版IO复用，提出后暴漏了很多问题。

poll 修复了 select 的很多问题。

但是poll仍然不是线程安全的，这就意味着不管服务器有多强悍，你也只能在一个线程里面处理一组 I/O 流。你当然可以拿多进程来配合了，不过然后你就有了多进程的各种问题。

epoll 可以说是 I/O 多路复用最新的一个实现，epoll 修复了poll 和select绝大部分问题，比如：

横轴 Dead connections 是链接数的意思，叫这个名字只是它的测试工具叫deadcon。纵轴是每秒处理请求的数量，可看到epoll每秒处理请求的数量基本不会随着链接变多而下降的。poll 和/dev/poll 就很惨了。但 epoll 有个致命的缺点是只有 linux 支持。

比如平常Nginx为何可以支持4W的QPS是因为它会使用目标平台上面最高效的I/O多路复用模型。

然后你会发现上面的提到过的操作都不是真正的异步，因为两个阶段总要等待会儿！而真正的异步 I/O 是内核数据准备好和数据从内核态拷贝到用户态这两个过程都不用等待。

很庆幸，Linux给我们准备了 aio_read 跟 aio_write 函数实现真实的异步，当用户发起aio_read请求后就会自动返回。内核会自动将数据从内核缓冲区拷贝到用户进程空间，应用进程啥都不用管。

我强力推荐C++后端开发免费学习地址：C/C++Linux服务器开发/后台架构师【零声教育】-学习视频教程-腾讯课堂

同步跟异步的区别在于数据从内核空间拷贝到用户空间是否由用户线程完成，这里又分为同步阻塞跟同步非阻塞两种。

我们以同步非阻塞为例，如下可看到，在将数据从内核拷贝到用户空间这一过程，是由用户线程阻塞完成的。

可发现，用户在调用之后会立即返回，由内核完成数据的拷贝工作，并通知用户线程，进行回调。

在Java中，我们使用socket进行网络通信，IO主要有三种模式，主要看内核支持哪些。

同步阻塞IO ，每个客户端的Socket连接请求，服务端都会对应有个处理线程与之对应，对于没有分配到处理线程的连接就会被阻塞或者拒绝。相当于是一个连接一个线程。

BIO特点：

常量：

主类：

服务端监听线程：

服务端处理线程：

客户端：

同步非阻塞IO之NIO ：服务器端保存一个Socket连接列表，然后对这个列表进行轮询，如果发现某个Socket端口上有数据可读时说明读就绪，则调用该socket连接的相应读操作。如果发现某个 Socket端口上有数据可写时说明写就绪，则调用该socket连接的相应写操作。如果某个端口的Socket连接已经中断，则调用相应的析构方法关闭该端口。这样能充分利用服务器资源，效率得到了很大提高，在进行IO操作请求时候再用个线程去处理，是一个请求一个线程。Java中使用Selector、Channel、Buffer来实现上述效果。

每个线程中包含一个 Selector 对象，它相当于一个通道管理器，可以实现在一个线程中处理多个通道的目的，减少线程的创建数量。远程连接对应一个channel，数据的读写通过buffer均在同一个 channel 中完成，并且数据的读写是非阻塞的。通道创建后需要注册在 selector 中，同时需要为该通道注册感兴趣事件（客户端连接服务端事件、服务端接收客户端连接事件、读事件、写事件）， selector 线程需要采用轮训的方式调用 selector 的 select 函数，直到所有注册通道中有兴趣的事件发生，则返回，否则一直阻塞。而后循环处理所有就绪的感兴趣事件。以上步骤解决BIO的两个瓶颈：

下面对以下三个概念做一个简单介绍，Java NIO由以下三个核心部分组成：

channel和buffer有好几种类型。下面是Java NIO中的一些主要channel的实现：

正如你所看到的，这些通道涵盖了UDP和TCP网络IO，以及文件IO。以下是Java NIO里关键的buffer实现：

在微服务阶段，一个请求可能涉及到多个不同服务之间的跨服务器调用，如果你想实现高性能的PRC框架来进行数据传输，那就可以基于Java NIO做个支持长连接、自定义协议、高并发的框架，比如Netty。Netty本身就是一个基于NIO的网络框架，封装了Java NIO那些复杂的底层细节，给你提供简单好用的抽象概念来编程。比如Dubbo底层就是用的Netty。

AIO是异步非阻塞IO，相比NIO更进一步，进程读取数据时只负责发送跟接收指令，数据的准备工作完全由操作系统来处理。

推荐一个零声教育C/C++后台开发的免费公开课程，个人觉得老师讲得不错，分享给大家：C/C++后台开发高级架构师，内容包括Linux，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒体，CDN，P2P，K8S，Docker，TCP/IP，协程，DPDK等技术内容，C/C++Linux服务器开发/后台架构师【零声教育】-学习视频教程-腾讯课堂立即学习

原文：阻塞、非阻塞、多路复用、同步、异步、BIO、NIO、AIO 一锅端

java nio多路复用是什么意思

就是NIO库可以利用Selector多路复用各个Socket连接。

提高连接效率，降低连接的阻塞。

java网络io模型有几种

#BIO---Blocking IO

- 每个socket一个线程，读写时线程处于阻塞状态。

优点：实现简单

缺点：无法满足高并发，高接入的需求

- 不使用线程池的BIO模型，除了无法满足高并发需求外，由于需要为每个请求创建一个线程，还可能因为接入大量不活跃连接而耗尽服务器资源。

- 使用线程池的BIO模型，虽然控制了线程数量，但由于其本质上读写仍是阻塞的，仍无法满足高并发需求。

#NIO---Non-Blocking IO（非阻塞IO）

##非阻塞IO和多路复用

非阻塞IO和多路复用实际上是两个不用的概念，由于两者通常结合在一起使用，因此两者往往被混为一谈。下面我将试着分清这两个概念：

###非阻塞IO

与BIO相对应，非阻塞IO的读写方法无论是否有数据都立即返回，因此可以通过轮询方式来实现，但轮询方式的效率并不比BIO有显著提高，因为每个连接仍然需要占用一个线程。下面是轮询方式实现的IO模式图：

###多路复用

- 多路复用结合非阻塞IO能够明显提高IO的效率，这也是Java1.4把非阻塞IO和多路复用同时发布的原因。

- 多路复用的核心是多路复用器(Selector)，它是需要操作系统底层支持的，简单的说，就是进程把多个socket和它们关心的事件（比如连接请求或数据已准备好）都注册在多路复用器上，操作系统会在事件发生时通知多路复用器，这样进程就可以通过多路复用器知道在那个socket上发生了什么时间，从而进行对应的处理。

- 多路复用的优点在于只需要一个线程监测（阻塞或轮询方式均可）多路选择器的状态，只有在有事件需要发生时才会真正的创建线程进行处理，因此更适合高并发多接入的应用环境。

- 在Linux系统下，多路复用的底层实现是epoll方法，与select/poll的顺序扫描不同，epoll采用效率更高的事件驱动方式，而且epoll方式并没有socket个数限制。

##BIO和NIO的比较

- BIO适用于连接长期保持的应用，比如一个复杂系统中模块之间通过长连接来进行通信。

- NIO加多路复用的模式更适合短连接、高并发、多接入的情形，比如网络服务器。

##NIO网络编程的常用接口

##Reactor模式

Reactor模式用于解决事件分发处理的问题，Handler把自己的channel和关注的事件注册到Selector中，当对应的事件发生在自己的channel上时，对应的handler就会得到通知并进行处理。

- 单线程的Reactor

消息的分发、读写、处理都在一个线程中处理，是Reactor最简单的实现方式，如果消息的处理需要较长时间，会影响效率。

```java

//Reactor类，负责分发事件并调用对应的handler

class Reactor implements Runnable {

final Selector selector;

final ServerSocketChannel serverSocket;

//Reactor初始化

Reactor(int port) throws IOException {

selector = Selector.open();

serverSocket = ServerSocketChannel.open();

serverSocket.socket().bind(new InetSocketAddress(port));

serverSocket.configureBlocking(false); //必须配置为非阻塞

//Acceptor会在Reactor初始化时就注册到Selector中，用于接受connect请求

SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);

sk.attach(new Acceptor()); //attach callback object, Acceptor

}

//分发消息并调用对应的handler

public void run() {

try {

while (!Thread.interrupted()) {

selector.select();

Set selected = selector.selectedKeys();

Iterator it = selected.iterator();

while (it.hasNext())

dispatch((SelectionKey)(it.next()); //Reactor负责dispatch收到的事件

selected.clear();

}

} catch (IOException ex) { /* ... */ }

}

void dispatch(SelectionKey k) {

Runnable r = (Runnable)(k.attachment()); //调用之前注册的callback对象

if (r != null)

r.run();

}

//Acceptor也是一个handler，负责创建socket并把新建的socket也注册到selector中

class Acceptor implements Runnable { // inner

public void run() {

try {

SocketChannel c = serverSocket.accept();

if (c != null)

new Handler(selector, c);

}

catch(IOException ex) { /* ... */ }

}

//Concrete Handler：用于收发和处理消息。

//在当前的实现中，使用Runnable接口作为每个具体Handler的统一接口

//如果在处理时需要参数和返回值，也可以为Handler另外声明一个统一接口来代替Runnable接口

final class Handler implements Runnable {

final SocketChannel socket;

final SelectionKey sk;

ByteBuffer input = ByteBuffer.allocate(MAXIN);

ByteBuffer output = ByteBuffer.allocate(MAXOUT);

static final int READING = 0, SENDING = 1;

int state = READING;

Handler(Selector sel, SocketChannel c) throws IOException {

socket = c; c.configureBlocking(false);

// Optionally try first read now

sk = socket.register(sel, 0);

sk.attach(this); //将Handler作为callback对象

sk.interestOps(SelectionKey.OP_READ); //第二步,接收Read事件

sel.wakeup();

}

boolean inputIsComplete() { /* ... */ }

boolean outputIsComplete() { /* ... */ }

void process() { /* ... */ }

public void run() {

try {

if (state == READING) read();

else if (state == SENDING) send();

} catch (IOException ex) { /* ... */ }

}

void read() throws IOException {

socket.read(input);

if (inputIsComplete()) {

process();

state = SENDING;

// Normally also do first write now

sk.interestOps(SelectionKey.OP_WRITE); //第三步,接收write事件

}

void send() throws IOException {

socket.write(output);

if (outputIsComplete()) sk.cancel(); //write完就结束了, 关闭select key

}

//上面的实现用Handler来同时处理Read和Write事件, 所以里面出现状态判断

//我们可以用State-Object pattern来更优雅的实现

class Handler { // ...

public void run() { // initial state is reader

socket.read(input);

if (inputIsComplete()) {

process();

sk.attach(new Sender()); //状态迁移, Read后变成write, 用Sender作为新的callback对象

sk.interest(SelectionKey.OP_WRITE);

sk.selector().wakeup();

}

class Sender implements Runnable {

public void run(){ // ...

socket.write(output);

if (outputIsComplete()) sk.cancel();

}

```

- 多线程Reacotr

处理消息过程放在其他线程中执行

```java

class Handler implements Runnable {

// uses util.concurrent thread pool

static PooledExecutor pool = new PooledExecutor(...);

static final int PROCESSING = 3;

// ...

synchronized void read() { // ...

socket.read(input);

if (inputIsComplete()) {

state = PROCESSING;

pool.execute(new Processer()); //使用线程pool异步执行

}

synchronized void processAndHandOff() {

process();

state = SENDING; // or rebind attachment

sk.interest(SelectionKey.OP_WRITE); //process完,开始等待write事件

}

class Processer implements Runnable {

public void run() { processAndHandOff(); }

}

```

- 使用多个selector

mainReactor只负责处理accept并创建socket，多个subReactor负责处理读写请求

```java

Selector[] selectors; //subReactors集合, 一个selector代表一个subReactor

int next = 0;

class Acceptor { // ...

public synchronized void run() { ...

Socket connection = serverSocket.accept(); //主selector负责accept

if (connection != null)

new Handler(selectors[next], connection); //选个subReactor去负责接收到的connection

if (++next == selectors.length) next = 0;

}

```

#AIO

AIO是真正的异步IO，它于JDK1.7时引入，它和NIO的区别在于：

- NIO仍然需要一个线程阻塞在select方法上，AIO则不需要

- NIO得到数据准备好的消息以后，仍然需要自己把消息复制到用户空间，AIO则是通过操作系统的支持把数据异步复制到用户空间以后再给应用进程发出信号。

IO多路复用

阻塞IO只能阻塞一个IO操作，IO复用模型能阻塞多个IO操作，所以才叫多路复用

读数据：

直到数据全拷贝至User Space后才返回

不断去Kernel做Polling，询问比如读操作是否完成，没完成则read()操作会返回EWOUDBLOCK，需要过一会再尝试执行一次read()。该模式会消耗大量CPU

之前等待时间主要消耗在等数据到达上。IO Multiplexing则是将等待数据到来和读取实际数据两个事情分开，好处是通过select()等IO Multiplexing的接口一次可以等待在多个Socket上。select()返回后，处于Ready状态的Socket执行读操作时候也会阻塞，只是只阻塞将数据从Kernel拷贝到User的时间

首先注册处理函数到SIGIO信号上，在等待数据到来过程结束后，系统触发SGIO信号，之后可以在信号处理函数中执行读数据操作，再唤醒Main Thread或直接唤醒Main Thread让它完成数据读取。整个过程没有一次阻塞。

问题：TCP下，连接断开/可读/可写等都会产生Signal，并且Signal没有提供好的方法去区分这些Signal到底为什么被触发

AIO是注册一个读任务，直到读任务完全完成后才会通知应用层。AIO是由内核通知IO操作什么时候完成，信号驱动IO是由内核告知何时启动IO操作

也存在挺多问题，比如如何去cancel一个读任务

除了AIO是异步IO，其他全是同步IO

fd_set: 一个long类型的数组，每一位可以表示一个文件描述符

问题：

返回条件与select一样。

fds还是关注的描述符列表。poll将events和reevents分开了，所以如果关注的events没有发生变化就可以重用fds，poll只修改rents不会动events。fds是个数组，不是fds_set，没有了上限。

相对于select，poll解决了fds长度上限问题，解决了监听描述符无法复用问题，但仍需在poll返回后遍历fds去找ready的描述符，也要清理ready描述符对应的revents，Kernel也同样是每次poll调用需要去遍历fds注册监听，poll返回时拆除监听，也仍有惊群问题，无法动态修改描述符的问题。

使用步骤：

优点：

缺点：

changelist用于传递关心的event

nchanges用于传递changelist的大小

eventlist用于当有事件产生后，将产生的事件放在这里

nevents用于传递eventlist大小

timeout 超时时间

kqueue高级的地方在于，它监听的不一定非要是Socket，不一定非要是文件，可以是一系列事件，所以struct kevent内参数叫filter，用于过滤出关心的事件。

kqueue有epoll所有优点，还能通过changelist一次注册多个关心的event，不需要像epoll那样每次调用epoll_ctl去配置

当我们执行epoll_ctl时，除了把socket放到epoll文件系统里file对象对应的红黑树上之外，还会给内核中断处理程序注册一个回调函数，告诉内核，如果这个句柄的中断到了，就把它放到准备就绪list链表里。所以，当一个socket上有数据到了，内核在把网卡上的数据copy到内核中后就来把socket插入到准备就绪链表里。

如此，一棵红黑树，一张准备就绪句柄链表，少量的内核cache，就帮我们解决了大并发下的socket处理问题。执行epoll_create时，创建了红黑树和就绪链表，执行epoll_ctl时，如果增加socket句柄，则检查在红黑树中是否存在，存在立即返回，不存在则添加到树干上，然后向内核注册回调函数，用于当中断事件来临时向准备就绪链表中插入数据。执行epoll_wait时立刻返回准备就绪链表里的数据即可。

Epoll有两种触发模式，一种Edge Trigger简称ET，一种Level Trigger简称LT。每个使用epoll_ctl注册在epoll描述符上的被监听的描述符都能单独配置自己的触发模式。

从使用角度的区别：ET模式下当一个文件描述符Ready后，需要以Non-Blocking方式一直操作这个FD直到操作返回EAGAIN错误位置，期间Ready这个事件只会触发epoll_wait一次返回。LT模式，如果FD上的事件一直处在Ready状态没处理完，则每次调用epoll_wait都会立即返回

场景：

Java的NIO提供了Selector类，用于跨平台的实现Socket Polling，即IO多路复用。BSD系统上对应的是Kqueue，Window上对应的是Select，Linux上对应的是LT的Epoll(为了跨平台统一，Windows上背后是Select，是LT的)

Selector的使用：

如何构建一个基于netty的后端服务器

直接上干货，这个是前奏，比较山寨的实现，大家可先自行看下

下面将分析手头上一个项目，运用的技术很全，值得学习，先做一个简单介绍，当然业务部分代码就不讲了。

整个工程采用maven来管理，主要的技术是spring+jedis+netty+disruptor.看这个组合，这个服务器端性能应该很不错。

这个工程又引发我对技术无限热爱，哈哈。

这

个工程，目前主要是针对一些基于json/xml/text格式的请求，同时也是支持标准手机请求的，当然，可以自定义一些其他格式或者pc端的请求，而

且针对不同URI，后面挂了不同的handler,这些可能都是一些web处理的基本思想，只是脱离了常规的web容器或者应用服务器。

xml工具采用xstram来处理，两个字，方便。

json工具采用jackson\不知道和业界出名的fastjson\gson\sf.json有何区别，待鉴定。

客

户端的请求，统一继承ClientRequestModel，经过编码统一转化为domainMessage，交由disruptor来处理，其实oop

里什么继承，实现，封装思想，大部分都在围绕一个东西在走，一句话，把看似各有棱角的东西如何转化为共同的东西，求同存异啊（比如，水，石头，空气等，如

果在这一层，我们没法统一用一个特征来表示，我们可以先把它转化为分子，那是不是可以用同一个东西来表示呢？如何高度抽象封装，这真是一门艺术）。

看这个工程对客户端请求，是如何一步步处理的，message-request-event 交由disruptor来处理，很美妙的思想。在了解这些之前，我们有必要深入学习一下disruptor,很特别的一个框架，宣言很牛逼，中文文档在这里（），E文好的同学请移步到这里（）

了解disruptor之前，先学习下ringbuffer是如何实现的？

1、ringbuffer的特别之处：

只有一个指针，没有尾指针，基于数组，且不会删除元素，元素会覆盖，充分利用缓存行，减少垃圾回收。

2、如何从ringbuffer读取数据：

------------------------------------------2013-9-9 补充-----------------------------------------------------

下面主要讲一下请求如何处理这块架构吧，其实架构这个东西，说简单一点，就是一种简单可扩展的实现方式，在某些程度上，不要太在意性能。

底层通信建立在netty之上，基本没做任何改动

Java代码

public class HttpServerPipelineFactory implements ChannelPipelineFactory {

private ChannelUpstreamHandler channelUpstreamHandler;

public ChannelPipeline getPipeline() throws Exception {

// Create a default pipeline implementation.

ChannelPipeline pipeline = pipeline();

// Uncomment the following line if you want HTTPS

//SSLEngine engine = SecureChatSslContextFactory.getServerContext().createSSLEngine();

//engine.setUseClientMode(false);

//pipeline.addLast("ssl", new SslHandler(engine));

pipeline.addLast("decoder", new HttpRequestDecoder());

// Uncomment the following line if you don't want to handle HttpChunks.

pipeline.addLast("aggregator", new HttpChunkAggregator(1048576));

pipeline.addLast("encoder", new HttpResponseEncoder());

// Remove the following line if you don't want automatic content compression.

pipeline.addLast("deflater", new HttpContentCompressor());

//pipeline.addLast("handler", new HttpRequestHandler());

pipeline.addLast("handler", channelUpstreamHandler);

return pipeline;

}

public void setChannelUpstreamHandler(ChannelUpstreamHandler channelUpstreamHandler) {

this.channelUpstreamHandler = channelUpstreamHandler;

}

网站建设知识

java多路复用器代码 http多路复用原理

阻塞、非阻塞、多路复用、同步、异步、BIO、NIO、AIO 一文搞定

java nio多路复用是什么意思

java网络io模型有几种

IO多路复用

如何构建一个基于netty的后端服务器

其他资讯