Java 并发编程（四）常用同步工具类

同步工具类可以使任何一种对象，只要该对象可以根据自身的状态来协调控制线程的控制流。阻塞队列可以作为同步工具类，其他类型的同步工具类还包括：信号量（Semaphore）、栅栏（Barrier）以及闭锁（Latch）。

闭锁

首先我们来介绍闭锁。

闭锁作用相当于一扇门：在闭锁到达某一状态之前，这扇门一直是关闭的，所有的线程都会在这扇门前等待（阻塞）。只有门打开后，所有的线程才会同时继续运行。

闭锁可以用来确保某些活动直到其它活动都完成后才继续执行，例如：

1、确保某个计算在其所有资源都被初始化之后才继续执行。二元闭锁（只有两个状态）可以用来表示“资源R已经被初始化”，而所有需要R操作都必须先在这个闭锁上等待。

2、确保某个服务在所有其他服务都已经启动之后才启动。这时就需要多个闭锁。让S在每个闭锁上等待，只有所有的闭锁都打开后才会继续运行。

3、等待直到某个操作的参与者（例如，多玩家游戏中的玩家）都就绪再继续执行。在这种情况下，当所有玩家都准备就绪时，闭锁将到达结束状态。

CountDownLatch 是一种灵活的闭锁实现，可以用在上述各种情况中使用。闭锁状态包含一个计数器，初始化为一个正数，表示要等待的事件数量。countDown() 方法会递减计数器，表示等待的事件中发生了一件。await() 方法则阻塞，直到计数器值变为0。

下面，我们使用闭锁来实现在主线程中计算多个子线程运行时间的功能。具体逻辑是使用两个闭锁，“起始门”用来控制子线程同时运行，“结束门”用来标识子线程是否都结束。

package org.bupt.xiaoye;

import java.util.concurrent.CountDownLatch;

public class Test {
	public static int nThread = 5;
	public static void main(String[] args) throws InterruptedException{
		final CountDownLatch startGate = new CountDownLatch(1);
		final CountDownLatch endGate = new CountDownLatch(nThread);
		for(int i =0;i<nThread;i++){
			new Thread(){
				@Override
				public void run(){
					try{
					startGate.await();
					Thread.sleep(300);}
					catch(InterruptedException e){
					}
					finally{
						System.out.println(Thread.currentThread().getName()+"ended");
						endGate.countDown();
					}
				}
			}.start();
		}
		long start = System.nanoTime();
		startGate.countDown();
		endGate.await();
		long end = System.nanoTime();
		System.out.println("Total time :"+(end - start));
	}
}

FutureTask

FutureTask也可以用作闭锁。它表示一种抽象的可生成结果的计算。是通过 Callable 来实现的，相当于一种可生成结果的 Runnable ，并且可处于以下三种状态：等待运行，正在运行，运行完成。当FutureTask进入完成状态后，它会停留在这个状态上。

Future.get 用来获取计算结果，如果FutureTask还未运行完成，则会阻塞。FutureTask 将计算结果从执行计算的线程传递到获取这个结果的线程，而FutureTask 的规范确保了这种传递过程能实现结果的安全发布。

FutureTask在Executor框架中表示异步任务，还可以用来表示一些时间较长的计算，这些计算可以在使用计算结果之前启动。

下面我们来构造一个简单的异步任务，来简单示范FutureTask的使用方法。

package org.bupt.xiaoye;

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

public class Preloader {
	private final FutureTask<Integer> future = new FutureTask<Integer>(
			new Callable<Integer>() {
				public Integer call() throws Exception {
					Thread.sleep(3000);
					return 969;
				}
			});

	private final Thread thread = new Thread(future);
	public void start() {
		thread.start();
	}

	public Integer get() throws Exception {
		try{
		return future.get();}
		catch(ExecutionException e){
			Throwable cause = e.getCause();
			throw launderThrowable(cause);
		}
	}
	private static Exception launderThrowable(Throwable cause) {
		if(cause instanceof RuntimeException)return (RuntimeException )cause;
		else if(cause instanceof Error) throw (Error) cause;
		else throw new IllegalStateException("Not Checked",cause);
	}
	public static void main(String[] args) throws Exception{
		Preloader p = new Preloader();
		p.start();
		long start = System.currentTimeMillis();
		System.out.println(p.get());
		System.out.println(System.currentTimeMillis()-start);
	}
}

信号量

