本文是《Netty、Redis、Zookeeper高并发实战》的读书笔记。
包含Netty部分的内容。未包含Redis和zookeeper的内容。
IO
Socket,文件的IO都是进程缓存区和Kernel缓存区的交换,当内核缓存需要进行真实的物理IO的时候,会发生系统中断(读中断,写中断),此时会发生进程的信息和状态保护。完成物理IO后,还需要回复进程的数据和状态信息。
4中主要的IO模型
1、同步阻塞IO(Blocking IO)
阻塞与非阻塞:kernel IO完成后才能返回用户控件执行用户的操作,否则,用户控件处于“卡死”状态(其实是在等待IO完成。阻塞指的是用户空间的状态。
同步与异步:是一种用户空间与内核空间的IO发起方式。
同步IO是指用户空间的线程是主动发起IO请求的一方,内核空间是被动接受方。
异步IO则反过来,是指系统内核是主动发起IO请求的一方,用户空间的线程是被动接受方。
传统的IO模型都是同步阻塞IO。在Java中,默认创建的socket都是阻塞的。
详细解释
直来直去的。属于过程编程。
优点:开发简单;阻塞线程会被挂起,基本不会占用CPU的资源。
2、同步非阻塞IO(Non-blocking IO)
非阻塞IO:指的是用户空间的线程不需要等待内核IO操作彻底完成,可以立即返回用户空间执行接下来的操作,即处于非阻塞状态。与此同时,内核会立即返回给用户一个状态值。
非阻塞IO要求socket被设置为NONBLOCKING。
此处的NIO(同步非阻塞IO)模型不是Java中的NIO(New IO)库。
详细解释
NIO中,一旦开始IO系统调用,会出现两种情况:
- 内核缓冲区中无数据,系统调用会立即返回,返回一个失败的信息
- 内核缓冲区中有数据,是阻塞的,直到数据从内核缓冲区复制到用户进程缓冲区。完成后,系统调用返回成功,用户进程开始处理用户空间的缓存数据。
特点
- 用户程序的线程需要不断的进行IO系统调用,轮询数据是否已经准备好,如果没有准备好,就继续轮询,直到按成IO系统调用为止。
- 每次发起IO系统调用,在内核等待数据过程中可以立即返回,用户线程不会阻塞,实时性较好
- 不断的轮询,占用大量的CPU时间,效率低下。
- 高并发场景下,同步非阻塞IO不可以用
3、IO多路复用(IO Multiplexing)
即经典的Reactor反应器设计模式,有时也称为异步阻塞IO,Java中的Selector选择器和Linux的epoll都是这种模型。
详细解释
IO多路复用,可以解决同步非阻塞IO中的轮询等待问题。
引入一种新的系统调用:查询IO的就绪状态。Linux中是select/epoll系统调用。通过该系统调用,一个线程可以见识多个文件描述符,一旦某个描述符就绪(一般是内核缓冲区可读/可写),内容能够将就绪的状态返回给应用程序。随后应用程序根据就绪状态进行相应的IO系统调用。
在IO多路复用模型中通过select/epoll系统调用,单个应用程序的线程可以不断地轮询成百上千的socket连接,当某个连接有IO就绪的状态,就返回对应的可以执行的读写操作。
过程
- 1,选择器注册。讲需要操作的目标socket连接提前注册到select/epoll的选择器中。
- 2.就绪状态的轮询。通过选择器的查询方法,查询注册过的所有socket连接的就绪状态。通过查询的系统调用,内核会返回一个就绪的socket列表。放任何一个注册过的socket中的数据准备好了,内核缓冲区有数据了,内核会降该socket加入到就绪列表中。
- 当用户线程调用了select查询方法,那么这个线程就会被阻塞掉
- 3,用户线程获得了就绪状态的列表后,根据其中的socket连接,发起read系统调用,用户线程阻塞。内核开始复制数据到用户空间缓冲区。
- 4、复制完成后,内核返回结果,用户线程解除阻塞。用户线程读取到了数据,就继续执行。
特点
-
IO多路复用模型涉及到两种系统调用:
- select/epoll的就绪查询
- IO操作(read.write)
-
与NIO相似,IO多路复用模型也需要轮询。线程需要不断的进行select/epoll轮询,查找出达到IO操作就绪的socket连接。
-
IO多路复用与NIO有密切关系,对于注册在选择器上的每一个可以查询的socket连接,一般都设置为同步非阻塞模型。
-
优点:
- 与一个线程维护一个连接的阻塞IO相比,使用select/epoll的最大优势在于,一个选择器查询线程可以同时处理成千上万的连接。大大减小了系统开销。
-
缺点:
- 本质上,select/epoll系统调用是阻塞的,属于同步IO。都需要在读写事件就绪后,由系统调用本身负责读写,也就是说这个读写过程是阻塞的。
4、异步IO(Asynchronous IO,AIO)
异步IO,指的是用户空间与内核空间的调用方式反过来。用户空间的线程变成被动接受者,而内核空间成了主动调用者。类似于Java中的回调模式,用户空间的线程向内核空间注册了各种IO事件的回调函数,由内核去主动调用。
详细解释
AIO基本流程:用户线程通过系统调用,向内核注册某个IO操作。内核在整个IO操作完成后(数据从网卡复制到用户内存缓冲区),通知用户线程继续执行后续的业务操作。
举个例子
- 1、用户线程发起一个read系统调用,然后就去做用户线程接下来的其他事情,用户线程不阻塞
- 2、内核开始IO操作,完成数据准备,复制到用户内存空间
- 3、内核给用户线程发生一个信号或者回调用户线程注册的回调接口,告诉用户线程read操作完成了
- 用户线程读取用户缓冲区的数据,完成后续的业务操作。
特点
- 用户线程非阻塞
- 用户线程需要接收内核IO完成的事件,或者用户线程需要注册一个IO操作完成的回调函数
- AIO也被称为信号驱动IO
- 缺点:
- 需要底层内核支持
- 用户程序仅需要进行事件的注册与接收,其他事情交给了操作系统
Netty使用的是IO多路复用模型。Linux底层仍使用的事epoll。
Java NIO基础详解
Java NIO
核心组件
- Channel(通道)
- Buffer(缓冲区)
- Selector(选择器)
属于IO多路复用模型。
Old IO(OIO)是面向流的,NIO是面向缓冲区的。
- OIO是面向字节流或者字符流的,一个个读取,NIO是面向缓冲区的,通过Channel和Buffer进行读取数据,可以随意读取Buffer中任意位置的数据
- OIO是阻塞的,NIO是非阻塞的
Channel
OIO中会有输入流和输出流,而NIO中的Channel可以进行读取与写入。
Selector
IO多路复用模型,指的是一个线程或者进程可以同时监视多个文件描述符,一旦其中一个或者多个文件描述符可读或者可写,系统内核就通知该线程。Selector选择器是一个IO事件的查询器。通过选择器,一个线程可以查询多个通道的IO事件的就绪状态。
首先把通道注册到Selector中,然后通过Selector内部的机制,可以查询注册的通道是否有就绪的IO事件(读/写/连接完成)。
一个选择器只需要一个线程进行监控。即,可以通过一个线程去管理多个通道。(背后是IO多路复用的支持)
Buffer
应用程序与Channel的主要交互操作就是进行数据的读写。通道的读取,就是把数据从Channel中读取到缓冲区,通道的写入,就是把数据从缓冲区写入到Channel中。
Buffer的本质是一个内存块(数组)。与普通的内存块(数组)不同的地方是,NIO Buffer对象提供了一组更加有效的读取写入的方法(可以交替读写)。
Buffer类是一个非线程安全的类。
有8中Buffer。ByteBuffer(使用最多),CharBuffer,DoubleBuffer,FloatBuffer,IntBuffer,LongBuffer,ShortBuffer,MappedByteBuffer。
Buffer类
内部有一个byte[]数组内存块,作为缓冲区。
三个重要属性:capacity(容量),position(读写位置),limit(读写限制)
标记属性:mark,可以将当前position临时存入mark中,需要的时候再恢复。
capacity:指的是写入对象的数量,不是byte[]的大小。比如写入的是IntBuffer,capacity是10,那么只能写入10个对象。
position是当前指针的位置。写入时候,position从0开始,最大值是limit,limit会被置为capacity大小。
当需要读取写入的Buffer时候,调用flip(),把写模式转换成读模式,这个时候,limit会被置为position,表示可以被读取的最大值,接着position会被重置为0,即为通道的开头了。
Buffer类几个重要方法
- allocate()创建缓冲区
intBuffer = IntBuffer.allocate(20);
- 创建后的intBuffer,处于写入模式,大小为20*4个字节
- put(),写入到缓冲区,
intBuffer.put(8);
- flip(),翻转模式;写模式翻转到读模式,会清楚mark中的值,每个模式下应该使用自己的mark,否则会混乱啊
- 读模式返回写模式:
intBuffer.clear()或者intBuffer.compact()
- get()
- rewind() 倒带,可以重复读,主要调整position的位置,mark被清除。
- mark()和reset(),mark()讲当前的position放入mark,reset是把mark恢复到position位置。
使用Buffer类基本步骤:
- 创建子类实例对象:使用allocate()
- 调用put
- 调用flip()
- 调用get()
- 调用clear()或者compact()
Channel类
一个Channel对于一个底层的文件描述符。还可以对应某个网络传输协议,Java中实现了几个,比如FileChannel(文件IO),SocketChannel(TCP连接),ServerSocketChannel(TCP连接监听),DatagramChannel(对于UDP)。
FileChannel
-
专门操作文件的
-
阻塞模式,不可设置非阻塞模式
-
文件获取,读取,写入,关闭
-
获取
String scrFile = "D:\ok.txt"; try { FileInputStream fis = new FileInputStream(scrFile); FileChannel channel = fis.getChannel(); } catch (FileNotFoundException e) { e.printStackTrace(); } try { FileOutputStream fileOutputStream = new FileOutputStream(scrFile); FileChannel channel = fileOutputStream.getChannel(); } catch (FileNotFoundException e) { e.printStackTrace(); } try { RandomAccessFile rw = new RandomAccessFile(scrFile, "rw"); FileChannel channel = rw.getChannel(); } catch (FileNotFoundException e) { e.printStackTrace(); }
-
读取
从通道中读取数据,读取的数据是要写入buffer中的,所以:
channel.read(byteBuffer);是把通道中的数据以ByteBuffer的格式写入到通道中的。
-
写入通道
buf.flip();处于读模式 channel.write(buf);这个方法是把buffer中的数据写入通道,此时buffer应该处于读状态。返回值是写入通道的字节数,长度。
-
关闭通道
channel.close();
-
强制刷新到磁盘
channel.fore(true);
SocketChannel与ServerSocketChannel
客户端:socketChannel
服务端:SocketChannel和ServerSocketChannel
都支持阻塞和非阻塞模式。
socketChannel.configureBlocking(false|true);
非阻塞模式下的通道操作是异步的。
-
获取SocketChannel
SocketChannel sc = SocketChannel.open();静态获取 sc.configureBlocking(false);设置非阻塞 sc.connect(new InetSocketAddress("127.0.0.1",8080));发起连接
由于非阻塞情况下,可能连接还没有建立,connect方法就返回了。
while(!sc.finishConnect()){ .... }
-
读取与写入
读取通道的内容,写入到buf中,与上面一致。
-
关闭通道
sc.shutdownOutput();
-
DatagramChannel
DatagramChannel.open();//创建一个udp通道
datagramChannel.configureBlocking(false);//设置为非阻塞
channel.socket().bind(new InetSocketAddress(8090));//绑定upd端口
读取数据使用receive()方法。channel.receive(buf);返回的事SocketAddress类型
写入。使用send(buf, new InetSocketAddress(IP,Port));
关闭。channel.close();
selector类
selector的使命是完成IO的多路复用。
一个通道代表一条连接通路,通过选择器可以同时监控多个通道的IO状况。选择器和通道是监控和被监控的关系。
一个单线程处理一个selector,一个selector可以监控多个Channel。
Channel和Selector之间的关系是通过注册的方式完成的。
Channel.register(Selector sel, int ops)方法可以把一个通道实例注册到一个Selector上。第一个参数是Selector实例,第二参数是Selector需要监控的IO事件类型。
IO事件类型:
- 可读:SelectionKey.OP_READ
- 可写:SelectionKey.OP_WRITE
- 连接:SelectionKey.OP_CONNECT
- 接收:SelectionKey.OP_ACCEPT
如果监控的事件是上面四种的组合,可以用“按位或”运算符来实现。
int key = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
IO操作不是对通道的IO操作,而是通道的某个IO操作的一种就绪状态,表示通道具备完成某个IO操作的条件。
比如:某个SocketChannel通道,完成了和对端的握手连接,那么该通道则处于“连接就绪”(OP_CONNECT)的状态。
通道是否可以监控
FileChannel不可以监控。能不能被监控(也就是被选择),要看该Channel是否集成SelectableChannel类。
SelectableChannel类实现了通道的可选择性所需要的的公共方法。
选择键SelectionKey
通道和选择器通过注册完成监控与被监控的关系后,就可以选择就绪事件。怎么选择呢?调用选择器的select();方法完成。
选择键:就是那些被选择器选中的IO事件(这些事件是之前被注册到Selector上的)。选择键集合中包含的都是被选择中的且注册过的IO事件。
选择键的功能强大:通过SelectionKey可以获得通道的IO事件类型,可以获得发生IO事件所在的通道,还可获得选出该选择键的选择器的实例。
选择器使用流程
- 1、获取选择器实例
- 2、把通道注册到选择器中
- 3、轮询感兴趣的IO就绪事件(选择键集合)
1、Selector selector = Selector.open();
2、将通道注册到选择器中
//获取通道
ServerSocketChannel ssc = ServerSocketChannel.open();
//设置非阻塞
ssc.configureBlocking(false);
//绑定链接
ssc.bind(new InetSocketAddress(port));
//注册到选择器上,并制定监听事件为“接收连接”事件
ssc.register(selector, SelectionKey.OP_ACCEPT);
3、选择出感兴趣的IO就绪事件
//轮询,选择感兴趣的IO就绪事件
while(selector.select() > 0){
Set selectedKeys = selector.selectedKeys();
Iterator keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()){
SelectionKey key = keyIterator.next();
//根据IO事件类型进行业务处理
if(key.isAcceptable()){
...
}else if(){
;;;;
}
//处理完后移除key
keyIterator.remove();
}
}
select()有多个重载版本。
select();//阻塞调用,一直到至少有一个通道发生了注册的IO事件
select(long timeout);//指定超时时间,同上
selectNow();//非阻塞,不管有没有IO事件,都立即返回
select();//返回值是int,表示发生了IO事件的通道数量。上次select到本次select之间发生的数量。
反应器模式(reactor模式)
反应器模式由两大角色组成:
- 1、Reactor反应器线程:负责响应IO事件,并且分发到Handlers处理器
- 2、handlers处理器:非阻塞当然执行业务处理逻辑
单线程的Reactor
reactor反应器模式类似于事件驱动模型。
当事件触发时,事件源会将事件dispatch分发到handler处理器进行处理。这个reactor的角色就类似于dispatch事件分发器。
多线程的Reactor
产生多个子反应器,和多个selector。每个线程负责一个选择器的查询和选择。
handler用到线程池。
java中的异步回调
join异步阻塞
join操作原理:阻塞当前进程,直到准备合并的目标线程执行完成。
主线程A,B线程,C线程。
A开始运行--> new B();new C(); --> B.start(); C.start();
接着A调用了B.join(); 有调用了C.join(); A 继续其他的事情。
分析:A从开始调用B.jpin();的时候,A就卡主了(阻塞了),需要B完成后,A才会继续执行C.join();这个时候,A又卡主了。C.join();执行完后,A才会继续执行。而且,B.join();C.join();执行是被合并到A线程里的,A却无法知道B,C的执行结果,不能获取B,C的返回值。
FutureTask类可以。
FutureTask异步回调之重武器
先说一下Runnable接口和Callable接口的区别。Runnable接口的run();方法没有返回值,Callable的call()有返回值,返回值是一个泛型。
FutureTask的构造函数需要传入一个Callable类型的实例,它有一个实例方法get来获取Callable的结果。
线程只有一个类Thread。那这个Callable的实例怎么执行run方法呢?
那就是FutureTask类的作用。FutureTask类内部有一个run方法。
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;//在FutureTask实例化的时候传入的Callable实例
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();//执行实例的call方法,结果存储到result上
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);//这里把call()的结果设置到FutureTask类的成员outcome中去,可以方便get到
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
这里有个事,FutureTask的实例的get方法还是阻塞的,要等call执行完才能get到,不然就会一直等待(也可以设置超时时间)。所以这个FutureTask的get方法还是异步阻塞的。谷歌的Guava框架实现了异步非阻塞调用获取线程执行结果。
Guava
继承了Java的Future接口。对异步回调机制做了增强
- 引入新的接口:ListenableFuture。使得Java的Future异步任务在Guava中能背监控和获得非阻塞异步执行的结果.
- ListenableFuture
extends Future - 增加一个方法:addListener(Runnable r, Executor e);作用是把FutureCallback的回调工作封装成一个内部的Runnable异步回调任务,在Callable异步任务完成后,回调FutureCallback进行onSuccess或者onFailure处理。
- ListenableFuture
- 引入新的接口:FutureCallback。独立的新接口。在异步任务完成后,根据结果,完成不同的回调处理,并且可以处理异步结果。
- 异步任务成功后,回调onSuccess(V result);
- 异步任务失败后,回调onFailure(Throwable var);
Guava异步回调流程
-
1、实现Java的Callable接口,创建异步执行逻辑(如果不需要返回值,那就实现Runnable接口)
-
2、创建Guava线程池
-
3、把第一步创建的Callable/Runnable异步执行逻辑的实例通过submit提交给Guava线程池,从而获得ListenableFuture异步任务实例
-
4、创建FutureCallback回调实例,通过FutureCallback讲回调实例绑定到ListenFuture异步任务上
package com.stat;
import com.google.common.util.concurrent.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class GuavaDemo {
public static final int SLEEP_GAP = 500;
public static String getCurThreadNmae() {
return Thread.currentThread().getName();
}
//业务逻辑烧水
static class HotWaterJob implements Callable<Boolean> {
@Override
public Boolean call() throws Exception {
System.out.println("shaoshui....1秒");
Thread.sleep(SLEEP_GAP);
return true;
}
}
static class WashJbo implements Callable<Boolean> {
@Override
public Boolean call() throws Exception {
System.out.println("洗杯子。。。1秒");
return true;
}
}
//新建一个一步业务类型,作为泡茶喝的主线程类
static class MainJbob implements Runnable {
boolean waterok = false;
boolean cupok = false;
int gap = SLEEP_GAP / 10;
@Override
public void run() {
while (true) {
try {
Thread.sleep(gap);
System.out.println("Doing other things");
} catch (InterruptedException e) {
e.printStackTrace();
}
if (waterok && cupok) {
drinkTea(waterok, cupok);//都准备好了,开始泡茶喝
System.out.println("喝上了茶了。。。。");
}
}
}
public void drinkTea(Boolean waterok, Boolean cupok) {
if (waterok && cupok) {
System.out.println("泡茶喝、。。。");
this.waterok = false;
this.cupok = false;
} else if (!waterok) {
System.out.println("烧水失败");
} else if (!cupok) {
System.out.println("被子未准备ok。");
}
}
}
public static void main(String[] args) {
final MainJbob mainJbob = new MainJbob();
Thread mainThread = new Thread(mainJbob);
mainThread.setName("主线程");
mainThread.start();
HotWaterJob hotWaterJob = new HotWaterJob();
WashJbo washJbo = new WashJbo();
//创建java线程池
ExecutorService jPool = Executors.newFixedThreadPool(10);
//包装java线程池,狗仔Guava线程池
ListeningExecutorService gPool = MoreExecutors.listeningDecorator(jPool);
ListenableFuture<Boolean> hotFuture = gPool.submit(hotWaterJob);
Futures.addCallback(hotFuture, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean aBoolean) {
if (aBoolean) {
mainJbob.waterok = true;
}
}
@Override
public void onFailure(Throwable throwable) {
System.out.println("烧水失败了。。。。。。。1111");
}
});
ListenableFuture<Boolean> washFuture = gPool.submit(washJbo);
Futures.addCallback(washFuture, new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean aBoolean) {
if (aBoolean) {
mainJbob.cupok = true;
}
}
@Override
public void onFailure(Throwable throwable) {
System.out.println("洗杯子失败111111111111111");
}
});
}
}