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

标签：java util 线程 import new 多线程 public

java多线程学习-java.util.concurrent详解(一) Latch/Barrier

    Java1.5提供了一个非常高效实用的多线程包：java.util.concurrent, 提供了大量高级工具，可以帮助开发者编写高效、易维护、结构清晰的Java多线程程序。从这篇blog起，我将跟大家一起共同学习这些新的Java多线程构件

1. CountDownLatch
    我们先来学习一下JDK1.5 API中关于这个类的详细介绍：
“一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。用给定的计数初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。”

    这就是说，CountDownLatch可以用来管理一组相关的线程执行，只需在主线程中调用CountDownLatch 的await方法（一直阻塞），让各个线程调用countDown方法。当所有的线程都只需完countDown了，await也顺利返回，不再阻塞了。在这样情况下尤其适用：将一个任务分成若干线程执行，等到所有线程执行完，再进行汇总处理。

    下面我举一个非常简单的例子。假设我们要打印1-100，最后再输出“Ok“。1-100的打印顺序不要求统一，只需保证“Ok“是在最后出现即可。

    解决方案：我们定义一个CountDownLatch，然后开10个线程分别打印（n-1）*10+1至（n-1）*10+10。主线程中调用await方法等待所有线程的执行完毕，每个线程执行完毕后都调用countDown方法。最后再await返回后打印“Ok”。

具体代码如下（本代码参考了JDK示例代码）：

Java代码

import java.util.concurrent.CountDownLatch;
/**
 * 示例：CountDownLatch的使用举例
 * @author janeky
 */
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();
			}
		}
	}
}

总结：CounDownLatch对于管理一组相关线程非常有用。上述示例代码中就形象地描述了两种使用情况。第一种是计算器为1，代表了两种状态，开关。第二种是计数器为N，代表等待N个操作完成。今后我们在编写多线程程序时，可以使用这个构件来管理一组独立线程的执行。

2. CyclicBarrier

我们先来学习一下JDK1.5 API中关于这个类的详细介绍：

“一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环的 barrier。

CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作很有用。

我们在学习CountDownLatch的时候就提到了CyclicBarrier。两者究竟有什么联系呢？引用[JCIP]中的描述“The key difference is that with a barrier, all the threads must come together at a barrier point at the same time in order to proceed. Latches are for waiting for events; barriers are for waiting for other threads。CyclicBarrier等待所有的线程一起完成后再执行某个动作。这个功能CountDownLatch也同样可以实现。但是CountDownLatch更多时候是在等待某个事件的发生。在CyclicBarrier中，所有的线程调用await方法，等待其他线程都执行完。

举一个很简单的例子，

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

代码如下（参考了网上给的一些教程）

Java代码