之前讲的闭锁控制访问的时间，而信号量则用来控制访问某个特定资源的操作数量，控制空间。而且闭锁只能够减少，一次性使用，而信号量则申请可释放，可增可减。 计数信号量还可以用来实现某种资源池，或者对容器施加边界。

Semaphone 管理这一组许可（permit），可通过构造函数指定。同时提供了阻塞方法acquire，用来获取许可。同时提供了release方法表示释放一个许可。

Semaphone 可以将任何一种容器变为有界阻塞容器，如用于实现资源池。例如数据库连接池。我们可以构造一个固定长度的连接池，使用阻塞方法 acquire和release获取释放连接，而不是获取不到便失败。（当然，一开始设计时就使用BlockingQueue来保存连接池的资源是一种更简单的方法）

例如我们将一个普通set容器变为可阻塞有界。

package org.bupt.xiaoye;

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

public class BoundedHashSet<T> {
	private final Set<T> set;
	private Semaphore sem;

	public BoundedHashSet(int bound) {
		if (bound < 1)
			throw new IllegalStateException();
		set = Collections.synchronizedSet(new HashSet<T>());
		sem = new Semaphore(bound);
	}

	public boolean add(T e) throws InterruptedException {
		sem.acquire();
		boolean wasAdded = false;
		try {
			wasAdded = set.add(e);
			return wasAdded;
		} finally {
			if (!wasAdded)
				sem.release();
		}
	}

	public boolean remove(T e) {
		boolean wasRemoved = set.remove(e);
		if (wasRemoved)
			sem.release();
		return wasRemoved;
	}

}

栅栏

栅栏（Bariier）类似于闭锁，它能阻塞一组线程知道某个事件发生。栅栏与闭锁的关键区别在于，所有的线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待等待时间，而栅栏用于等待线程。

CyclicBarrier 可以使一定数量的参与方反复的在栅栏位置汇聚，它在并行迭代算法中非常有用：将一个问题拆成一系列相互独立的子问题。当线程到达栅栏位置时，调用await() 方法，这个方法是阻塞方法，直到所有线程到达了栅栏位置，那么栅栏被打开，此时所有线程被释放，而栅栏将被重置以便下次使用。

另一种形式的栅栏是Exchanger，它是一种两方（Two-Party）栅栏，各方在栅栏位置上交换数据。例如当一个线程想缓冲区写入数据，而另一个线程从缓冲区中读取数据。这些线程可以使用 Exchanger 来汇合，并将慢的缓冲区与空的缓冲区交换。当两个线程通过 Exchanger 交换对象时，这种交换就把这两个对象安全的发布给另一方。

Exchanger 可能被视为 SynchronousQueue 的双向形式。我们也可以用两个SynchronousQueue来实现 Exchanger的功能。

class FillAndEmpty {
   Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
   DataBuffer initialEmptyBuffer = ... a made-up type
   DataBuffer initialFullBuffer = ...

   class FillingLoop implements Runnable {
     public void run() {
       DataBuffer currentBuffer = initialEmptyBuffer;
       try {
         while (currentBuffer != null) {
           addToBuffer(currentBuffer);
           if (currentBuffer.isFull())
             currentBuffer = exchanger.exchange(currentBuffer);
         }
       } catch (InterruptedException ex) { ... handle ... }
     }
   }

   class EmptyingLoop implements Runnable {
     public void run() {
       DataBuffer currentBuffer = initialFullBuffer;
       try {
         while (currentBuffer != null) {
           takeFromBuffer(currentBuffer);
           if (currentBuffer.isEmpty())
             currentBuffer = exchanger.exchange(currentBuffer);
         }
       } catch (InterruptedException ex) { ... handle ...}
     }
   }

   void start() {
     new Thread(new FillingLoop()).start();
     new Thread(new EmptyingLoop()).start();
   }
  }
时间： 2025-01-06 12:10:40

Java 并发编程（四）常用同步工具类的相关文章

java并发编程中常用的工具类 Executor

/***************************************************  * TODO: description .  * @author: gao_chun  * @since:  2015-4-17  * @version: 1.0.0  * @remark: 转载请注明出处  **************************************************/ java.util.concurrent.Executor 使用 Execut

Java并发编程之多线程同步

线程安全就是防止某个对象或者值在多个线程中被修改而导致的数据不一致问题,因此我们就需要通过同步机制保证在同一时刻只有一个线程能够访问到该对象或数据,修改数据完毕之后,再将最新数据同步到主存中,使得其他线程都能够得到这个最新数据.下面我们就来了解Java一些基本的同步机制. Java提供了一种稍弱的同步机制即volatile变量,用来确保将变量的更新操作通知到其他线程.当把变量声明为volatile类型后,编译器与运行时都会注意到这个变量是共享的.然而,在访问volatile变量时不会执行加锁操作

