java.util.concurrent包 （备忘）

2024-05-16 03:18
文章标签 java util 备忘 concurrent

本文主要是介绍java.util.concurrent包 （备忘），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

newFixedThreadPool

创建一个固定大小的线程池。

shutdown()：用于关闭启动线程，如果不调用该语句，jvm不会关闭。

awaitTermination()：用于等待子线程结束，再继续执行下面的代码。该例中我设置一直等着子线程结束。

Java代码   收藏代码
  1. public class Test {  
  2.   
  3.     public static void main(String[] args) throws IOException, InterruptedException {  
  4.         ExecutorService service = Executors.newFixedThreadPool(2);  
  5.         for (int i = 0; i < 4; i++) {  
  6.             Runnable run = new Runnable() {  
  7.                 @Override  
  8.                 public void run() {  
  9.                     System.out.println("thread start");  
  10.                 }  
  11.             };  
  12.             service.execute(run);  
  13.         }  
  14.         service.shutdown();  
  15.         service.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);  
  16.         System.out.println("all thread complete");  
  17.     }  
  18. }  

 

输出：
thread start
thread start
thread start
thread start
all thread complete

newScheduledThreadPool

这个先不说，我喜欢用spring quartz.

CyclicBarrier

假设有只有的一个场景：每个线程代表一个跑步运动员，当运动员都准备好后，才一起出发，只要有一个人没有准备好，大家都等待.

 

Java代码   收藏代码
  1. import java.io.IOException;  
  2. import java.util.Random;  
  3. import java.util.concurrent.BrokenBarrierException;  
  4. import java.util.concurrent.CyclicBarrier;  
  5. import java.util.concurrent.ExecutorService;  
  6. import java.util.concurrent.Executors;  
  7.   
  8. class Runner implements Runnable {  
  9.   
  10.     private CyclicBarrier barrier;  
  11.   
  12.     private String name;  
  13.   
  14.     public Runner(CyclicBarrier barrier, String name) {  
  15.         super();  
  16.         this.barrier = barrier;  
  17.         this.name = name;  
  18.     }  
  19.   
  20.     @Override  
  21.     public void run() {  
  22.         try {  
  23.             Thread.sleep(1000 * (new Random()).nextInt(8));  
  24.             System.out.println(name + " 准备OK.");  
  25.             barrier.await();  
  26.         } catch (InterruptedException e) {  
  27.             e.printStackTrace();  
  28.         } catch (BrokenBarrierException e) {  
  29.             e.printStackTrace();  
  30.         }  
  31.         System.out.println(name + " Go!!");  
  32.     }  
  33. }  
  34.   
  35. public class Race {  
  36.   
  37.     public static void main(String[] args) throws IOException, InterruptedException {  
  38.         CyclicBarrier barrier = new CyclicBarrier(3);  
  39.   
  40.         ExecutorService executor = Executors.newFixedThreadPool(3);  
  41.         executor.submit(new Thread(new Runner(barrier, "zhangsan")));  
  42.         executor.submit(new Thread(new Runner(barrier, "lisi")));  
  43.         executor.submit(new Thread(new Runner(barrier, "wangwu")));  
  44.   
  45.         executor.shutdown();  
  46.     }  
  47.   
  48. }  

 

输出：
wangwu 准备OK.
zhangsan 准备OK.
lisi 准备OK.
lisi Go!!
zhangsan Go!!
wangwu Go!!

ThreadPoolExecutor

 

newFixedThreadPool生成一个固定的线程池，顾名思义，线程池的线程是不会释放的，即使它是Idle。这就会产生性能问题，比如如 果线程池的大小为200，当全部使用完毕后，所有的线程会继续留在池中，相应的内存和线程切换（while(true)+sleep循环）都会增加。如果 要避免这个问题，就必须直接使用ThreadPoolExecutor()来构造。可以像Tomcat的线程池一样设置“最大线程数”、“最小线程数”和 “空闲线程keepAlive的时间”。

 

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler)  