import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TestCyclicBarrier {

	public static void main(String[] args) {
	
		ExecutorService exec = Executors.newCachedThreadPool();     
		final Random random=new Random();
		
		final CyclicBarrier barrier=new CyclicBarrier(4,new Runnable(){
			@Override
			public void run() {
				System.out.println("大家都到齐了，开始happy去");
			}});
		
		for(int i=0;i<4;i++){
			exec.execute(new Runnable(){
				@Override
				public void run() {
					try {
						Thread.sleep(random.nextInt(1000));
					} 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();
	}

}

关于await方法要特别注意一下，它有可能在阻塞的过程中由于某些原因被中断

总结：CyclicBarrier就是一个栅栏，等待所有线程到达后再执行相关的操作。barrier 在释放等待线程后可以重用。

更多的Java编程资料，欢迎访问我的blog：

http://janeky.iteye.com，希望能够与你有更多的交流

未完待续

src.rar (1.3 KB)
下载次数: 24

java多线程学习-java.util.concurrent详解(二)Semaphore/FutureTask/Exchanger

前一篇文章 http://janeky.iteye.com/category/124727
我们学习了java.util.concurrent的CountDownLatch和CyclicBarrier
今天我们继续共同来探讨其他的多线程组件
-----------------------------------------------------------------------------
3. Semaphore
    我们先来学习一下JDK1.5 API中关于这个类的详细介绍：
“一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。”

    我们一般用它来控制某个对象的线程访问对象

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

具体代码如下（参考 [JCIP]）

Java代码

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

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++) {//三个线程同时操作add
			exec.execute(new Runnable() {
				public void run() {
					try {
						set.add(Thread.currentThread().getName());
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			});
		}

		for (int j = 0; j < 3; j++) {//三个线程同时操作remove
			exec.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);
		}
	}
}

总结：Semaphore通常用于对象池的控制

4．FutureTask

我们先来学习一下JDK1.5 API中关于这个类的详细介绍：

“取消的异步计算。利用开始和取消计算的方法、查询计算是否完成的方法和获取计算结果的方法，此类提供了对 Future 的基本实现。仅在计算完成时才能获取结果；如果计算尚未完成，则阻塞 get 方法。一旦计算完成，就不能再重新开始或取消计算。

可使用 FutureTask 包装 Callable 或 Runnable 对象。因为 FutureTask 实现了 Runnable，所以可将 FutureTask 提交给 Executor 执行。

除了作为一个独立的类外，此类还提供了 protected 功能，这在创建自定义任务类时可能很有用。 “

应用举例：

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

具体代码如下（参考 [JCIP]）

Java代码

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

public class TestFutureTask {

	public static void main(String[] args) {
		ExecutorService exec=Executors.newCachedThreadPool();
		
		FutureTask<String> task=new FutureTask<String>(new Callable<String>(){//FutrueTask的构造参数是一个Callable接口
			@Override
			public 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其实就是新建了一个线程单独执行，使得线程有一个返回值，方便程序的编写

5. Exchanger

我们先来学习一下JDK1.5 API中关于这个类的详细介绍：

“可以在pair中对元素进行配对和交换的线程的同步点。每个线程将条目上的某个方法呈现给 exchange 方法，与伙伴线程进行匹配，并且在返回时接收其伙伴的对象。Exchanger 可能被视为 SynchronousQueue 的双向形式。Exchanger 可能在应用程序（比如遗传算法和管道设计）中很有用。 “

应用举例：

有两个缓存区，两个线程分别向两个缓存区fill和take，当且仅当一个满了，两个缓存区交换

Java代码

import java.util.ArrayList;
import java.util.concurrent.Exchanger;

public class TestExchanger {

	public static void main(String[] args) {
		final Exchanger<ArrayList<Integer>> exchanger = new Exchanger<ArrayList<Integer>>();
		final ArrayList<Integer> buff1 = new ArrayList<Integer>(10);
		final ArrayList<Integer> buff2 = new ArrayList<Integer>(10);

		new Thread(new Runnable() {
			@Override
			public void run() {
				ArrayList<Integer> buff = buff1;
				try {
					while (true) {
						if (buff.size() >= 10) {
							buff = exchanger.exchange(buff);//开始跟另外一个线程交互数据
							System.out.println("exchange buff1");
							buff.clear();
						}
						buff.add((int)(Math.random()*100));
						Thread.sleep((long)(Math.random()*1000));
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}).start();
		
		new Thread(new Runnable(){
			@Override
			public void run() {
				ArrayList<Integer> buff=buff2;
				while(true){
					try {
						for(Integer i:buff){
							System.out.println(i);
						}
						Thread.sleep(1000);
						buff=exchanger.exchange(buff);//开始跟另外一个线程交换数据
						System.out.println("exchange buff2");
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}}).start();
	}
}

总结：Exchanger在特定的使用场景比较有用（两个伙伴线程之间的数据交互）

----------------------------------------------------------------------------------

更多的java多线程资料，欢迎访问

http://janeky.iteye.com/category/124727

java多线程学习-java.util.concurrent详解(三)ScheduledThreadPoolExecutor

前一篇blog http://janeky.iteye.com/category/124727我们学习了java多线程的信号量/FutureTask
----------------------------------------------------------------------------------

6. ScheduledThreadPoolExecutor
    我们先来学习一下JDK1.5 API中关于这个类的详细介绍：

    "可另行安排在给定的延迟后运行命令，或者定期执行命令。需要多个辅助线程时，或者要求 ThreadPoolExecutor 具有额外的灵活性或功能时，此类要优于 Timer。
    一旦启用已延迟的任务就执行它，但是有关何时启用，启用后何时执行则没有任何实时保证。按照提交的先进先出 (FIFO) 顺序来启用那些被安排在同一执行时间的任务。

    虽然此类继承自 ThreadPoolExecutor，但是几个继承的调整方法对此类并无作用。特别是，因为它作为一个使用 corePoolSize 线程和一个无界队列的固定大小的池，所以调整 maximumPoolSize 没有什么效果。"

    在JDK1.5之前，我们关于定时/周期操作都是通过Timer来实现的。但是Timer有以下几种危险[JCIP]

a. Timer是基于绝对时间的。容易受系统时钟的影响。
b. Timer只新建了一个线程来执行所有的TimeTask。所有TimeTask可能会相关影响
c. Timer不会捕获TimerTask的异常，只是简单地停止。这样势必会影响其他TimeTask的执行。

    如果你是使用JDK1.5以上版本，建议用ScheduledThreadPoolExecutor代替Timer。它基本上解决了上述问题。它采用相对时间，用线程池来执行TimerTask，会出来TimerTask异常。

    下面通过一个简单的实例来阐述ScheduledThreadPoolExecutor的使用。

我们定期让定时器抛异常
    我们定期从控制台打印系统时间

代码如下（参考了网上的一些代码，在此表示感谢）

Java代码

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(){//每隔一段时间就触发异常
			@Override
			public void run() {
				throw new RuntimeException();
			}}, 1000, 5000, TimeUnit.MILLISECONDS);
		
		exec.scheduleAtFixedRate(new Runnable(){//每隔一段时间打印系统时间，证明两者是互不影响的
			@Override
			public void run() {
				System.out.println(System.nanoTime());
			}}, 1000, 2000, TimeUnit.MILLISECONDS);
	}

}

总结：是时候把你的定时器换成 ScheduledThreadPoolExecutor了

--------------------------------------------------------------------

更多的java多线程资料，欢迎访问

http://janeky.iteye.com/category/124727

java多线程学习-java.util.concurrent详解(四) BlockingQueue

前面一篇blog http://janeky.iteye.com/category/124727
我们主要探讨了ScheduledThreadPoolExecutor定时器的使用
---------------------------------------------------------------------------------
7.BlockingQueue
    “支持两个附加操作的 Queue，这两个操作是：获取元素时等待队列变为非空，以及存储元素时等待空间变得可用。“

    这里我们主要讨论BlockingQueue的最典型实现：LinkedBlockingQueue 和ArrayBlockingQueue。两者的不同是底层的数据结构不够，一个是链表，另外一个是数组。

    后面将要单独解释其他类型的BlockingQueue和SynchronousQueue

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

    代码如下：

Java代码

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class TestBlockingQueue {

	public static void main(String[] args) {
		final BlockingQueue<Integer> queue=new LinkedBlockingQueue<Integer>(3);
		final Random random=new Random();
		
		class Producer implements Runnable{
			@Override
			public void run() {
				while(true){
					try {
						int i=random.nextInt(100);
						queue.put(i);//当队列达到容量时候，会自动阻塞的
						if(queue.size()==3)
						{
							System.out.println("full");
						}
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		}
		
		class Consumer implements Runnable{
			@Override
			public 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();
	}

}

总结：BlockingQueue使用时候特别注意take 和 put

8. DelayQueue

我们先来学习一下JDK1.5 API中关于这个类的详细介绍：

“它是包含Delayed 元素的一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的 Delayed 元素。如果延迟都还没有期满，则队列没有头部，并且 poll 将返回 null。当一个元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于等于 0 的值时，将发生到期。即使无法使用 take 或 poll 移除未到期的元素，也不会将这些元素作为正常元素对待。例如，size 方法同时返回到期和未到期元素的计数。此队列不允许使用 null 元素。”

在现实生活中，很多DelayQueue的例子。就拿上海的SB会来说明，很多国家地区的开馆时间不同。你很早就来到园区，然后急急忙忙地跑到一些心仪的馆区，发现有些还没开，你吃了闭门羹。

仔细研究DelayQueue，你会发现它其实就是一个PriorityQueue的封装（按照delay时间排序），里面的元素都实现了Delayed接口，相关操作需要判断延时时间是否到了。

在实际应用中，有人拿它来管理跟实际相关的缓存、session等

下面我就通过 “上海SB会的例子来阐述DelayQueue的用法”

代码如下：

Java代码

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;
		}
		
		@Override
		public long getDelay(TimeUnit arg0) {
			long n=trigger-System.currentTimeMillis();
			return n;
		}

		@Override
		public 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;
		}
	}
}

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

9. SynchronousQueue

我们先来学习一下JDK1.5 API中关于这个类的详细介绍：

“一种阻塞队列，其中每个插入操作必须等待另一个线程的对应移除操作，反之亦然。同步队列没有任何内部容量，甚至连一个队列的容量都没有。不能在同步队列上进行 peek，因为仅在试图要移除元素时，该元素才存在；除非另一个线程试图移除某个元素，否则也不能（使用任何方法）插入元素；也不能迭代队列，因为其中没有元素可用于迭代。队列的头是尝试添加到队列中的首个已排队插入线程的元素；如果没有这样的已排队线程，则没有可用于移除的元素并且 poll() 将会返回 null。对于其他 Collection 方法（例如 contains），SynchronousQueue 作为一个空 collection。此队列不允许 null 元素。

同步队列类似于 CSP 和 Ada 中使用的 rendezvous 信道。它非常适合于传递性设计，在这种设计中，在一个线程中运行的对象要将某些信息、事件或任务传递给在另一个线程中运行的对象，它就必须与该对象同步。 “

看起来很有意思吧。队列竟然是没有内部容量的。这个队列其实是BlockingQueue的一种实现。每个插入操作必须等待另一个线程的对应移除操作，反之亦然。它给我们提供了在线程之间交换单一元素的极轻量级方法

应用举例：我们要在多个线程中传递一个变量。

代码如下（其实就是生产者消费者模式）

Java代码

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;
		}

		@Override
		public 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;
		}

		@Override
		public 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();
	}

}

总结：SynchronousQueue主要用于单个元素在多线程之间的传递

------------------------------------------------------------

更多的Java多线程资料，欢迎访问

http://janeky.iteye.com/category/124727

标签：java,util,线程,import,new,多线程,public
From： https://blog.51cto.com/u_16174476/6883033

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

java多线程学习-java.util.concurrent详解(一) Latch/Barrier

java多线程学习-java.util.concurrent详解(二)Semaphore/FutureTask/Exchanger

java多线程学习-java.util.concurrent详解(三)ScheduledThreadPoolExecutor

java多线程学习-java.util.concurrent详解(四) BlockingQueue

相关文章

赞助商

阅读排行