JAVA并发编程4_线程同步之volatile关键字

上一篇博客JAVA并发编程3_线程同步之synchronized关键字中讲解了JAVA中保证线程同步的关键字synchronized,其实JAVA里面还有个较弱的同步机制volatile.volatile关键字是JAVA中的轻量级的同步机制,用来将变量的更新操作同步到其他线程.从内存可见性的角度来说,写入volatile变量相当于退出同步代码块,读取volatile变量相当于进入同步代码块. 旧的内存模型:保证读写volatile都直接发生在main memory中. 在新的内存模型下(1.5)

Java并发编程-非阻塞同步方式原子类（Atomic）的使用

非阻塞同步 在大多数情况下,我们为了实现线程安全都会使用Synchronized或lock来加锁进行线程的互斥同步,但互斥同步的最主要的问题就是进行线程的阻塞和唤醒所带来的性能问题,因此这种阻塞也称作阻塞同步.从处理问题的方式上说,互斥同步属于一种悲观的并发策略,总是认为只要不去做正确的同步措施,那就肯定会出现问题,无论共享数据是否真的会出现竞争,它都会进行加锁.用户态核心态转换.维护锁的计数器和检查是否有被阻塞的线程需要被唤醒等操作. 随着硬件指令集的发展,我们有了另一个选择:基于冲突检测的乐

【Java并发编程二】同步容器和并发容器

一.同步容器 在Java中,同步容器包括两个部分,一个是vector和HashTable,查看vector.HashTable的实现代码,可以看到这些容器实现线程安全的方式就是将它们的状态封装起来,并在需要同步的方法上加上关键字synchornized. 另一个是Collections类中提供的静态工厂方法创建的同步包装类. 同步容器都是线程安全的.但是对于复合操作(迭代.缺少即加入.导航:根据一定的顺序寻找下一个元素),有时可能需要使用额外的客户端加锁进行保护.在一个同步容器中,复合操作是安全

JAVA并发编程3_线程同步之synchronized关键字

在上一篇博客里讲解了JAVA的线程的内存模型,见:JAVA并发编程2_线程安全&内存模型,接着上一篇提到的问题解决多线程共享资源的情况下的线程安全问题. 不安全线程分析 public class Test implements Runnable { private int i = 0; private int getNext() { return i++; } @Override public void run() { // synchronized while (true) { synchro

Java 并发编程(四)：如何保证对象的线程安全性

本篇来谈谈 Java 并发编程:如何保证对象的线程安全性. 01.前言 先让我吐一句肺腑之言吧,不说出来会憋出内伤的.<Java 并发编程实战>这本书太特么枯燥了,尽管它被奉为并发编程当中的经典之作,但我还是忍不住.因为第四章"对象的组合"我整整啃了两周的时间,才啃出来点肉丝. 读者朋友们见谅啊.要怪只能怪我自己的学习能力有限,真读不了这种生硬无趣的技术书.但是为了学习,为了进步,为了将来(口号喊得有点大了),只能硬着头皮上. 请随我来,我尽量写得有趣点. 02.线程安全类

JUC常用同步工具类——CountDownLatch，CyclicBarrier，Semaphore

在 JUC 下包含了一些常用的同步工具类,今天就来详细介绍一下,CountDownLatch,CyclicBarrier,Semaphore 的使用方法以及它们之间的区别. 一.CountDownLatch 先看一下,CountDownLatch 源码的官方介绍. 意思是,它是一个同步辅助器,允许一个或多个线程一直等待,直到一组在其他线程执行的操作全部完成. public CountDownLatch(int count) { if (count < 0) throw new IllegalAr

JAVA 并发编程-传统线程同步通信技术（四）

首先介绍几个概念: wait()方法 wait()方法使得当前线程必须要等待,等到另外一个线程调用notify()或者notifyAll()方法. 当前的线程必须拥有当前对象的monitor,也即lock,就是锁. 线程调用wait()方法,释放它对锁的拥有权,然后等待另外的线程来通知它(通知的方式是notify()或者notifyAll()方法),这样它才能重新获得锁的拥有权和恢复执行. 要确保调用wait()方法的时候拥有锁,即,wait()方法的调用必须放在synchronized方法或s