corePoolSize：池中所保存的线程数，包括空闲线程（非最大同时干活的线程数）。如果池中线程数多于 corePoolSize，则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止。

maximumPoolSize：线程池中最大线程数

keepAliveTime：线程空闲回收的时间

unit：keepAliveTime的单位

workQueue：保存任务的队列，可以如下选择：

 

  •   无界队列： new LinkedBlockingQueue<Runnable>();
  •   有界队列： new ArrayBlockingQueue<Runnable>(8);你不想让客户端无限的请求吃光你的CPU和内存吧，那就用有界队列

handler：当提交任务数大于队列size会抛出RejectedExecutionException，可选的值为：

 

  • ThreadPoolExecutor.CallerRunsPolicy 等待队列空闲
  • ThreadPoolExecutor.DiscardPolicy：丢弃要插入队列的任务
  • ThreadPoolExecutor.DiscardOldestPolicy：删除队头的任务

关于corePoolSize和maximumPoolSize：

 

Java官方Docs写道：
当新任务在方法 execute(java.lang.Runnable) 中提交时，如果运行的线程少于 corePoolSize，则创建新线程来处理请求（即使存在空闲线程）。如果运行的线程多于 corePoolSize 而少于 maximumPoolSize，则仅当队列（queue）满时才创建新线程。如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务。

 

Java代码   收藏代码
  1. public class Test {  
  2.   
  3.     public static void main(String[] args) {  
  4.         BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();  
  5.         ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 6, 1, TimeUnit.DAYS, queue);  
  6.   
  7.         for (int i = 0; i < 20; i++) {  
  8.             final int index = i;  
  9.             executor.execute(new Runnable() {  
  10.                 public void run() {  
  11.                     try {  
  12.                         Thread.sleep(4000);  
  13.                     } catch (InterruptedException e) {  
  14.                         e.printStackTrace();  
  15.                     }  
  16.                     System.out.println(String.format("thread %d finished", index));  
  17.                 }  
  18.             });  
  19.         }  
  20.         executor.shutdown();  
  21.     }  
  22. }  

 

原子变量（Atomic )

并发库中的BlockingQueue是一个比较好玩的类，顾名思义，就是阻塞队列。该类主要提供了两个方法put()和take()，前者将一个 对象放到队列中，如果队列已经满了，就等待直到有空闲节点；后者从head取一个对象，如果没有对象，就等待直到有可取的对象。

 

下面的例子比较简单，一个读线程，用于将要处理的文件对象添加到阻塞队列中，另外四个写线程用于取出文件对象，为了模拟写操作耗时长的特点，特让线 程睡眠一段随机长度的时间。另外，该Demo也使用到了线程池和原子整型（AtomicInteger），AtomicInteger可以在并发情况下达 到原子化更新，避免使用了synchronized，而且性能非常高。由于阻塞队列的put和take操作会阻塞，为了使线程退出，在队列中添加了一个 “标识”，算法中也叫“哨兵”，当发现这个哨兵后，写线程就退出。

 

