I/O多路复用
最基本的Socket模型:可以让进程跨主机通信.
具体流程是这样：服务端开辟一个Socket,为这个Socket绑定IP地址(找到机器的网卡)和端口号(找到进程)，通过listen()函数进行监听。在客户端通过connect()函数发起连接后，服务端的Socket会维护两个队列，一个是还没有完全建立连接的队列，叫TCP半连接队列；另一个是已经建立连接(完成了三次握手)的队列，叫全连接队列。服务端会通过accept()函数从全连接队列拿出一个socket,后续传输都用这个Socket。
多进程模型
为了让一个服务端服务多个客户端，由此产生了多进程模型。
本质就是让一个父进程去处理和多个客户端的连接，然后每连一个客户端，都fork一个父进程的子进程，让这个子进程用于和新增的客户端进行数据的读写。
但这种方式弊端还是很大的，因为创建进程的开销是很大的，需要为这个进程分配虚拟内存，栈，全局变量等信息，在这个进程进行上下文切换的时候是很费劲的。
因此这个模型也无法支持太多的并发量。
多线程模型
既然进程这个单位是很重且开销很大的，因此我们可以用线程去替代线程。
同一个进程的多个线程可以去共享进程中的部分资源，如文件描述符列表，全局数据，堆等，这些共享资源是无需切换的。需要切换的只有线程的私有数据，如帮助线程上下文切换的寄存器。所以线程的创建的开销是很小的。
多线程模型的过程是这样的，父进程同样负责跟多个客户端创建连接，然后将已经创建好连接的socket放入到队列中。在这个父进程中的多个线程负责拿锁去从这个队列里面取出socket,然后往对应的取出的socket中进行读写。
但就算是这样，创建线程的开销即使不大，也没办法支撑太多的并发，只是比多进程模型更优秀。
既然为每个请求分配一个线程或者进程的方式开销都有些大，那么有可能只用一个进程取维护多个Socket呢？
下面我们就要说说I/O复用技术了，其实I/O多路复用技术优点类似于CPU时间片的利用。虽然一个进程同一时间内只能处理一个socket，但是如果说这个socket的处理时间只有1ms，那么放到1s内来看，它其实也是有100的并发量的。
select/poll/epoll是三个多路复用的接口。
select/poll/epoll
select实现多路复用的方式是，把已经连接的socket的文件描述符放入内核中，内核负责遍历这些socket，当检查有读或者写事件的时候，就把这个socket标记为可读或者可写，接着把这些文件描述符集合拷贝回用户态。用户态再遍历取出可读或者可写的socket并对其进行处理。
文件描述符集合就是每一项都指向一个打开的文件。
poll与select的区别就是poll是用链表存储。
epoll的内核对socket的存储结构和select/poll就不同了，它们是线性的存储结构，每次执行这两个方法的时候都是把整个socket传给内核。而epoll是用红黑树存储，当有一个新的需要检测的socket来临时，只需要传一个即可，大幅减少了内核和用户空间的拷贝过程。
此外，epoll的内核还维护了一个链表来记录就绪事件，当检测的socket有就绪事件发生时，就会通过回调函数把这个事件加入这个就绪事件链表当中，当用户态需要这个就绪链表的时候，只会返回有事件发生的文件描述符列表，拷贝回用户态，无需像select/poll一样,无论这个socket是否有事件发生，都全部拷贝回去。
Socket编程实现I/O多路复用模型
Server端：主线程负责处理连接事件，读线程交给异步支线线程处理，达到I/O多路复用的效果，同时这也是Netty框架的部分底层思想实现噢~
public class SimpleServer {
public static void main(String[] args) throws IOException {
System.out.println("这里是服务端");
//创建服务端Channel
ServerSocketChannel serverChannel = ServerSocketChannel.open();
//设置非阻塞
serverChannel.configureBlocking(false);
//创建Selector
Selector selector = Selector.open();
//0代表不对任何事件感兴趣
SelectionKey selectionKey = serverChannel.register(selector, 0, serverChannel);
//对连接接收事件感兴趣
selectionKey.interestOps(SelectionKey.OP_ACCEPT);
//绑定端口
serverChannel.bind(new InetSocketAddress(8080));
NioEventLoop executor = new NioEventLoop();
//主线程负责连接
while (true){
//当没有事件到来的时候，这里是阻塞的,有事件的时候会自动运行
selector.select();
//如果有事件到来，这里可以得到注册到该selector上的所有的key，每一个key上都有一个channel
Set selectionKeys = selector.selectedKeys();
//得到集合的迭代器
Iterator keyIterator = selectionKeys.iterator();
while (keyIterator.hasNext()){
//得到每一个key
SelectionKey key = keyIterator.next();
//首先要从集合中把key删除，否则会一直报告该key
keyIterator.remove();
//接下来就要处理事件，判断selector轮询到的是什么事件，并根据事件作出回应
//如果是连接事件
if(key.isAcceptable()){
//之前把服务端channel注册到selector上时候，把serverChannel放进来了
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
//接收客户端channel
SocketChannel clientChannel = channel.accept();
clientChannel.configureBlocking(false);
SelectionKey clientSocketKey = clientChannel.register(selector, 0, clientChannel);
//将客户端channel设置为可读事件
clientSocketKey.interestOps(SelectionKey.OP_READ);
System.out.println("客户端连接成功"+System.currentTimeMillis());
//worker线程开始从客户端读数据，把客户端的channel交给worker
executor.register(clientChannel,executor);
//用channel写回一条信息
clientChannel.write(ByteBuffer.wrap("服务端写回客户端成功".getBytes()));
System.out.println("向客户端发送数据成功"+System.currentTimeMillis());
}
}
}
}
}
public class NioEventLoop extends SingleThreadEventLoop{
protected Selector selector ;
public NioEventLoop() throws IOException {
this.selector = Selector.open();
}
public Selector getSelector(){
return this.selector;
}
//循环阻塞，如果有事件发生，或者队列有东西，就放行
private void select() throws IOException {
while (true){
//阻塞等待事件，如果3s都没有事件过来，可能是没有初始化。
int select = selector.select(3000);
if(select != 0 || hasTasks()){
break;
}
}
}
private void runAllTasks() {
for (;;){
Runnable task = tasksQueue.poll();
if(task == null){
break;
}
System.out.println("开始处理注册事件");
task.run();
}
}
private void processSelectedKeys() throws IOException {
System.out.println("开始处理I/O事件");
Iterator iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()){
SelectionKey key = iterator.next();
iterator.remove();
//由于连接事件被处理完了，只剩下只读事件了
if (key.isReadable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
int len = 0;
len = socketChannel.read(buffer);
if (len == -1) {
socketChannel.close();
break;
} else {
buffer.flip();
System.out.println(Charset.defaultCharset().decode(buffer).toString() + System.currentTimeMillis());
}
}
}
}
@Override
public void run(){
while (true) {
try {
//等待任务
select();
//I/O任务来了先处理I/O任务
processSelectedKeys();
} catch (IOException e) {
throw new RuntimeException(e);
}finally {
//最后处理队列任务。第一遍走过来的时候会先处理队列任务
runAllTasks();
}
}
}
}
public abstract class SingleThreadEventExecutor implements Executor {
private volatile boolean isSingle = false;
private RejectedExecutionHandler rejectedExecutionHandler;
protected Queue tasksQueue;
protected Thread thread;
//初始化的时候，1.构建线程池的属性，如队列，拒绝策略。2.构建当前线程的Selector
public SingleThreadEventExecutor() {
this.tasksQueue = new LinkedBlockingQueue(Integer.MAX_VALUE);
this.rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();
}
protected final void reject(Runnable task) {
// rejectedExecutionHandler.rejectedExecution(task, this);
}
protected boolean inEventLoop(Thread thread){
return this.thread == thread;
}
protected boolean hasTasks(){
return this.tasksQueue.isEmpty();
}
@Override
public void execute(Runnable task) {
//此时仍然是主线程。先把任务放入队列，后续处理
addTask(task);
System.out.println("任务添加完成");
startThread(task);
}
protected final void addTask(Runnable task){
this.tasksQueue.add(task);
}
private void startThread(Runnable task) {
if (isSingle) {
return;
}
isSingle = true;
System.out.println("新线程任务成功创建");
//这是个异步线程,单线程执行器的核心
new Thread(()->{
thread = Thread.currentThread();
SingleThreadEventExecutor.this.run();
System.out.println("新线程任务跑完了");
}).start();
}
protected abstract void run();
}
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor{
private void register0(SocketChannel socketChannel,NioEventLoop nioEventLoop){
try {
socketChannel.configureBlocking(false);
socketChannel.register(nioEventLoop.getSelector(), SelectionKey.OP_READ);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
//注册socketChannel到当前的selector上
public void register(SocketChannel socketChannel,NioEventLoop nioEventLoop){
//先判断当前线程是否是执行器线程，如果不是，说明是主线程，那么就还没被构造。
if(inEventLoop(Thread.currentThread())){
register0(socketChannel,nioEventLoop);
}else {
//否则是第一次注册，就先构造线程即可。
execute(()->{
register0(socketChannel,nioEventLoop);
System.out.println("executor执行完成");
});
}
}
}
