首页 文章 精选 留言 我的

精选列表

搜索[AI编程],共10000篇文章
优秀的个人博客,低调大师

Java 并发编程:Callable+Future+FutureTask详解

Runnable其中Runnable应该是我们最熟悉的接口,它只有一个run()函数,用于将耗时操作写在其中,该函数没有返回值。然后使用某个线程去执行该runnable即可实现多线程,Thread类在调用start()函数后就是执行的是Runnable的run()函数。Runnable的声明如下 : publicinterfaceRunnable{/**@seejava.lang.Thread#run()*/publicabstractvoidrun(); } CallableCallable与Runnable的功能大致相似,Callable中有一个call()函数,但是call()函数有返回值,而Runnable的run()函数不能将结果返回给客户程序。Callable的声明如下 : publicinterfaceCallable{/***Computesaresult,orthrowsanexceptionifunabletodoso. * *@returncomputedresult *@throwsExceptionifunabletocomputearesult*/Vcall()throwsException; } FutureExecutor就是Runnable和Callable的调度容器,Future就是对于具体的Runnable或者Callable任务的执行结果进行 取消、查询是否完成、获取结果、设置结果操作。get方法会阻塞,直到任务返回结果(Future简介)。Future声明如下 : publicinterfaceFuture{/***Attemptstocancelexecutionofthistask.Thisattemptwill *failifthetaskhasalreadycompleted,hasalreadybeencancelled, *orcouldnotbecancelledforsomeotherreason.Ifsuccessful, *andthistaskhasnotstartedwhencanceliscalled, *thistaskshouldneverrun.Ifthetaskhasalreadystarted, *thenthemayInterruptIfRunningparameterdetermines *whetherthethreadexecutingthistaskshouldbeinterruptedin *anattempttostopthetask.*/booleancancel(booleanmayInterruptIfRunning);/***Returnstrueifthistaskwascancelledbeforeitcompleted *normally.*/booleanisCancelled();/***Returnstrueifthistaskcompleted. **/booleanisDone();/***Waitsifnecessaryforthecomputationtocomplete,andthen *retrievesitsresult. * *@returnthecomputedresult*/Vget()throwsInterruptedException,ExecutionException;/***Waitsifnecessaryforatmostthegiventimeforthecomputation *tocomplete,andthenretrievesitsresult,ifavailable. * *@paramtimeoutthemaximumtimetowait *@paramunitthetimeunitofthetimeoutargument *@returnthecomputedresult*/Vget(longtimeout,TimeUnitunit)throwsInterruptedException,ExecutionException,TimeoutException; } FutureTask FutureTask是一个RunnableFuture public class FutureTaskRunnableFuture实现了Runnbale又实现了Futrue publicinterfaceRunnableFutureextendsRunnable,Future{/***SetsthisFuturetotheresultofitscomputation *unlessithasbeencancelled.*/voidrun(); } 另外FutureTaslk还可以包装Runnable和Callable,由构造函数注入依赖。publicFutureTask(Callablecallable){if(callable==null)thrownewNullPointerException();this.callable=callable;this.state=NEW;//ensurevisibilityofcallable}publicFutureTask(Runnablerunnable,Vresult){this.callable=Executors.callable(runnable,result);this.state=NEW;//ensurevisibilityofcallable} 上面代码块可以看出:Runnable注入会被Executors.callable()函数转换为Callable类型,即FutureTask最终都是执行Callable类型的任务。该适配函数的实现如下 : publicstaticCallablecallable(Runnabletask,Tresult){if(task==null)thrownewNullPointerException();returnnewRunnableAdapter(task,result); } RunnableAdapter适配器 /** *Acallablethatrunsgiventaskandreturnsgivenresult*/staticfinalclassRunnableAdapterimplementsCallable{finalRunnabletask;finalTresult; RunnableAdapter(Runnabletask,Tresult){this.task=task;this.result=result; }publicTcall(){ task.run();returnresult; } } FutureTask实现Runnable,所以能通过Thread包装执行, FutureTask实现Runnable,所以能通过提交给ExcecuteService来执行 注:ExecuteService:创建线程池实例对象,其中有submit(Runnable)、submit(Callable)方法 还可以直接通过get()函数获取执行结果,该函数会阻塞,直到结果返回。 因此FutureTask是Future也是Runnable,又是包装了的Callable( 如果是Runnable最终也会被转换为Callable )。 Callable和Future接口的区别 1.Callable规定的方法是call(),而Runnable规定的方法是run().2.Callable的任务执行后可返回值,而Runnable的任务是不能返回值的。3.call()方法可抛出异常,而run()方法是不能抛出异常的。4.运行Callable任务可拿到一个Future对象,Future表示异步计算的结果。5.它提供了检查计算是否完成的方法,以等待计算的完成,并检索计算的结果。6.通过Future对象可了解任务执行情况,可取消任务的执行,还可获取任务执行的结果。7.Callable是类似于Runnable的接口,实现Callable接口的类和实现Runnable的类都是可被其它线程执行的任务。 示例: packagecom.xzf.callable;importjava.util.concurrent.Callable;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;importjava.util.concurrent.Future;importjava.util.concurrent.FutureTask;publicclassRunnableFutureTask{staticExecutorServiceexecutorService=Executors.newSingleThreadExecutor();//创建一个单线程执行器publicstaticvoidmain(String[]args){ runnableDemo(); futureDemo(); }/***newThread(Runnablearg0).start();用Thread()方法开启一个新线程 *runnable,无返回值*/staticvoidrunnableDemo(){newThread(newRunnable(){ publicvoidrun(){ System.out.println("runnabledemo:"+fibc(20));//有值} }).start(); }/***Runnable实现的是voidrun()方法,无返回值 *Callable实现的是Vcall()方法,并且可以返回执行结果 *Runnable可以提交给Thread,在包装下直接启动一个线程来执行 *Callable一般都是提交给ExecuteService来执行*/staticvoidfutureDemo(){try{ Futureresult1=executorService.submit(newRunnable(){publicvoidrun(){ fibc(20); } }); System.out.println("futureresultfromrunnable:"+result1.get());//run()无返回值所以为空,result1.get()方法会阻塞Futureresult2=executorService.submit(newCallable(){publicIntegercall()throwsException{returnfibc(20); } }); System.out.println("futureresultfromcallable:"+result2.get());//call()有返回值,result2.get()方法会阻塞FutureTaskresult3=newFutureTask(newCallable(){publicIntegercall()throwsException{returnfibc(20); } }); executorService.submit(result3); System.out.println("futureresultfromFutureTask:"+result3.get());//call()有返回值,result3.get()方法会阻塞/*因为FutureTask实现了Runnable,因此它既可以通过Thread包装来直接执行,也可以提交给ExecuteService来执行*/FutureTaskresult4=newFutureTask(newRunnable(){publicvoidrun(){ fibc(20); } },fibc(20)); executorService.submit(result4); System.out.println("futureresultfromexecuteServiceFutureTask:"+result4.get());//call()有返回值,result3.get()方法会阻塞//这里解释一下什么FutureTask实现了Runnable结果不为null,这就用到FutureTask对Runnable的包装,所以Runnable注入会被Executors.callable()函数转换成Callable类型FutureTaskresult5=newFutureTask(newRunnable(){publicvoidrun(){ fibc(20); } },fibc(20));newThread(result5).start(); System.out.println("futureresultfromThreadFutureTask:"+result5.get());//call()有返回值,result5.get()方法会阻塞}catch(Exceptione){ e.printStackTrace(); }finally{ executorService.shutdown(); } }staticintfibc(intnum){if(num==0){return0; }if(num==1){return1; }returnfibc(num-1)+fibc(num-2); } }

优秀的个人博客,低调大师

【博客大赛】+ 网络编程Netty之ByteBuf详解

Netty中的ByteBuf优势 NIO使用的ByteBuffer有哪些缺点 1: 无法动态扩容,ByteBuffer的长度是固定的,是初始指定的值,不能够再进行扩容了,当写入的内容大于ByteBuffer的容量时,会报越界异常 2.: API使用复杂,当要读取数据时,需要调用buffer.flip()方法,转换为读取模式,如果稍微不注意就可能出现错误,读取不到数据或者读取的数据是错误的 ByteBuf的优势和做了哪些增强 1: API操作起来更加的方便,可以直接写或者直接读 2:支持动态扩容,当写入的数据大于ByteBuf的容量时,会动态扩容,不会报错 3:提供了多种ByteBuf的实现,可以更加灵活的使用 4:提供了高效的零拷贝机制 5:ByteBuf可以内存复用 ByteBuf操作示例 ByteBuf操作 ==ByteBuf中有三个重要的属性:==1:capacity容量,初始指定的ByteBuf的大小 2:readIndex读取位置,顺序读的时候,记录读取数据的索引值 3:writeIndex写入位置,顺序写的时候,记录写入数据的索引值 ==ByteBuf常用的方法:==1:getByte和setByte,获取指定索引处的数据,是随机获取的,不会改变readIndex和writeIndex的值 2:read*,顺序读,会改变readIndex的值 3:write*,顺序写,会改变writeIndex的值 4:discardReadBytes,清除读过的内容 5:clear,清除缓冲区 6:搜索操作 7:标记和重置 8:引用计数和释放 简单的Demo示例 /** * ByteBuf的使用示例 */ public class ByteBufDemo { public static void main(String[] args) { //分配非池化,10个字节的ByteBuf ByteBuf buf = Unpooled.buffer(10); //看下ByteBuf System.out.println("------------------------原始的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); //写入内容到ByteBuf byte[] bytes = {1, 2, 3, 4, 5}; buf.writeBytes(bytes); System.out.println("------------------------写入内容后的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); //从ByteBuf中读取内容 buf.readByte(); buf.readByte(); System.out.println("------------------------读取一些内容后的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); //清除读过的内容 //把读过的数据清除后,readIndex变为0,writeIndex变为3 //后面尚未读取的内容,会复制到前面去,把原来的值覆盖掉 //再次写入时,3,4,5后面的4,5会被覆盖掉 buf.discardReadBytes(); System.out.println("------------------------清除读过的数据后的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); //再次写入内容到ByteBuf byte[] bytesO = {6}; buf.writeBytes(bytesO); System.out.println("------------------------再次写入内容后的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); //清空读和写的索引值 //readIndex和writeIndex会重置为0,ByteBuf中的内容并不会重置 buf.clear(); System.out.println("------------------------清空读和写的索引值后的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); //再次写入内容到ByteBuf byte[] bytes2 = {1, 2, 3}; buf.writeBytes(bytes2); System.out.println("------------------------再次写入内容后的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); //清空ByteBuf的内容 //不会重置readIndex和writeIndex buf.setZero(0, buf.capacity()); System.out.println("------------------------清空ByteBuf的内容后的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); //再次写入超出指定容量的数据到ByteBuf //会进行扩容 byte[] bytes3 = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12}; buf.writeBytes(bytes3); System.out.println("------------------------再次写入超出指定容量的数据后的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); System.out.println("ByteBuf中的内容:" + Arrays.toString(buf.array()) + "\n"); } } 输出结果:上面的例子是使用堆内的ByteBuf,下面看下堆外的ByteBuf例子: //分配非池化,10个字节的directBuffer ByteBuf buf = Unpooled.directBuffer(10); //看下ByteBuf System.out.println("------------------------原始的ByteBuf-------------------------------"); System.out.println("ByteBuf参数:" + buf.toString()); directBuffer不能够使用array方法,否则会报错:java.lang.UnsupportedOperationException: direct buffer;而且使用ByteBuf是用它底层的分配器分配的,不是new一个出来,下面会具体说下。上图中,可以看到,readIndex和writeIndex把缓冲区分成了三块,readIndex会小于或者等于writeIndex,这个应该好理解,我还没有写到那里,你就去读取了,能读取到什么呢。 堆内和堆外内存 socket是操作系统底层提供给上层应用使用的网络通信API,当要去读取或者写入的数据在JVM的堆中,那么就先需要把JVM堆中需要读取的数据拷贝一份到操作系统中,然后socket再去读取,而直接内存的好处是socket可以直接读取,少了拷贝这一步操作。 ByteBuf动态扩容 下面以堆内的ByteBuf为例,查看源码,分析ByteBuf的动态扩容:动态扩容肯定是写入数据的时候,ByteBuf的容量不够了,才去扩容的,所以需要跟踪下面的代码: buf.writeBytes(bytes); 跟踪上面的writeBytes,首先进入了ByteBuf这个抽象类中,进入了下面这个抽象方法: public abstract ByteBuf writeBytes(byte[] src); 它的实现类如下:进入第一个AbstractByteBuf的方法: @Override public ByteBuf writeBytes(byte[] src) { writeBytes(src, 0, src.length); return this; } 再次调用了下面的方法: @Override public ByteBuf writeBytes(byte[] src, int srcIndex, int length) { //检查是否可以写入 ensureWritable(length); setBytes(writerIndex, src, srcIndex, length); //把当前的写入位置加上写入数据的长度 writerIndex += length; return this; } src是需要写入的数据,length是写入数据的长度然后会进入ensureWritable方法,传入的参数是:写入数据的长度 @Override public ByteBuf ensureWritable(int minWritableBytes) { //参数校验 checkPositiveOrZero(minWritableBytes, "minWritableBytes"); //检查容量是否可以写入这么多数据 ensureWritable0(minWritableBytes); return this; } //检查参数是否小于0 public static int checkPositiveOrZero(int i, String name) { if (i < 0) { throw new IllegalArgumentException(name + ": " + i + " (expected: >= 0)"); } return i; } 参数校验完成后会进入ensureWritable0方法: final void ensureWritable0(int minWritableBytes) { //确保缓冲区可以访问 ensureAccessible(); //如果写入的数据长度小于等于剩余可写数据的容量,就直接返回 //就是说,容量足够写入,不需要扩容 if (minWritableBytes <= writableBytes()) { return; } if (checkBounds) { //maxCapacity是int的最大值 //检查写入的数据长度是否比可以写入的最大容量还要大 //是的话就抛异常 if (minWritableBytes > maxCapacity - writerIndex) { throw new IndexOutOfBoundsException(String.format( "writerIndex(%d) + minWritableBytes(%d) exceeds maxCapacity(%d): %s", writerIndex, minWritableBytes, maxCapacity, this)); } } //正式的扩容方法 int newCapacity = alloc().calculateNewCapacity(writerIndex + minWritableBytes, maxCapacity); //把扩容后的新容量设置进去 capacity(newCapacity); } 进入AbstractByteBufAllocator类的扩容方法: //常量 4M static final int CALCULATE_THRESHOLD = 1048576 * 4; // 4 MiB page @Override public int calculateNewCapacity(int minNewCapacity, int maxCapacity) { //校验参数 checkPositiveOrZero(minNewCapacity, "minNewCapacity"); //minNewCapacity = writerIndex + minWritableBytes //已经写入的数据索引加上当前写入的数据长度,就是需要的最小的容量 //判断是否比最大容量还大,是的话就抛异常 if (minNewCapacity > maxCapacity) { throw new IllegalArgumentException(String.format( "minNewCapacity: %d (expected: not greater than maxCapacity(%d)", minNewCapacity, maxCapacity)); } final int threshold = CALCULATE_THRESHOLD; // 4 MiB page //如果需要的最小容量等于4M,就直接返回4M,作为扩容后的容量 if (minNewCapacity == threshold) { return threshold; } //如果需要的最小容量大于4M,就按照下面的扩容方式扩容 if (minNewCapacity > threshold) { //newCapacity = 15 / 4194304 * 4194304 int newCapacity = minNewCapacity / threshold * threshold; //如果计算出的容量大于最大容量减去4M,就把最大容量赋值给新的容量 if (newCapacity > maxCapacity - threshold) { newCapacity = maxCapacity; } else { newCapacity += threshold; } return newCapacity; } //如果需要的最小容量小于4M,就按照下面的方式扩容 int newCapacity = 64; while (newCapacity < minNewCapacity) { newCapacity <<= 1; } return Math.min(newCapacity, maxCapacity); } 再看下capacity方法:下面的把扩容后的容量放到ByteBuf,就是使用了arraycopy方法 @Override public ByteBuf capacity(int newCapacity) { checkNewCapacity(newCapacity); int oldCapacity = array.length; byte[] oldArray = array; if (newCapacity > oldCapacity) { byte[] newArray = allocateArray(newCapacity); System.arraycopy(oldArray, 0, newArray, 0, oldArray.length); setArray(newArray); freeArray(oldArray); } else if (newCapacity < oldCapacity) { byte[] newArray = allocateArray(newCapacity); int readerIndex = readerIndex(); if (readerIndex < newCapacity) { int writerIndex = writerIndex(); if (writerIndex > newCapacity) { writerIndex(writerIndex = newCapacity); } System.arraycopy(oldArray, readerIndex, newArray, readerIndex, writerIndex - readerIndex); } else { setIndex(newCapacity, newCapacity); } setArray(newArray); freeArray(oldArray); } return this; } 下面是跟踪的代码步骤:==总结下动态扩容机制:==1:write*方法调用的时候,会通过ensureWritable0方法检查2:calculateNewCapacity方法是用来计算容量的方法 ==扩容计算方法:==1:需要的容量没有超过4M,会从64字节开始扩容,每次增加一倍,直到计算出来的容量满足需要的最小容量,假如,当前大小是256,已经写入了200字节,再次写入60字节,需要的最小容量是260字节,那么扩容后的容量是64 2 2 2=5122:需要的容量超过4M,扩容计算方法为:新容量 = 新容量的最小要求 / 4M 4M + 4M,假如当前大小是3M,已经写了2M,再写入3M,需要的最小容量是5M,那么扩容后的容量是 5 / 4 * 4 + 4 = 8M 图示1:需要的容量小于4M:图示2:需要的容量大于4M: ByteBuf有哪些实现 ByteBuf从3个维度,有8种实现方式: ByteBuf类图 //堆内 ByteBuf buf = Unpooled.buffer(10); //堆外 ByteBuf buf = Unpooled.directBuffer(10); ByteBuf提供了Unpooled非池化的类,可以直接使用,没有提供Pool池化的类,下面追踪源码看下ByteBuf是怎样分配的: Unpooled.buffer分配方式 首先进入Unpooled类: private static final ByteBufAllocator ALLOC = UnpooledByteBufAllocator.DEFAULT; //使用默认的分配器分配堆内buffer public static ByteBuf buffer(int initialCapacity) { return ALLOC.heapBuffer(initialCapacity); } 下面会进入接口类ByteBufAllocator: //分配一个指定容量的堆内buf ByteBuf heapBuffer(int initialCapacity); 然后进入AbstractByteBufAllocator抽象类: //如果没有指定初始容量,默认的初始容量大小是256 static final int DEFAULT_INITIAL_CAPACITY = 256; //最大容量,为int的最大值 static final int DEFAULT_MAX_CAPACITY = Integer.MAX_VALUE; @Override public ByteBuf heapBuffer(int initialCapacity) { return heapBuffer(initialCapacity, DEFAULT_MAX_CAPACITY); } @Override public ByteBuf heapBuffer(int initialCapacity, int maxCapacity) { //如果初始化的容量是0,最大的容量也是0,就返回一个空的Buf if (initialCapacity == 0 && maxCapacity == 0) { return emptyBuf; } validate(initialCapacity, maxCapacity); return newHeapBuffer(initialCapacity, maxCapacity); } //校验参数 private static void validate(int initialCapacity, int maxCapacity) { //检查参数 checkPositiveOrZero(initialCapacity, "initialCapacity"); //如果初始化的容量大于最大容量,就抛异常 if (initialCapacity > maxCapacity) { throw new IllegalArgumentException(String.format( "initialCapacity: %d (expected: not greater than maxCapacity(%d)", initialCapacity, maxCapacity)); } } 然后是newHeapBuffer抽象方法: protected abstract ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity); 因为这里初始化的是非池化的,所以会进入UnpooledByteBufAllocator类: @Override protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) { //PlatformDependent.hasUnsafe()是检查当前操作系统是否支持unsafe操作 //根据支持与否,进入不同的类 return PlatformDependent.hasUnsafe() ? new InstrumentedUnpooledUnsafeHeapByteBuf(this, initialCapacity, maxCapacity) : new InstrumentedUnpooledHeapByteBuf(this, initialCapacity, maxCapacity); } 支持Unsafe操作的进入下面: InstrumentedUnpooledUnsafeHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) { super(alloc, initialCapacity, maxCapacity); } 不支持Unsafe的进入下面这个: InstrumentedUnpooledHeapByteBuf(UnpooledByteBufAllocator alloc, int initialCapacity, int maxCapacity) { super(alloc, initialCapacity, maxCapacity); } 现在以支持Unsafe操作往下面走,进入UnpooledUnsafeHeapByteBuf类: UnpooledUnsafeHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) { super(alloc, initialCapacity, maxCapacity); } 再次调用了父类UnpooledHeapByteBuf: //分配器 private final ByteBufAllocator alloc; //byte数组,ByteBuf数据底层就是使用这个存储 byte[] array; public UnpooledHeapByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) { super(maxCapacity); //检查分配器是否为空 checkNotNull(alloc, "alloc"); //如果初始化的容量大于最大容量,就抛异常 if (initialCapacity > maxCapacity) { throw new IllegalArgumentException(String.format( "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity)); } this.alloc = alloc; //设置当前的数组是分配之后的数组 setArray(allocateArray(initialCapacity)); //初始化readIndex和writeIndex setIndex(0, 0); } //分配数组 protected byte[] allocateArray(int initialCapacity) { //返回一个具有initialCapacity容量大小的byte数组 return new byte[initialCapacity]; } //set数组 private void setArray(byte[] initialArray) { array = initialArray; tmpNioBuf = null; } AbstractByteBuf类下的setIndex方法: //初始化readerIndex和writerIndex @Override public ByteBuf setIndex(int readerIndex, int writerIndex) { if (checkBounds) { checkIndexBounds(readerIndex, writerIndex, capacity()); } setIndex0(readerIndex, writerIndex); return this; } final void setIndex0(int readerIndex, int writerIndex) { this.readerIndex = readerIndex; this.writerIndex = writerIndex; } 上面走到AbstractByteBuf后,就分配完了一个非池化、堆内的ByteBuf,下面是追踪的代码:==总结:==可以看到,分配一个非池化、堆内的ByteBuf,它的底层就是byte数组 Unpooled.directBuffer分配方式 首先进入的也是Unpooled类: public static ByteBuf directBuffer(int initialCapacity) { return ALLOC.directBuffer(initialCapacity); } 然后进入ByteBufAllocator抽象类: ByteBuf directBuffer(int initialCapacity); 然后到AbstractByteBufAllocator类: @Override public ByteBuf directBuffer(int initialCapacity) { return directBuffer(initialCapacity, DEFAULT_MAX_CAPACITY); } @Override public ByteBuf directBuffer(int initialCapacity, int maxCapacity) { //如果初始化的容量和最大容量都是0,就返回一个空的Buf if (initialCapacity == 0 && maxCapacity == 0) { return emptyBuf; } //校验参数 validate(initialCapacity, maxCapacity); return newDirectBuffer(initialCapacity, maxCapacity); } protected abstract ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity); 由于分配的也是一个非池化的,所以newDirectBuffer会进入UnpooledByteBufAllocator类中的实现类: @Override protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) { final ByteBuf buf; //同样的,会判断是否支持unsafe操作 if (PlatformDependent.hasUnsafe()) { buf = noCleaner ? new InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf(this, initialCapacity, maxCapacity) : new InstrumentedUnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity); } else { buf = new InstrumentedUnpooledDirectByteBuf(this, initialCapacity, maxCapacity); } return disableLeakDetector ? buf : toLeakAwareBuffer(buf); } 以InstrumentedUnpooledUnsafeNoCleanerDirectByteBuf为例,后面两个其实也相差不大,进入UnpooledUnsafeNoCleanerDirectByteBuf类的构造方法: UnpooledUnsafeNoCleanerDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) { super(alloc, initialCapacity, maxCapacity); } 再次调用的父类UnpooledUnsafeDirectByteBuf: ByteBuffer buffer; public UnpooledUnsafeDirectByteBuf(ByteBufAllocator alloc, int initialCapacity, int maxCapacity) { super(maxCapacity); if (alloc == null) { throw new NullPointerException("alloc"); } //校验参数 checkPositiveOrZero(initialCapacity, "initialCapacity"); checkPositiveOrZero(maxCapacity, "maxCapacity"); if (initialCapacity > maxCapacity) { throw new IllegalArgumentException(String.format( "initialCapacity(%d) > maxCapacity(%d)", initialCapacity, maxCapacity)); } this.alloc = alloc; setByteBuffer(allocateDirect(initialCapacity), false); } //分配的是一个NIO中的ByteBuffer protected ByteBuffer allocateDirect(int initialCapacity) { return ByteBuffer.allocateDirect(initialCapacity); } final void setByteBuffer(ByteBuffer buffer, boolean tryFree) { if (tryFree) { ByteBuffer oldBuffer = this.buffer; if (oldBuffer != null) { if (doNotFree) { doNotFree = false; } else { freeDirect(oldBuffer); } } } this.buffer = buffer; memoryAddress = PlatformDependent.directBufferAddress(buffer); tmpNioBuf = null; capacity = buffer.remaining(); } ByteBuffer类下面的allocateDirect: public static ByteBuffer allocateDirect(int capacity) { return new DirectByteBuffer(capacity); } 代码跟踪图:==总结:==分配非池化、堆外的ByteBuf,可以看到底层是NIO的DirectByteBuffer实现的 ByteBufAllocator类图 ByteBuf内存复用 分配池化内存 在上面根据源码知道了怎么去分配非池化内存,那么池化内存要怎么分配呢?看下面的图示:上面就是分配池化内存的步骤,接下来会根据源码具体分析 内存缓存池 jemalloc内存分配机制1:内存池中有三大区域,分别是:tiny、small、normal2:每个区域分了不同大小的格子,每个格子只能缓存对应大小的内存块3:支持最大的格子内存是32kb,超过这个大小的不能被缓存,只能被释放掉4:每个类型的格子都有对应的数量:tiny:512个,small:256个,normal:64个,例如tiny区域的每个大小的格子都有512个,如果满了就不会被回收,内存会被释放掉 回收池化内存 分配池化内存的过程 上面分析了分配非池化内存,下面看下怎么分配池化内存: ByteBufAllocator allocator = ByteBufAllocator.DEFAULT; //分配的内存最大长度为496 ByteBuf buf1 = allocator.ioBuffer(495); System.out.printf("buf1: 0x%X%n", buf1.memoryAddress()); //此时会被回收到tiny的512b格子中 buf1.release(); //从tiny的512b格子去取 ByteBuf buf2 = allocator.ioBuffer(495); System.out.printf("buf2: 0x%X%n", buf2.memoryAddress()); buf2.release(); 先来看下ByteBufAllocator类: //默认ByteBuf分配器,在ByteBufUtil中初始化 ByteBufAllocator DEFAULT = ByteBufUtil.DEFAULT_ALLOCATOR; 跟踪第一次的allocator.ioBuffer(495)代码,首先进入AbstractByteBufAllocator类: @Override public ByteBuf ioBuffer(int initialCapacity) { //如果支持Unsafe,就分配堆外内存 if (PlatformDependent.hasUnsafe()) { return directBuffer(initialCapacity); } //不支持Unsafe,就分配堆内内存 return heapBuffer(initialCapacity); } 然后调用了该类下面的directBuffer方法: @Override public ByteBuf directBuffer(int initialCapacity) { return directBuffer(initialCapacity, DEFAULT_MAX_CAPACITY); } @Override public ByteBuf directBuffer(int initialCapacity, int maxCapacity) { //如果初始化的容量和最大容量等于0,就返回一个空的ByteBuf if (initialCapacity == 0 && maxCapacity == 0) { return emptyBuf; } validate(initialCapacity, maxCapacity); return newDirectBuffer(initialCapacity, maxCapacity); } //校验参数 private static void validate(int initialCapacity, int maxCapacity) { checkPositiveOrZero(initialCapacity, "initialCapacity"); if (initialCapacity > maxCapacity) { throw new IllegalArgumentException(String.format( "initialCapacity: %d (expected: not greater than maxCapacity(%d)", initialCapacity, maxCapacity)); } } 然后会进入池化的ByteBuf分配器PooledByteBufAllocator类,可以实现内存的复用: // cache sizes 缓存默认值 DEFAULT_TINY_CACHE_SIZE = SystemPropertyUtil.getInt("io.netty.allocator.tinyCacheSize", 512); DEFAULT_SMALL_CACHE_SIZE = SystemPropertyUtil.getInt("io.netty.allocator.smallCacheSize", 256); DEFAULT_NORMAL_CACHE_SIZE = SystemPropertyUtil.getInt("io.netty.allocator.normalCacheSize", 64); @Override protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) { //从当前线程中获取cache对象 PoolThreadCache cache = threadCache.get(); //从cache中获取Arena //Arena可以理解为一个netty提供的实际进行buf分配和管理的工具 PoolArena<ByteBuffer> directArena = cache.directArena; final ByteBuf buf; //如果有directArena就分配池化内存 if (directArena != null) { buf = directArena.allocate(cache, initialCapacity, maxCapacity); } else { //如果没有directArena,就使用非池化Unpooled buf = PlatformDependent.hasUnsafe() ? UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) : new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity); } return toLeakAwareBuffer(buf); } 再次跟踪后进入PoolArena类:可以看到下面有三种类型tiny、small、normal enum SizeClass { Tiny, Small, Normal } PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) { //获取一个ByteBuf对象 PooledByteBuf<T> buf = newByteBuf(maxCapacity); //分配内存 allocate(cache, buf, reqCapacity); return buf; } @Override protected PooledByteBuf<ByteBuffer> newByteBuf(int maxCapacity) { //如果支持Unsafe,就初始化一个PooledUnsafeDirectByteBuf if (HAS_UNSAFE) { return PooledUnsafeDirectByteBuf.newInstance(maxCapacity); } else { //不支持Unsafe,就初始化一个PooledDirectByteBuf return PooledDirectByteBuf.newInstance(maxCapacity); } } 下面进入PooledUnsafeDirectByteBuf类:从线程回收栈中获取一个buf,如果栈中没有,就会创建一个新的,如果有,就会返回栈中的buf //调用RECYCLER.get()时,线程栈中没有可以复用的时,会调用newObject方法,此时创建出来的buf是空的 private static final Recycler<PooledUnsafeDirectByteBuf> RECYCLER = new Recycler<PooledUnsafeDirectByteBuf>() { @Override protected PooledUnsafeDirectByteBuf newObject(Handle<PooledUnsafeDirectByteBuf> handle) { return new PooledUnsafeDirectByteBuf(handle, 0); } }; static PooledUnsafeDirectByteBuf newInstance(int maxCapacity) { //RECYCLER,回收机制 PooledUnsafeDirectByteBuf buf = RECYCLER.get(); //取出来的可能是之前的buf,使用之前清理一下 buf.reuse(maxCapacity); return buf; } 然后再次回到PoolArena类中的allocate方法,分配内存: private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) { //将需要的内存大小计算为2^n final int normCapacity = normalizeCapacity(reqCapacity); //需要分配的内存是否是tiny或者small类型 if (isTinyOrSmall(normCapacity)) { // capacity < pageSize int tableIdx; PoolSubpage<T>[] table; boolean tiny = isTiny(normCapacity); if (tiny) { // < 512 //分配一个tiny内存 if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) { // was able to allocate out of the cache so move on return; } tableIdx = tinyIdx(normCapacity); table = tinySubpagePools; } else { if (cache.allocateSmall(this, buf, reqCapacity, normCapacity)) { // was able to allocate out of the cache so move on return; } tableIdx = smallIdx(normCapacity); table = smallSubpagePools; } final PoolSubpage<T> head = table[tableIdx]; synchronized (head) { final PoolSubpage<T> s = head.next; if (s != head) { assert s.doNotDestroy && s.elemSize == normCapacity; long handle = s.allocate(); assert handle >= 0; s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity); incTinySmallAllocation(tiny); return; } } synchronized (this) { //分配一块新的内存 allocateNormal(buf, reqCapacity, normCapacity); } incTinySmallAllocation(tiny); return; } if (normCapacity <= chunkSize) { if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) { // was able to allocate out of the cache so move on return; } synchronized (this) { allocateNormal(buf, reqCapacity, normCapacity); ++allocationsNormal; } } else { // Huge allocations are never served via the cache so just call allocateHuge allocateHuge(buf, reqCapacity); } } PoolThreadCache类下的allocateTiny方法: boolean allocateTiny(PoolArena<?> area, PooledByteBuf<?> buf, int reqCapacity, int normCapacity) { return allocate(cacheForTiny(area, normCapacity), buf, reqCapacity); } //从cache中获取buf private MemoryRegionCache<?> cacheForTiny(PoolArena<?> area, int normCapacity) { int idx = PoolArena.tinyIdx(normCapacity); if (area.isDirect()) { return cache(tinySubPageDirectCaches, idx); } return cache(tinySubPageHeapCaches, idx); } 根据需要的容量获取对应的格子,走到PoolArena类下面的tinyIdx方法: static int tinyIdx(int normCapacity) { return normCapacity >>> 4; } PoolThreadCache类下的allocate方法,把缓存格子的内存分配到buf private boolean allocate(MemoryRegionCache<?> cache, PooledByteBuf buf, int reqCapacity) { if (cache == null) { // no cache found so just return false here return false; } boolean allocated = cache.allocate(buf, reqCapacity); if (++ allocations >= freeSweepAllocationThreshold) { allocations = 0; trim(); } return allocated; } 下面是具体跟踪代码的步骤图:上面的源码是以tiny类型为例,其他两种类型类似,当第一次分配创建了一块新的内存,然后被成功回收到内存缓冲池后,再次分配对应大小的内存,会直接从内存缓冲池中取,不会再次分配一块新的内存了 内存回收的过程 接下来跟踪release()方法,看下内存回收的过程 buf1.release(); 第一次进入AbstractReferenceCountedByteBuf类:Buf的引用计数器,用于内存复用,有一个计数器refCnt,retain()计数器加一,release()计数器减一,直到计数器为0,才调用deallocate()释放,deallocate()方法由具体的buf自己实现。 @Override public boolean release() { return release0(1); } private boolean release0(int decrement) { int rawCnt = nonVolatileRawCnt(), realCnt = toLiveRealCnt(rawCnt, decrement); //判断当前buf有没有被引用了,没有的话就调用deallocate if (decrement == realCnt) { if (refCntUpdater.compareAndSet(this, rawCnt, 1)) { deallocate(); return true; } return retryRelease0(decrement); } return releaseNonFinal0(decrement, rawCnt, realCnt); } 进入PooledByteBuf类: @Override protected final void deallocate() { if (handle >= 0) { final long handle = this.handle; //表示当前的buf不在使用任何一块内存区域 this.handle = -1; //设置memory为null memory = null; //释放buf的内存 chunk.arena.free(chunk, tmpNioBuf, handle, maxLength, cache); tmpNioBuf = null; chunk = null; //把buf对象放入对象回收栈 recycle(); } } 再次进入PoolArena类: void free(PoolChunk<T> chunk, ByteBuffer nioBuffer, long handle, int normCapacity, PoolThreadCache cache) { //判断是否是unpooled if (chunk.unpooled) { int size = chunk.chunkSize(); destroyChunk(chunk); activeBytesHuge.add(-size); deallocationsHuge.increment(); } else { //判断是哪种类型,tiny、small、normal SizeClass sizeClass = sizeClass(normCapacity); //放入缓存 if (cache != null && cache.add(this, chunk, nioBuffer, handle, normCapacity, sizeClass)) { // cached so not free it. return; } freeChunk(chunk, handle, sizeClass, nioBuffer); } } //计算内存区域是哪种类型 private SizeClass sizeClass(int normCapacity) { if (!isTinyOrSmall(normCapacity)) { return SizeClass.Normal; } return isTiny(normCapacity) ? SizeClass.Tiny : SizeClass.Small; } 然后到PoolThreadCache类: boolean add(PoolArena<?> area, PoolChunk chunk, ByteBuffer nioBuffer, long handle, int normCapacity, SizeClass sizeClass) { MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass); if (cache == null) { return false; } //加入到缓存队列 return cache.add(chunk, nioBuffer, handle); } private MemoryRegionCache<?> cache(PoolArena<?> area, int normCapacity, SizeClass sizeClass) { //判断是哪种类型,然后把内存回收到哪一块 switch (sizeClass) { case Normal: return cacheForNormal(area, normCapacity); case Small: return cacheForSmall(area, normCapacity); case Tiny: return cacheForTiny(area, normCapacity); default: throw new Error(); } } private MemoryRegionCache<?> cacheForTiny(PoolArena<?> area, int normCapacity) { int idx = PoolArena.tinyIdx(normCapacity); if (area.isDirect()) { return cache(tinySubPageDirectCaches, idx); } return cache(tinySubPageHeapCaches, idx); } 上述跟踪代码步骤图: ByteBuf零拷贝机制 Netty的零拷贝机制,是一种应用层的实现,和底层的JVM、操作系统内存机制没有过多的关联 几种示例 ==一:CompositeByteBuf,将多个ByteBuf合并为一个逻辑上的ByteBuf,避免了各个ByteBuf之间的拷贝== public static void test1() { ByteBuf buf1 = Unpooled.buffer(4); ByteBuf buf2 = Unpooled.buffer(3); byte[] bytes1 = {1,2}; byte[] bytes2 = {3,4,5}; buf1.writeBytes(bytes1); buf2.writeBytes(bytes2); CompositeByteBuf byteBuf = Unpooled.compositeBuffer(); byteBuf = byteBuf.addComponents(true, buf1, buf2); System.out.println("byteBuf: " + byteBuf.toString()); } 上面输出结果,ridx是顺序读的读取位置,widx是顺序写的写入位置,cap是新的ByteBuf的容量,components是指新的ByteBuf是由几个ByteBuf组成==二:wrappedBuffer()方法,将byte[]数组包装成ByteBuf对象== public static void test2() { byte[] bytes = {1,2,3,4,5}; ByteBuf buf = Unpooled.wrappedBuffer(bytes); System.out.println("buf:" + buf.toString()); } 输出结果中:ridx是顺序读的读取位置,widx是顺序写的写入位置,cap是ByteBuf的容量,新的ByteBuf里存的是数组的引用地址,实质操作的还是原来的数组==三:slice()方法,将一个ByteBuf对象切分成多个ByteBuf对象== public static void test3() { ByteBuf buf = Unpooled.wrappedBuffer("hello".getBytes()); ByteBuf byteBuf = buf.slice(1,2); System.out.println("byteBuf:" + byteBuf.toString()); } 输出结果中,可以看到,有两个ByteBuf,其中一个是原有的,新的ByteBuf中存放了原来的ByteBuf的引用地址,另一个是分割后的ByteBuf的引用地址

优秀的个人博客,低调大师

Java函数式编程-3.流(Stream)

List<String> words = Arrays.asList("hello","world"); List<String> chars = words.stream() .flatMap(word -> Stream.of(word.split(""))) .collect(Collectors.toList()); Jdk8中新增的特性旨在帮助程序员写出更好的代码,其中对核心类库的改进是很关键的一部分。对核心类库的改进主要包括集合类的API和新引入的流(Stream)。流使程序员得以站在更高的抽象层次上对集合进行操作。 1.什么是流(Stream)? Stream是特定类型的对象形成的一个队列并支持聚合操作。 Java中的Stream并不会存储元素,而是按需计算。流的来源可以是集合,数组,I/O channel, 产生器generator 等。聚合操作类似SQL语句一样的操作, 比如filter, map, reduce, find, match, sorted等。和以前的Collection操作不同, Stream操作还有两个基础的特征:Pipelining中间操作都会返回流对象本身。 这样多个操作可以串联成一个管道,如同流式风格(fluent style)。 这样做可以对操作进行优化, 比如延迟执行(laziness)和短路( short-circuiting)。 在 Java 8 中, 集合接口有两个方法来生成流: stream() − 为集合创建串行流。 parallelStream() − 为集合创建并行流。 List<String> list = Arrays.asList("a", "b", "c", "d", null); List<String> filtered = list.stream().filter(e -> Objects.nonNull(e)).collect(Collectors.toList()); long count = list.parallelStream().filter(e -> Objects.nonNull(e)).count(); 2.内部迭代 Java程序员在使用集合时,一个通用的模式时在集合上进行迭代,在迭代过程中处理每一个元素并返回处理结果。通常代码时这样的: List<Integer> list = Arrays.asList(1, 2, 3, 4, 5); int sum = 0; Iterator<Integer> iterator = list.iterator(); while (iterator.hasNext()) { sum += iterator.next(); } System.out.println(sum); 这段代码本身没什么问题就写起来有点麻烦,就其背后原理来看其实就是一个封装了迭代的语法糖,首先调用iterator方法产生一个新的Iterator对象进而控制整个迭代过程,这就是外部迭代。 然而外部迭代的问题在于首先,很难抽象出后续提到的复杂操作,另外从本质上来讲只是一种串行化操作。总体来看使用for循环会将行为和方法混为一谈。另一种方法是内部迭代,和调用iterator()作用一样使用stream()方法,该方法不是返回一个控制迭代的Iterator对象,而是返回内部迭代中的相应接口:Stream。示例如下: List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); Integer sum = list.stream().filter(e -> e > 5).reduce(0, Integer::sum); System.out.println(sum); 3.实现机制 通常在Java 中调用一个方法计算机会随即执行操作:比如,System.out.println("hello world!")会在终端上输出一条信息。Stream里的一些方法略有不同,它们虽是普通Java方法但是返回的Stream对象并不是一个新的集合,而是创建新集合的配方。 list.stream().filter(e -> e > 5); 上述代码并未做什么实际性的工作,filter只是刻画出了Stream但是并没有产生新的集合,像filter这样只描述Stream,最终不产生新集合的方法惰性求值方法,而像sum()这样最终从Stream产出值的方法叫做及早求值方法。如果在filter方法中添加一条输出语句就会比较清楚的看出差异。 List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); list.stream().filter(e -> { System.out.println(e); return e > 5; }); 如果将上述语句添加一个及早求值方法就可以看到集合中的元素被打印出来。 List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); list.stream().filter(e -> { System.out.println(e); return e > 5; }).findFirst(); 判断一个操作是惰性求值还是及早求值很简单,只要返回值是Stream那么是惰性求值。使用这些操作的理想方式就是形成一个惰性求值链,最后用一个及早求值的操作返回想要的结果,这正是他的合理之处。那为什么要区分惰性求值和及早求值呢?原因很简单只需要一次迭代就可以得到全部结果,例如 :上述代码只需要找到第一个大于5的数而不需要比较所有集合元素。 4.常用的流操作 为了更好的学习Stream,我们接下来学习一些常用的stream操作。 filter 如果需要在遍历集合元素时清洗非法元素可以使用filter方法,我们在上述讨论过程中已经使用过该方法。filter接收一个函数作为参数,该函数使用lambda表达式表示,如果被清洗元素符合预期则返回true,其实看源码会发现这个lambda表达式是我们之前讲过的 Predicate<T>函数。 collect 这个方法是由Stream生成一个List,是一个及早求值操作。Jdk8中借鉴了很多Scala的设计思想,Stream的of方法可以用来生成一个新的Stream,我们来看下边的一个例子。 List<String> list = Stream.of("a", "b", "c","d","e").collect(Collectors.toList()); map map函数是用来讲一个类型转换为另外一种类型,参数同样是一个lambda表达式对应我们之前讲过的Function<T, R> 函数。因为map是一个惰性求值方法我们配合之前的collect看一个例子。 List<String> list = Stream.of("a", "b", "c").map(e -> e.toUpperCase()).collect(Collectors.toList()); System.out.println(list); flatMap 和map类似,不同的是其每个元素转换得到的是Stream对象,最终会把转换得到的多个Stream连接成一个Stream。 List<String> words = Arrays.asList("hello","world"); List<String> chars = words.stream() .flatMap(word -> Stream.of(word.split(""))) .collect(Collectors.toList()); 上述代码我们可以得到输出结果 :[h, e, l, l, o, w, o, r, l, d],如果将flatMap换成map会发现我们得到了一个List<Stream<String>> ,大家感兴趣可以上手一试。 max和min 和之前讲的函数不同,这两个方法要求参数是一个比较器Comparator<T> Integer min = Stream.of(1,2,3,4,5).min(Integer::compare).get(); System.out.println(min); Integer max = Stream.of(1,2,3,4,5).max(Integer::compare).get(); System.out.println(max); reduce 该函数可以实现从一组值中生成一个值,在上述例子中用到的 sum、max、min本质上都是reduce操作。Stream的求和结果每一步都将Stream中的元素累加至accumulator,遍历至Stream中的最后一个元素时,accumulator的值就是所有元素的和。我们看如下例子 Integer sum = Stream.of(1, 2, 3, 4, 5).reduce(0, (acc, e) -> acc + e); System.out.println(sum); 这段代码如果你对scala熟悉那么会很容易理解,acc作为累加器存储了中间计算结果,整个lambda函数本质上是我们之前讲过的BinaryOperator。

