java多线程学习--java.util.concurrent

2024-09-04 12:32
文章标签 java 学习 多线程 util concurrent

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

题记:util和concurrent 包是后续重点先看的和学习的模块

原文地址:http://www.cnblogs.com/sunhan/p/3817806.html

 

CountDownLatch，api 文档：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CountDownLatch.html

A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

假设我们要打印1-100，最后再输出“Ok“。1-100的打印顺序不要求统一，只需保证“Ok“是在最后出现即可。
解决方案：我们定义一个CountDownLatch，然后开10个线程分别打印（n-1）*10+1至（n-1）*10+10。主线程中调用await 方法等待所有线程的执行完毕，每个线程执行完毕后都调用countDown方法。最后再await返回后打印“Ok”。

复制代码
复制代码 
package thread;import java.util.concurrent.CountDownLatch;public class TestCountDownLatch {private static final int N = 10;public static void main(String[] args) throws InterruptedException {CountDownLatch doneSignal = new CountDownLatch(N);CountDownLatch startSignal = new CountDownLatch(1);// 开始执行信号for (int i = 1; i <= N; i++) {new Thread(new Worker(i, doneSignal, startSignal)).start();// 线程启动了}System.out.println("begin------------");startSignal.countDown();// 开始执行啦doneSignal.await();// 等待所有的线程执行完毕System.out.println("Ok");}static class Worker implements Runnable {private final CountDownLatch doneSignal;private final CountDownLatch startSignal;private int beginIndex;Worker(int beginIndex, CountDownLatch doneSignal,CountDownLatch startSignal) {this.startSignal = startSignal;this.beginIndex = beginIndex;this.doneSignal = doneSignal;}public void run() {try {startSignal.await(); // 等待开始执行信号的发布beginIndex = (beginIndex - 1) * 10 + 1;for (int i = beginIndex; i < beginIndex + 10; i++) {System.out.println(i);}} catch (InterruptedException e) {e.printStackTrace();} finally {doneSignal.countDown();}}}
}
复制代码
复制代码

 

CyclicBarrier，api 文档：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CyclicBarrier.html

A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. The barrier is called cyclic because it can be re-used after the waiting threads are released.

举一个很简单的例子，今天晚上我们哥们4个去Happy。就互相通知了一下：晚上八点准时到xx酒吧门前集合，不见不散！。有个哥们住的近，早早就到了。有的事务繁忙，刚好踩点到了。无论怎样，先来的都不能独自行动，只能等待所有人

复制代码
复制代码 
public class TestCyclicBarrier {public static void main(String[] args) {//new 一个线程池ExecutorService exec = Executors.newCachedThreadPool();final Random random = new Random();final CyclicBarrier barrier = new CyclicBarrier(4, new Runnable() {@Overridepublic void run() {System.out.println("大家都到齐了，开始happy去");}});for (int i = 0; i < 4; i++) {exec.execute(new Runnable() {@Overridepublic void run() {try {Thread.sleep(random.nextInt(10000));} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+ "到了，其他哥们呢");try {barrier.await();// 等待其他哥们} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}});}exec.shutdown();}
}
复制代码
复制代码

Semaphore api：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html

A counting semaphore. Conceptually, a semaphore maintains a set of permits. Each acquire() blocks if necessary until a permit is available, and then takes it. Each release() adds a permit, potentially releasing a blocking acquirer. However, no actual permit objects are used; the Semaphore just keeps a count of the number available and acts accordingly.

例如：对于某个容器，我们规定，最多只能容纳n个线程同时操作 使用信号量来模拟实现

复制代码
复制代码 
public class TestSemaphore {public static void main(String[] args) {ExecutorService exec = Executors.newCachedThreadPool();TestSemaphore t = new TestSemaphore();final BoundedHashSet<String> set = t.getSet();for (int i = 0; i < 3; i++) {// 三个线程同时操作addexec.execute(new Runnable() {public void run() {try {set.add(Thread.currentThread().getName());} catch (InterruptedException e) {e.printStackTrace();}}});}for (int j = 0; j < 3; j++) {// 三个线程同时操作removeexec.execute(new Runnable() {public void run() {set.remove(Thread.currentThread().getName());}});}exec.shutdown();}public BoundedHashSet<String> getSet() {return new BoundedHashSet<String>(2);// 定义一个边界约束为2的线程}class BoundedHashSet<T> {private final Set<T> set;private final Semaphore semaphore;public BoundedHashSet(int bound) {this.set = Collections.synchronizedSet(new HashSet<T>());//①this.semaphore = new Semaphore(bound, true);}public void add(T o) throws InterruptedException {semaphore.acquire();// 信号量控制可访问的线程数目set.add(o);System.out.printf("add:%s%n", o);}public void remove(T o) {if (set.remove(o))semaphore.release();// 释放掉信号量System.out.printf("remove:%s%n", o);}}
}
复制代码
复制代码

解释① ：Collection类中提供了多个synchronizedXxx方法，该方法返回指定集合对象对应的同步对象，从而解决多线程并发访问集合时线程的安全问题。java中常用的HashSet、ArrayList、HashMap都是线程不安全的，如果多条线程访问他们，而且多于一条的线程试图修改它们，则可能出错。以下方法直接将新建的集合传给了Collections的synchronizedXxx方法，这样就直接获取它们的线程安全实现版本。

        Collection c = Collections.synchronizedCollection(new ArrayList());List l = Collections.synchronizedList(new ArrayList());Set s = Collections.synchronizedSet(new HashSet());Map m = Collections.synchronizedMap(new HashMap());

FutureTask api：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/FutureTask.html

A cancellable asynchronous computation. This class provides a base implementation of Future, with methods to start and cancel a computation, query to see if the computation is complete, and retrieve the result of the computation. The result can only be retrieved when the computation has completed; the get methods will block if the computation has not yet completed. Once the computation has completed, the computation cannot be restarted or cancelled (unless the computation is invoked using runAndReset()).

应用举例：我们的算法中有一个很耗时的操作，在编程的是，我们希望将它独立成一个模块，调用的时候当做它是立刻返回的，并且可以随时取消的

复制代码
复制代码 
public class TestFutureTask {public static void main(String[] args) {ExecutorService exec = Executors.newCachedThreadPool();FutureTask<String> task = new FutureTask<String>(new Callable<String>() {// FutrueTask的构造参数是一个Callable接口@Overridepublic String call() throws Exception {return Thread.currentThread().getName();// 这里可以是一个异步操作}});try {exec.execute(task);// FutureTask实际上也是一个线程String result = task.get();// 取得异步计算的结果，如果没有返回，就会一直阻塞等待System.out.printf("get:%s%n", result);} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}}
}
复制代码
复制代码

总结：FutureTask其实就是新建了一个线程单独执行，使得线程有一个返回值，方便程序的编写

Exchanger api：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Exchanger.html

复制代码
复制代码 
package thread;import java.util.ArrayList;
import java.util.concurrent.Exchanger;
/***    1. Exchanger用于在2个线程中交换对象。2. return_object = exchanger.exchange(exch_object)3. 例子中Producer向ArrayList中缓慢填充随机整数，Consumer从另一个ArrayList中缓慢取出整数并输出。4. 当Producer的ArrayList填满，并且Consumer的ArrayList为空时，2个线程才交换ArrayList。* @author Administrator*/
public class ExchangerTest {private static Exchanger<ArrayList<Integer>> exchanger = null;private static ArrayList<Integer> buffer1 = null;private static ArrayList<Integer> buffer2 = null;public static void main(String[] args) throws Exception {exchanger = new Exchanger<ArrayList<Integer>>();buffer1 = new ArrayList<Integer>(10);buffer2 = new ArrayList<Integer>(10);Thread pth = new ProducerThread();Thread cth = new ConsumerThread();pth.start();cth.start();Thread.sleep(60 * 1000);System.out.println("main: interrupting threads.");pth.interrupt();cth.interrupt();pth.join();cth.join();System.out.println("main: end.");}private static class ProducerThread extends Thread {@Overridepublic void run() {ArrayList<Integer> buff = buffer1;try {while (true) {if (buff.size() >= 10) {// 与consumer交换buffer.System.out.println("producer: exchanging.");buff = exchanger.exchange(buff);buff.clear();}// 随机产生一个0-100的整数。int x = (int) (Math.random() * 100);buff.add(x);System.out.println("producer: " + x);// 随机等待0-3秒 。int t = (int) (Math.random() * 3);Thread.sleep(t * 1000);}} catch (InterruptedException e) {System.out.println("producer: interrupted.");}}}private static class ConsumerThread extends Thread {@Overridepublic void run() {ArrayList<Integer> buff = buffer2;try {while (true) {for (Integer x : buff) {System.out.println("consumer: " + x);// 随机等待0-3秒 。int t = (int) (Math.random() * 3);Thread.sleep(t * 1000);}// 与producer交换buffer。System.out.println("consumer: exchanging.");buff = exchanger.exchange(buff);}} catch (InterruptedException e) {System.out.println("consumer: interrupted.");}}}
}
复制代码
复制代码

在JDK1.5之前，我们关于定时/周期操作都是通过Timer来实现的。但是Timer有以下几种危险[JCIP] 
a. Timer是基于绝对时间的。容易受系统时钟的影响。 
b. Timer只新建了一个线程来执行所有的TimeTask。所有TimeTask可能会相关影响 
c. Timer不会捕获TimerTask的异常，只是简单地停止。这样势必会影响其他TimeTask的执行。 
如果你是使用JDK1.5以上版本，建议用ScheduledThreadPoolExecutor代替Timer。它基本上解决了上述问题。它采用相对时间，用线程池来执行TimerTask，会出来TimerTask异常。

复制代码
复制代码 
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;public class TestScheduledThreadPoolExecutor {public static void main(String[] args) {ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1);exec.scheduleAtFixedRate(new Runnable() {// 每隔一段时间就触发异常@Overridepublic void run() {throw new RuntimeException();}}, 1000, 5000, TimeUnit.MILLISECONDS);exec.scheduleAtFixedRate(new Runnable() {// 每隔一段时间打印系统时间，证明两者是互不影响的@Overridepublic void run() {System.out.println(System.nanoTime());}}, 1000, 2000, TimeUnit.MILLISECONDS);}}
复制代码
复制代码

BlockingQueue API文档：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/BlockingQueue.html

A Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.

BlockingQueue的经典用途是 生产者-消费者模式

复制代码
复制代码 
package thread;import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;public class BlockingQueueTest {public static void main(String[] args) {final BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>(3);final Random random = new Random();class Producer implements Runnable {@Overridepublic void run() {while (true) {try {int i = random.nextInt(100);queue.put(i);// 当队列达到容量时候，会自动阻塞的if (queue.size() == 3) {System.out.println("full");Thread.sleep(1000);}} catch (InterruptedException e) {e.printStackTrace();}}}}class Consumer implements Runnable {@Overridepublic void run() {while (true) {try {queue.take();// 当队列为空时，也会自动阻塞Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}}new Thread(new Producer()).start();new Thread(new Consumer()).start();}}
复制代码
复制代码

DelayQueue

在现实生活中，很多DelayQueue的例子。就拿上海的SB会来说明，很多国家地区的开馆时间不同。你很早就来到园区，然后急急忙忙地跑到一些心仪的馆区，发现有些还没开，你吃了闭门羹。 
仔细研究DelayQueue，你会发现它其实就是一个PriorityQueue的封装（按照delay时间排序），里面的元素都实现了Delayed接口，相关操作需要判断延时时间是否到了。 
在实际应用中，有人拿它来管理跟实际相关的缓存、session等 
下面我就通过 “上海SB会的例子来阐述DelayQueue的用法”

复制代码
复制代码 
package thread;
import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class TestDelayQueue {private class Stadium implements Delayed {long trigger;public Stadium(long i) {trigger = System.currentTimeMillis() + i;}@Overridepublic long getDelay(TimeUnit arg0) {long n = trigger - System.currentTimeMillis();return n;}@Overridepublic int compareTo(Delayed arg0) {return (int) (this.getDelay(TimeUnit.MILLISECONDS) - arg0.getDelay(TimeUnit.MILLISECONDS));}public long getTriggerTime() {return trigger;}}public static void main(String[] args) throws Exception {Random random = new Random();DelayQueue<Stadium> queue = new DelayQueue<Stadium>();TestDelayQueue t = new TestDelayQueue();for (int i = 0; i < 5; i++) {queue.add(t.new Stadium(random.nextInt(30000)));}Thread.sleep(2000);while (true) {Stadium s = queue.take();// 延时时间未到就一直等待if (s != null) {System.out.println(System.currentTimeMillis()- s.getTriggerTime());// 基本上是等于0}if (queue.size() == 0)break;}}
}
复制代码
复制代码

总结：适用于需要延时操作的队列管理

SynchronousQueue API：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/SynchronousQueue.html

这个队列其实是BlockingQueue的一种实现。每个插入操作必须等待另一个线程的对应移除操作，反之亦然。它给我们提供了在线程之间交换单一元素的极轻量级方法 
应用举例：我们要在多个线程中传递一个变量。

复制代码
复制代码 
package thread;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
public class TestSynchronousQueue {class Producer implements Runnable {private BlockingQueue<String> queue;List<String> objects = Arrays.asList("one", "two", "three");public Producer(BlockingQueue<String> q) {this.queue = q;}@Overridepublic void run() {try {for (String s : objects) {queue.put(s);// 产生数据放入队列中System.out.printf("put:%s%n", s);}queue.put("Done");// 已完成的标志} catch (InterruptedException e) {e.printStackTrace();}}}class Consumer implements Runnable {private BlockingQueue<String> queue;public Consumer(BlockingQueue<String> q) {this.queue = q;}@Overridepublic void run() {String obj = null;try {while (!((obj = queue.take()).equals("Done"))) {System.out.println(obj);// 从队列中读取对象Thread.sleep(3000); // 故意sleep，证明Producer是put不进去的}} catch (InterruptedException e) {e.printStackTrace();}}}public static void main(String[] args) {BlockingQueue<String> q = new SynchronousQueue<String>();TestSynchronousQueue t = new TestSynchronousQueue();new Thread(t.new Producer(q)).start();new Thread(t.new Consumer(q)).start();}
}

这篇关于java多线程学习--java.util.concurrent的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

Spring Boot集成/输出/日志级别控制/持久化开发实践

Spring Boot集成/输出/日志级别控制/持久化开发实践

《SpringBoot集成/输出/日志级别控制/持久化开发实践》SpringBoot默认集成Logback,支持灵活日志级别配置（INFO/DEBUG等）,输出包含时间戳、级别、类名等信息,并可通过... 目录一、日志概述1.1、Spring Boot日志简介1.2、日志框架与默认配置1.3、日志的核心作用
阅读更多...
破茧 JDBC:MyBatis 在 Spring Boot 中的轻量实践指南

破茧 JDBC:MyBatis 在 Spring Boot 中的轻量实践指南

《破茧JDBC:MyBatis在SpringBoot中的轻量实践指南》MyBatis是持久层框架,简化JDBC开发,通过接口+XML/注解实现数据访问,动态代理生成实现类,支持增删改查及参数... 目录一、什么是 MyBATis二、 MyBatis 入门2.1、创建项目2.2、配置数据库连接字符串2.3、入
阅读更多...
Springboot项目启动失败提示找不到dao类的解决

Springboot项目启动失败提示找不到dao类的解决

《Springboot项目启动失败提示找不到dao类的解决》SpringBoot启动失败,因ProductServiceImpl未正确注入ProductDao,原因：Dao未注册为Bean,解决：在启... 目录错误描述原因解决方法总结***************************APPLICA编
阅读更多...
深度解析Spring Security 中的 SecurityFilterChain核心功能

深度解析Spring Security 中的 SecurityFilterChain核心功能

《深度解析SpringSecurity中的SecurityFilterChain核心功能》SecurityFilterChain通过组件化配置、类型安全路径匹配、多链协同三大特性,重构了Spri... 目录Spring Security 中的SecurityFilterChain深度解析一、Security
阅读更多...
SpringBoot多环境配置数据读取方式

SpringBoot多环境配置数据读取方式

《SpringBoot多环境配置数据读取方式》SpringBoot通过环境隔离机制,支持properties/yaml/yml多格式配置,结合@Value、Environment和@Configura... 目录一、多环境配置的核心思路二、3种配置文件格式详解2.1 properties格式（传统格式）1.
阅读更多...
Apache Ignite 与 Spring Boot 集成详细指南

Apache Ignite 与 Spring Boot 集成详细指南

《ApacheIgnite与SpringBoot集成详细指南》ApacheIgnite官方指南详解如何通过SpringBootStarter扩展实现自动配置,支持厚/轻客户端模式,简化Ign... 目录 一、背景：为什么需要这个集成？ 二、两种集成方式（对应两种客户端模型） 三、方式一：自动配置 Thick
阅读更多...
Spring WebClient从入门到精通

Spring WebClient从入门到精通

《SpringWebClient从入门到精通》本文详解SpringWebClient非阻塞响应式特性及优势,涵盖核心API、实战应用与性能优化,对比RestTemplate,为微服务通信提供高效解决... 目录一、WebClient 概述1.1 为什么选择 WebClient？1.2 WebClient 与
阅读更多...
Java.lang.InterruptedException被中止异常的原因及解决方案

Java.lang.InterruptedException被中止异常的原因及解决方案

《Java.lang.InterruptedException被中止异常的原因及解决方案》Java.lang.InterruptedException是线程被中断时抛出的异常,用于协作停止执行,常见于... 目录报错问题报错原因解决方法Java.lang.InterruptedException 是 Jav
阅读更多...
深入浅出SpringBoot WebSocket构建实时应用全面指南

深入浅出SpringBoot WebSocket构建实时应用全面指南

《深入浅出SpringBootWebSocket构建实时应用全面指南》WebSocket是一种在单个TCP连接上进行全双工通信的协议,这篇文章主要为大家详细介绍了SpringBoot如何集成WebS... 目录前言为什么需要 WebSocketWebSocket 是什么Spring Boot 如何简化 We
阅读更多...
java中pdf模版填充表单踩坑实战记录(itextPdf、openPdf、pdfbox)

java中pdf模版填充表单踩坑实战记录(itextPdf、openPdf、pdfbox)

《java中pdf模版填充表单踩坑实战记录(itextPdf、openPdf、pdfbox)》：本文主要介绍java中pdf模版填充表单踩坑的相关资料,OpenPDF、iText、PDFBox是三... 目录准备Pdf模版方法1：itextpdf7填充表单（1）加入依赖（2）代码（3）遇到的问题方法2：pd
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2