Java代码   收藏代码
  1. import java.io.File;  
  2. import java.io.FileFilter;  
  3. import java.util.concurrent.BlockingQueue;  
  4. import java.util.concurrent.ExecutorService;  
  5. import java.util.concurrent.Executors;  
  6. import java.util.concurrent.LinkedBlockingQueue;  
  7. import java.util.concurrent.atomic.AtomicInteger;  
  8.   
  9. public class Test {  
  10.   
  11.     static long randomTime() {  
  12.         return (long) (Math.random() * 1000);  
  13.     }  
  14.   
  15.     public static void main(String[] args) {  
  16.         // 能容纳100个文件  
  17.         final BlockingQueue<File> queue = new LinkedBlockingQueue<File>(100);  
  18.         // 线程池  
  19.         final ExecutorService exec = Executors.newFixedThreadPool(5);  
  20.         final File root = new File("D:\\dist\\blank");  
  21.         // 完成标志  
  22.         final File exitFile = new File("");  
  23.         // 读个数  
  24.         final AtomicInteger rc = new AtomicInteger();  
  25.         // 写个数  
  26.         final AtomicInteger wc = new AtomicInteger();  
  27.         // 读线程  
  28.         Runnable read = new Runnable() {  
  29.             public void run() {  
  30.                 scanFile(root);  
  31.                 scanFile(exitFile);  
  32.             }  
  33.   
  34.             public void scanFile(File file) {  
  35.                 if (file.isDirectory()) {  
  36.                     File[] files = file.listFiles(new FileFilter() {  
  37.                         public boolean accept(File pathname) {  
  38.                             return pathname.isDirectory() || pathname.getPath().endsWith(".log");  
  39.                         }  
  40.                     });  
  41.                     for (File one : files)  
  42.                         scanFile(one);  
  43.                 } else {  
  44.                     try {  
  45.                         int index = rc.incrementAndGet();  
  46.                         System.out.println("Read0: " + index + " " + file.getPath());  
  47.                         queue.put(file);  
  48.                     } catch (InterruptedException e) {  
  49.                     }  
  50.                 }  
  51.             }  
  52.         };  
  53.         exec.submit(read);  
  54.         // 四个写线程  
  55.         for (int index = 0; index < 4; index++) {  
  56.             // write thread  
  57.             final int num = index;  
  58.             Runnable write = new Runnable() {  
  59.                 String threadName = "Write" + num;  
  60.   
  61.                 public void run() {  
  62.                     while (true) {  
  63.                         try {  
  64.                             Thread.sleep(randomTime());  
  65.                             int index = wc.incrementAndGet();  
  66.                             File file = queue.take();  
  67.                             // 队列已经无对象  
  68.                             if (file == exitFile) {  
  69.                                 // 再次添加"标志"，以让其他线程正常退出  
  70.                                 queue.put(exitFile);  
  71.                                 break;  
  72.                             }  
  73.                             System.out.println(threadName + ": " + index + " " + file.getPath());  
  74.                         } catch (InterruptedException e) {  
  75.                         }  
  76.                     }  
  77.                 }  
  78.   
  79.             };  
  80.             exec.submit(write);  
  81.         }  
  82.         exec.shutdown();  
  83.     }  
  84.   
  85. }  
 

CountDownLatch

 

从名字可以看出，CountDownLatch是一个倒数计数的锁，当倒数到0时触发事件，也就是开锁，其他人就可以进入了。在一些应用场合中，需要等待某个条件达到要求后才能做后面的事情；同时当线程都完成后也会触发事件，以便进行后面的操作。 

CountDownLatch最重要的方法是countDown()和await()，前者主要是倒数一次，后者是等待倒数到0，如果没有到达0，就只有阻塞等待了。

一个CountDouwnLatch实例是不能重复使用的，也就是说它是一次性的，锁一经被打开就不能再关闭使用了，如果想重复使用，请考虑使用CyclicBarrier。

下面的例子简单的说明了CountDownLatch的使用方法，模拟了100米赛跑，10名选手已经准备就绪，只等裁判一声令下。当所有人都到达终点时，比赛结束。

 

Java代码   收藏代码
  1. import java.util.concurrent.CountDownLatch;  
  2. import java.util.concurrent.ExecutorService;  
  3. import java.util.concurrent.Executors;  
  4.   
  5. public class Test {  
  6.   
  7.     public static void main(String[] args) throws InterruptedException {  
  8.   
  9.         // 开始的倒数锁  
  10.         final CountDownLatch begin = new CountDownLatch(1);  
  11.   
  12.         // 结束的倒数锁  
  13.         final CountDownLatch end = new CountDownLatch(10);  
  14.   
  15.         // 十名选手  
  16.         final ExecutorService exec = Executors.newFixedThreadPool(10);  
  17.   
  18.         for (int index = 0; index < 10; index++) {  
  19.             final int NO = index + 1;  
  20.             Runnable run = new Runnable() {  
  21.                 public void run() {  
  22.                     try {  
  23.                         begin.await();  
  24.                         Thread.sleep((long) (Math.random() * 10000));  
  25.                         System.out.println("No." + NO + " arrived");  
  26.                     } catch (InterruptedException e) {  
  27.                     } finally {  
  28.                         end.countDown();  
  29.                     }  
  30.                 }  
  31.             };  
  32.             exec.submit(run);  
  33.         }  
  34.         System.out.println("Game Start");  
  35.         begin.countDown();  
  36.         end.await();  
  37.         System.out.println("Game Over");  
  38.         exec.shutdown();  
  39.     }  
  40.   
  41. }  