优秀的个人博客,低调大师

Java 并发编程:AQS 的公平性

所谓公平是指所有线程对临界资源申请访问权限的成功率都一样,它不会让某些线程拥有优先权。通过几篇文章的分析我们知道了JDK的AQS的锁是基于CLH锁进行优化的,而其中使用了FIFO队列,也就是说等待队列是一个先进先出的队列。那是否就可以说每条线程获取锁时就是公平的呢?关于公平性,严格来说应该分成三个点来看:入队阶段、唤醒阶段以及闯入策略。 友情链接: 什么是JDK内置并发框架AQS AQS的原子性如何保证 AQS对CLH锁的优化 入队阶段  唤醒阶段 当线程节点成功加入等待队列后便成为等待队列中的节点,而且这是一个先入先出队列,那么我们可以得到一个结论:队列中的所有节点是公平的。因为等待队列中的所有节点都按照顺序等待自己被前驱节点唤醒并获取锁,所以等待队列中的节点具有公平性。  闯入策略 闯入策略是AQS框架为了提升性能而设计的一个策略,具体是指一个新线程到达共享资源边界时不管等待队列中是否存在其它等待节点,新线程都将优先尝试去获取锁,这看起来就像是闯入行为。闯入策略破坏了公平性,AQS框架对外体现的公平性主要也由此体现。 AQS提供的锁获取操作运用了可闯入算法,即如果有新线程到来先进行一次获取尝试,不成功的情况下才将当前线程加入等待队列。如下图,等待队列中节点线程按照顺序一个接一个尝试去获取共享资源的使用权。而某一时刻头结点线程准备尝试获取的同时另外一条线程闯入,新线程并非直接加入等待队列的尾部,而是先跟头结点线程竞争获取资源。闯入线程如果成功获取共享资源则直接执行,头结点线程则继续等待下一次尝试。如此一来闯入线程成功插队,后来的线程比早到的线程先执行,说明AQS锁获取算法是不严格公平的。 闯入逻辑 下图是包含了闯入策略的锁获取算法伪代码,我们主要关注红色方框的逻辑。它会优先直接去尝试获取锁,如果获取失败(即闯入失败)才创建节点并加入到等待队列的尾部。 为什么需要闯入策略 为什么要使用闯入策略呢?闯入策略通常可以提升总吞吐量。由于一般同步器颗粒度比较小,也可以说共享资源的范围较小,而线程从阻塞状态到被唤醒所消耗的时间周期可能是通过共享资源时间周期的几倍甚至几十倍。 如此一来线程唤醒过程中将存在一个很大的时间周期空窗期,导致资源没有得到充分利用,同时如果每个线程都先入队再唤醒的话也会导致效率低下。为了避免没必要的线程挂起和唤醒,也为了提高吞吐量,于是引入这种闯入策略。它可以充分利用阻塞唤醒空窗期,也避免了无谓的挂起和唤醒操作,从而大大增加了吞吐率。 闯入机制的实现对外提供一种竞争调节机制,开发者可以在自定义同步器中定义闯入尝试获取的次数。假设次数为n则不断重复获取直到n次都获取不成功才把线程加入等待队列中,随着次数n的增加可以增大成功闯入的几率。同时,这种闯入策略可能导致等待队列中的线程饥饿,因为锁可能一直被闯入的线程获取。但由于一般持有同步器的时间很短暂所以能避免饥饿的发生,反之如果持有锁的时间较长,则将大大增加等待队列无限等待的风险。 总结 实际情况中我们要根据需求制定策略,在一个公平性要求很高的场景,则可以把闯入策略去除掉以达到公平。在自定义同步器中可以通过AQS预留方法tryAcquire方法实现,只需判断当前线程是否为等待队列中头结点对应的线程即可。若不是则直接返回false,尝试获取失败。但这种公平性是相对于Java语义层面上的公平性,在现实中JVM的实现可能也会直接影响线程执行的顺序。

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称,一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集,帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册