首页 文章 精选 留言 我的

精选列表

搜索[并发原语],共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); } }

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

微服务架构的高并发问题

1 准备环境 1.1 准备商品微服务和订单微服务 其中商品微服务的findById()方法设置休眠2秒,用来模拟网络波动等情况: package com.sunxiaping.product.controller; import com.sunxiaping.product.domain.Product; import com.sunxiaping.product.service.ProductService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.web.bind.annotation.*; @RestController @RequestMapping(value = "/product") public class ProductController { @Autowired private ProductService productService; @Value("${server.port}") private String port; @Value("${spring.cloud.client.ip-address}") private String ip; @PostMapping(value = "/save") public String save(@RequestBody Product product) { productService.save(product); return "新增成功"; } @GetMapping(value = "/findById/{id}") public Product findById(@PathVariable(value = "id") Long id) { try { //休眠2秒,用来模拟 网络波动等情况 Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } Product product = productService.findById(id); product.setProductName("访问的地址是:" + ip + ":" + port); return product; } } 设置订单微服务的Tomcat的最大线程数是10: server: port: 9002 # 微服务的端口号 tomcat: max-threads: 10 # 最大线程数是10 spring: application: name: service-order # 微服务的名称 datasource: url: jdbc:mysql://192.168.1.57:3306/test?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true driver-class-name: com.mysql.cj.jdbc.Driver username: root password: 123456 jpa: generate-ddl: true show-sql: true open-in-view: true database: mysql jmx: unique-names: true # 配置Eureka eureka: instance: # 实例的名称 instance-id: service-order:9002 # 显示IP信息 prefer-ip-address: true lease-renewal-interval-in-seconds: 5 # 发送心跳续约间隔(默认30秒) lease-expiration-duration-in-seconds: 10 # Eureka Client发送心跳给Eureka Server端后,续约到期时间(默认90秒) client: healthcheck: enabled: true service-url: # Eureka Server的地址 # defaultZone: http://localhost:9000/eureka/ defaultZone: http://eureka7001.com:7001/eureka/,http://eureka7002.com:7002/eureka/ # Ribbon的重试机制 service-product: ribbon: # 修改ribbon的负载均衡策略 服务名 - ribbon - NFLoadBalancerRuleClassName :负载均衡策略 # NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule # 修改ribbon的负载均衡策略为权重策略 # Ribbon的重试机制参数 ConnectTimeout: 250 # Ribbon的连接超时时间 ReadTimeout: 1000 # Ribbon的数据读取超时时间 OkToRetryOnAllOperations: true # 是否对所有操作都进行重试 MaxAutoRetriesNextServer: 50 # 切换实例的重试次数 MaxAutoRetries: 1 # 对当前实例的重试次数 # 微服务info内容详细信息 info: app.name: xxx company.name: xxx build.artifactId: $project.artifactId$ build.version: $project.version$ # 开启日志debug logging: level: root: info 订单微服务中的SpringConfig.java package com.sunxiaping.order.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.client.RestTemplate; @Configuration public class SpringConfig { @Bean // @LoadBalanced public RestTemplate restTemplate() { return new RestTemplate(); } } 订单微服务的OrderController.java package com.sunxiaping.order.controller; import com.sunxiaping.order.domain.Product; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.client.RestTemplate; @RestController @RequestMapping(value = "/order") public class OrderController { @Autowired private RestTemplate restTemplate; /** * @param id * @return */ @GetMapping(value = "/buy/{id}") public Product buy(@PathVariable(value = "id") Long id) { Product product = restTemplate.getForObject("http://localhost:9001/product/findById/" + id, Product.class); return product; } @GetMapping(value = "/findOrder") public String findOrder() { return "商品查询到了"; } } 2 使用Jmeter测试接口 使用JMeter性能测试工具以50个线程每个线程循环50次测试:http://localhost:9002/order/buy/1接口,然后通过浏览器调用http://localhost:9002/order/findOrder接口,发现特别慢。 3 系统负载过高存在的问题 3.1 问题分析 在微服务架构中,我们将业务拆成一个个的服务,服务和服务之间可以相互调用,由于网络原因或者自身的原因,服务并不能保证100%可用,如果单个服务出现问题,调用这个服务就会出现网络延迟,此时如果有大量的网络请求涌入,会形成任务累计,导致服务瘫痪。 换句话说,Tomcat等容器会以线程池的方式对所有的请求进行统一的管理,如果某个方法可能存着耗时问题,随着外面积压的请求越来越多,势必会造成系统的崩溃、瘫痪等。 为了不影响其他接口的正常访问:对多个服务之间进行隔离。 服务隔离的方式: :one:线程池隔离。 :two:信号量隔离(计数器,就是对某个方法进行设置阈值,如果超过了阈值,直接报错)。 4 线程池隔离的方式处理积压问题 4.1 在订单微服务中引入相关jar包的Maven坐标 <dependency> <groupId>com.netflix.hystrix</groupId> <artifactId>hystrix-metrics-event-stream</artifactId> <version>1.5.12</version> </dependency> <dependency> <groupId>com.netflix.hystrix</groupId> <artifactId>hystrix-javanica</artifactId> <version>1.5.12</version> </dependency> 4.2 配置线程池 配置HystrixCommand接口的实现类,在实现类中可以对线程池进行配置: package com.sunxiaping.order.command; import com.netflix.hystrix.*; import com.sunxiaping.order.domain.Product; import org.springframework.web.client.RestTemplate; public class OrderCommand extends HystrixCommand<Product> { private RestTemplate restTemplate; private Long id; public OrderCommand(RestTemplate restTemplate, Long id) { super(setter()); this.restTemplate = restTemplate; this.id = id; } private static Setter setter() { // 服务分组 HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("order_product"); // 服务标识 HystrixCommandKey commandKey = HystrixCommandKey.Factory.asKey("product"); // 线程池名称 HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("order_product_pool"); /** * 线程池配置 * withCoreSize : 线程池大小为10 * withKeepAliveTimeMinutes: 线程存活时间15秒 * withQueueSizeRejectionThreshold :队列等待的阈值为100,超过100执行拒绝策略 */ HystrixThreadPoolProperties.Setter threadPoolProperties = HystrixThreadPoolProperties.Setter().withCoreSize(50) .withKeepAliveTimeMinutes(15).withQueueSizeRejectionThreshold(100); // 命令属性配置Hystrix 开启超时 HystrixCommandProperties.Setter commandProperties = HystrixCommandProperties.Setter() // 采用线程池方式实现服务隔离 .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.THREAD) // 禁止 .withExecutionTimeoutEnabled(false); return Setter.withGroupKey(groupKey).andCommandKey(commandKey).andThreadPoolKey(threadPoolKey) .andThreadPoolPropertiesDefaults(threadPoolProperties).andCommandPropertiesDefaults(commandProperties); } @Override protected Product run() throws Exception { System.out.println(Thread.currentThread().getName()); return restTemplate.getForObject("http://localhost:9001/product/findById/" + id, Product.class); } /** * 服务降级 * * @return */ @Override protected Product getFallback() { Product product = new Product(); product.setProductName("不好意思,出错了"); return product; } } 4.3 修改Controller package com.sunxiaping.order.controller; import com.sunxiaping.order.command.OrderCommand; import com.sunxiaping.order.domain.Product; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.client.RestTemplate; @RestController @RequestMapping(value = "/order") public class OrderController { @Autowired private RestTemplate restTemplate; /** * 使用OrderCommand调用远程远程服务 * * @param id * @return */ @GetMapping(value = "/buy/{id}") public Product buy(@PathVariable(value = "id") Long id) { return new OrderCommand(restTemplate, id).execute(); } @GetMapping(value = "/findOrder") public String findOrder() { System.out.println(Thread.currentThread().getName()); return "商品查询到了"; } }

资源下载

更多资源
Mario

Mario

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

腾讯云软件源

腾讯云软件源

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

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

WebStorm

WebStorm

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

用户登录
用户注册