使用Callable和Future实现线程等待和多线程返回值

假设在main线程启动一个线程，然后main线程需要等待子线程结束后，再继续下面的操作，我们会通过join方法阻塞main线程，代码如下：
Java代码   收藏代码
  1. Runnable runnable = ...;  
  2. Thread t = new Thread(runnable);  
  3. t.start();  
  4. t.join();  
  5. ......  
通过JDK1.5线程池管理的线程可以使用Callable和Future实现（join()方法无法应用到在线程池线程）
Java代码   收藏代码
  1. import java.util.concurrent.Callable;  
  2. import java.util.concurrent.ExecutionException;  
  3. import java.util.concurrent.ExecutorService;  
  4. import java.util.concurrent.Executors;  
  5. import java.util.concurrent.Future;  
  6.   
  7. public class Test {  
  8.   
  9.     public static void main(String[] args) throws InterruptedException, ExecutionException {  
  10.         System.out.println("start main thread");  
  11.         final ExecutorService exec = Executors.newFixedThreadPool(5);  
  12.           
  13.         Callable<String> call = new Callable<String>() {  
  14.             public String call() throws Exception {  
  15.                 System.out.println("  start new thread.");  
  16.                 Thread.sleep(1000 * 5);  
  17.                 System.out.println("  end new thread.");  
  18.                 return "some value.";  
  19.             }  
  20.         };  
  21.         Future<String> task = exec.submit(call);  
  22.         Thread.sleep(1000 * 2);  
  23.         task.get(); // 阻塞，并待子线程结束，  
  24.         exec.shutdown();  
  25.         exec.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);  
  26.         System.out.println("end main thread");  
  27.     }  
  28.   
  29. }  
 
Java代码   收藏代码
  1. import java.util.ArrayList;  
  2. import java.util.List;  
  3. import java.util.concurrent.Callable;  
  4. import java.util.concurrent.ExecutionException;  
  5. import java.util.concurrent.ExecutorService;  
  6. import java.util.concurrent.Executors;  
  7. import java.util.concurrent.Future;  
  8.   
  9. /** 
  10. * 多线程返回值测试 
  11. */  
  12. public class ThreadTest {  
  13.   
  14.     public static void main(String[] args) throws InterruptedException, ExecutionException {  
  15.         System.out.println("start main thread");  
  16.         int threadCount = 5;  
  17.         final ExecutorService exec = Executors.newFixedThreadPool(threadCount);  
  18.   
  19.         List<Future<Integer>> tasks = new ArrayList<Future<Integer>>();  
  20.         for (int i = 0; i < threadCount; i++) {  
  21.             Callable<Integer> call = new Callable<Integer>() {  
  22.                 public Integer call() throws Exception {  
  23.                     Thread.sleep(1000);  
  24.                     return 1;  
  25.                 }  
  26.             };  
  27.             tasks.add(exec.submit(call));  
  28.         }  
  29.         long total = 0;  
  30.         for (Future<Integer> future : tasks) {  
  31.             total += future.get();  
  32.         }  
  33.         exec.shutdown();  
  34.         System.out.println("total: " + total);  
  35.         System.out.println("end main thread");  
  36.     }  
  37. }  
 

CompletionService

这个东西的使用上很类似上面的example，不同的是，它会首先取完成任务的线程。下面的参考文章里，专门提到这个，大家有兴趣可以看下，例子：

 

 

