首页 文章 精选 留言 我的

精选列表

搜索[socket编程],共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); } }

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

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的实现可能也会直接影响线程执行的顺序。

资源下载

更多资源
Nacos

Nacos

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

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

WebStorm

WebStorm

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

用户登录
用户注册