Java代码   收藏代码
  1. import java.util.concurrent.Callable;  
  2. import java.util.concurrent.CompletionService;  
  3. import java.util.concurrent.ExecutionException;  
  4. import java.util.concurrent.ExecutorCompletionService;  
  5. import java.util.concurrent.ExecutorService;  
  6. import java.util.concurrent.Executors;  
  7. import java.util.concurrent.Future;  
  8.   
  9. public class Test {  
  10.     public static void main(String[] args) throws InterruptedException,  
  11.     ExecutionException {  
  12.         ExecutorService exec = Executors.newFixedThreadPool(10);  
  13.         CompletionService<String> serv =  
  14.         new ExecutorCompletionService<String>(exec);  
  15.         for (int index = 0; index < 5; index++) {  
  16.             final int NO = index;  
  17.             Callable<String> downImg = new Callable<String>() {  
  18.                 public String call() throws Exception {  
  19.                     Thread.sleep((long) (Math.random() * 10000));  
  20.                     return "Downloaded Image " + NO;  
  21.                 }  
  22.             };  
  23.             serv.submit(downImg);  
  24.         }  
  25.         Thread.sleep(1000 * 2);  
  26.         System.out.println("Show web content");  
  27.         for (int index = 0; index < 5; index++) {  
  28.             Future<String> task = serv.take();  
  29.             String img = task.get();  
  30.             System.out.println(img);  
  31.         }  
  32.         System.out.println("End");  
  33.         // 关闭线程池  
  34.         exec.shutdown();  
  35.     }  
  36. }  

 

 

Semaphore信号量

 

拿到信号量的线程可以进入代码，否则就等待。通过acquire()和release()获取和释放访问许可。下面的例子只允许5个线程同时进入执行acquire()和release()之间的代码

 

 

Java代码   收藏代码
  1. import java.util.concurrent.ExecutorService;  
  2. import java.util.concurrent.Executors;  
  3. import java.util.concurrent.Semaphore;  
  4.   
  5. public class Test {  
  6.   
  7.     public static void main(String[] args) {  
  8.         // 线程池  
  9.         ExecutorService exec = Executors.newCachedThreadPool();  
  10.         // 只能5个线程同时访问  
  11.         final Semaphore semp = new Semaphore(5);  
  12.         // 模拟20个客户端访问  
  13.         for (int index = 0; index < 20; index++) {  
  14.             final int NO = index;  
  15.             Runnable run = new Runnable() {  
  16.                 public void run() {  
  17.                     try {  
  18.                         // 获取许可  
  19.                         semp.acquire();  
  20.                         System.out.println("Accessing: " + NO);  
  21.                         Thread.sleep((long) (Math.random() * 10000));  
  22.                         // 访问完后，释放  
  23.                         semp.release();  
  24.                     } catch (InterruptedException e) {  
  25.                     }  
  26.                 }  
  27.             };  
  28.             exec.execute(run);  
  29.         }  
  30.         // 退出线程池  
  31.         exec.shutdown();  
  32.     }  
  33.   
  34. }  


参考：

jdk1.5中的线程池使用简介

http://www.java3z.com/cwbwebhome/article/article2/2875.html

CAS原理

http://www.blogjava.net/syniii/archive/2010/11/18/338387.html?opt=admin

jdk1.5中java.util.concurrent包编写多线程

http://hi.baidu.com/luotoo/blog/item/b895c3c2d650591e0ef47731.html

ExecutorSerive vs CompletionService

http://www.coderanch.com/t/491704/threads/java/ExecutorSerive-vs-CompletionService

 

这篇关于java.util.concurrent包 （备忘）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/993720。

相关文章

Java实现Excel与HTML互转

Java实现Excel与HTML互转

《Java实现Excel与HTML互转》Excel是一种电子表格格式,而HTM则是一种用于创建网页的标记语言,虽然两者在用途上存在差异,但有时我们需要将数据从一种格式转换为另一种格式,下面我们就来看看... Excel是一种电子表格格式，广泛用于数据处理和分析，而HTM则是一种用于创建网页的标记语言。虽然两
阅读更多...
java图像识别工具类(ImageRecognitionUtils)使用实例详解

java图像识别工具类(ImageRecognitionUtils)使用实例详解

《java图像识别工具类(ImageRecognitionUtils)使用实例详解》：本文主要介绍如何在Java中使用OpenCV进行图像识别,包括图像加载、预处理、分类、人脸检测和特征提取等步骤... 目录前言1. 图像识别的背景与作用2. 设计目标3. 项目依赖4. 设计与实现 ImageRecogni
阅读更多...
Java中Springboot集成Kafka实现消息发送和接收功能

Java中Springboot集成Kafka实现消息发送和接收功能

《Java中Springboot集成Kafka实现消息发送和接收功能》Kafka是一个高吞吐量的分布式发布-订阅消息系统,主要用于处理大规模数据流,它由生产者、消费者、主题、分区和代理等组件构成,Ka... 目录一、Kafka 简介二、Kafka 功能三、POM依赖四、配置文件五、生产者六、消费者一、Kaf
阅读更多...
Java访问修饰符public、private、protected及默认访问权限详解

Java访问修饰符public、private、protected及默认访问权限详解

《Java访问修饰符public、private、protected及默认访问权限详解》：本文主要介绍Java访问修饰符public、private、protected及默认访问权限的相关资料,每... 目录前言1. public 访问修饰符特点：示例：适用场景：2. private 访问修饰符特点：示例：
阅读更多...
详解Java如何向http/https接口发出请求

详解Java如何向http/https接口发出请求

《详解Java如何向http/https接口发出请求》这篇文章主要为大家详细介绍了Java如何实现向http/https接口发出请求,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 用Java发送web请求所用到的包都在java.net下，在具体使用时可以用如下代码，你可以把它封装成一
阅读更多...
SpringBoot使用Apache Tika检测敏感信息

SpringBoot使用Apache Tika检测敏感信息

《SpringBoot使用ApacheTika检测敏感信息》ApacheTika是一个功能强大的内容分析工具,它能够从多种文件格式中提取文本、元数据以及其他结构化信息,下面我们来看看如何使用Ap... 目录Tika 主要特性1. 多格式支持2. 自动文件类型检测3. 文本和元数据提取4. 支持 OCR（光学
阅读更多...
Java内存泄漏问题的排查、优化与最佳实践

Java内存泄漏问题的排查、优化与最佳实践

《Java内存泄漏问题的排查、优化与最佳实践》在Java开发中,内存泄漏是一个常见且令人头疼的问题,内存泄漏指的是程序在运行过程中,已经不再使用的对象没有被及时释放,从而导致内存占用不断增加,最终... 目录引言1. 什么是内存泄漏？常见的内存泄漏情况2. 如何排查 Java 中的内存泄漏？2.1 使用 J
阅读更多...
JAVA系统中Spring Boot应用程序的配置文件application.yml使用详解

JAVA系统中Spring Boot应用程序的配置文件application.yml使用详解

《JAVA系统中SpringBoot应用程序的配置文件application.yml使用详解》：本文主要介绍JAVA系统中SpringBoot应用程序的配置文件application.yml的... 目录文件路径文件内容解释1. Server 配置2. Spring 配置3. Logging 配置4. Ma
阅读更多...
Java 字符数组转字符串的常用方法

Java 字符数组转字符串的常用方法

《Java字符数组转字符串的常用方法》文章总结了在Java中将字符数组转换为字符串的几种常用方法,包括使用String构造函数、String.valueOf()方法、StringBuilder以及A... 目录1. 使用String构造函数1.1 基本转换方法1.2 注意事项2. 使用String.valu
阅读更多...
java脚本使用不同版本jdk的说明介绍

java脚本使用不同版本jdk的说明介绍

《java脚本使用不同版本jdk的说明介绍》本文介绍了在Java中执行JavaScript脚本的几种方式,包括使用ScriptEngine、Nashorn和GraalVM,ScriptEngine适用... 目录Java脚本使用不同版本jdk的说明1.使用ScriptEngine执行javascript2.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2