线程知识复习

一.线程的实现方式

1.继承Thread

2.实现Runnable接口

 

二.线程的状态

1.New(新生线程)

   当你new一个Thread，newThread(r),这时处于线程的新生状态，此时程序还没有真正的运行。

2.Runnable(可运行的)

   当启动start()方法时，此时线程处于可运行状态，不一定运行之中，这取决与线程是否得到CPU的运行时间片。事实上，一个线程并不是一直处于运行状态，偶尔需要被中断，让其他线程有运行的机会。

3.Blocked(被阻塞)

  当发生以下情况被阻塞

  -线程调用sleep方法处于睡眠状态

  -线程进行I/O操作被中断，等待I/O操作完成。

  -线程试图取得一个锁对象，而此时这个所对象被其他线程持有，等待锁被释放

  -线程在等待某个触发条件

  -有人调用了线程的suspend方法，除非等待别人调用resume方法将这个线程挂起。这个方法已经过时，而你不能再你的代码中调用这个方法。

4.Dead(死亡线程)

  -run方法运行完毕导致线程正常死亡

  -因为一个未捕获的异常使得线程死亡

 

三.同步

 1. 当多个线程共享数据的时候，需要注意对数据进行同步保护。可以使用synchronized关键字，您可以将synchronized这种方式看做一个隐式的锁，然后这个隐士的锁只能关联一个条件对象。JDK5.0使用锁对象的概念。

 private Lock myLock = new ReenTrantLock();

 public void sysnMethod(){

    myLock.lock();          //获取锁对象,一旦获得锁，其他线程进入时候被阻塞，等待此线程释放

   try{

   }catch(){

   }finally{

      myLock.unLock();

   }

}

 

每个对象拥有自己的ReenTrantLock对象，不同对象有不同的锁。

2.条件对象

  当某些线程具备某个条件时才能运行。

private Lock myLock = new ReenTrantLock();

 private Condition hasMoreMoney;

hasMoreMoney = myLock.newCondition();//myLock获取一个条件对象

public void transfermoney(){

myLock.lock();  //获取锁

try{

       if(nowMoney<tranferMoney){

         hasMoreMoney.await(); //如果不满足条件，条件对象阻塞此线程，并且释放锁对象，使得此对象进入等待条件集中，等待条件满足。

     }

    hasMoreMoney.signalAll();  //因为线程进入等待条件集中，即使获得锁，也不能被激活，他需要其他线程唤醒他，signal()//随机唤醒等待条件集中线程的任一线程

   }catch(){

   }finally{

      myLock.unLock();

   }

}

一旦线程被唤醒后，并满足条件对象，继续运行程序，而不是重新运行程序。

 

隐式锁和条件缺点：

1.您不能中断一个试图获得锁的线程

2.您不能堆试图获得锁的线程设置超时

3.隐式锁的条件一个显得不够用

4.虚拟机的加锁机制不能很好映射到硬件的加锁机制上

3.常用方法解释

nitifyAll()----------解除在该对象上调用wait()的线程的阻塞，这个方法只能再同步方法或者同步块中使用，如果当前线程不是锁的持有者，抛出异常。

notify()------------随机解除该对象因调用wait()的任一线程的阻塞。

wait()-------------导致线程进入等待状态直到被通知。这个方法只能再同步方法或者同步块中使用。此时放弃对象的锁。

 

四.监视器

1.监视器的特性

2.每个监视器类的对象都有一个相关的锁

3.这个锁负责对所有方法加锁

4.这个锁可以有任意个关联条件

5.Volatile域------为一个同步机制的实例域提供了免锁机制。允许多个线程并发更新。访问一个Volatile变量比访问一般变量要慢，因为它是线程安全的。

在以下三个条件下，对一个域的访问时安全的。

        -域是vilatile的

        -域是final，并且在构造器调用完成后访问

        -对域的访问有锁保护

 

五.死锁

六.公平锁

公平锁会优待那些等待时间较长的线程，但是它会降低了性能。也不能做到真正的公平。

 

七.锁测试和超时

if(myLock.tryLock()){  //测试能否获得锁，获得返回TRUE，否则false

    try{

  }catch(){

  }finally{

    myLock.unLock();

  }

}else{

   //做别的事情

}

 

可以myLock.teyLock(100,TimeUnit.MILLSECONDS)  //设置超时时间

 

八.读/写锁

读/写锁的必要步骤

1.创建一个ReentrantReadWriteLock对象

ReentrantReadWriteLock rw = new ReentrantReadWriteLock();

2.抽取读锁和写锁

Lock readLock = rw.readLock();

Lock writeLock = rw.writeLock();

3.对读取得线程加读锁

public void getAllMoney{

readLock.lock();

 try{

  ......

 }finally{

   readLock.unLock();

 }

}

4.对写数据的线程加写锁

 public void transfer(){

writeLock .lock();

 try{

  ......

 }finally{

   writeLock .unLock();

 }

}

 

九.stop和suspend方法为什么弃用

stop方法会破坏对象，当一个线程获得一个锁对象时，取款后，但是被stop，但是没有存入，最后终止了，那总款不正确，suspend不会破坏对象，但是很容易导致死锁，因为suspend挂起一个拥有锁的线程，它不会释放锁，而等待别人resume，这样，加入别人没有resume，而另外一个线程正在试图得到锁而被挂起，最后死锁了。

 

十.阻塞队列

相信大家对队列都很熟悉，队列抱着先进先出的原则。

BlockingQueue的操作方法               

add	向队列添加一个元素	队列满，抛出一个异常
remove	删除一个队列元素	队列为空，抛出一个异常
element	返回队列头部元素	队列为空，抛出一个异常
offer	添加元素，返回true	如果队列满，返回false
poll	删除并返回头部元素，返回true	队列空，返回null
peek	返回头部元素	队列空，返回null
put	添加一个元素	队列满，阻塞
take	获取并移除此队列的头部	队列空，阻塞

 有一个经典的例子来说明阻塞队列：

package thread.queue;

import java.io.File;
import java.util.concurrent.BlockingQueue;

import javax.management.Query;

public class FileEnumerationTask implements Runnable{

	private BlockingQueue<File> queue;
	public static File DUMMY = new File("");
	private File startingDirectory;
	
	
	public FileEnumerationTask(BlockingQueue<File> queue, File startingDirectory) {
		super();
		this.queue = queue;
		this.startingDirectory = startingDirectory;
	}

	public void enumrate(File directory) throws InterruptedException{
		File[] files=directory.listFiles();
		for(File f:files){
			if(f.isDirectory()){
				enumrate(f);
			}else{
				queue.put(f);
			}
		}
	}
	@Override
	public void run() {
		// TODO Auto-generated method stub
		
		try {
			enumrate(startingDirectory);
			queue.put(DUMMY);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

}

 

package thread.queue;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.Scanner;
import java.util.concurrent.BlockingQueue;

public class SearchTask implements Runnable {
	private BlockingQueue<File> queue;
	private String keyWord;
	
	
	public SearchTask(BlockingQueue<File> queue, String keyWord) {
		super();
		this.queue = queue;
		this.keyWord = keyWord;
	}

	public void search(File file) throws FileNotFoundException{
		Scanner in  = new Scanner(new FileInputStream(file));
		int lineNumber=0;
		while(in.hasNextLine()){
			lineNumber++;
			String line = in.nextLine();
			if(line.contains(keyWord)){
				System.out.printf("%s:%d:%s%n",file.getPath(),lineNumber,line);
			}
		}
		in.close();
	}

	@Override
	public void run() {
		// TODO Auto-generated method stub
		boolean done =false;
		while(!done){
			try {
				File file  = queue.take();
				if(file==FileEnumerationTask.DUMMY){
					queue.put(file);   //如果是结束文件符号，此线程退出循环，并将这个结束文件符号还交给队列
					done=true;
				}else{
					try {
						search(file);
					} catch (FileNotFoundException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

}

 

package thread.queue;

import java.io.File;
import java.util.Scanner;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueTest {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
	/*	Scanner in = new Scanner(System.in);
		System.out.println("Enter basic directory:");
		String startingDirectory = in.nextLine();
		System.out.println("Enter keyword:");
		String keyword = in.nextLine();*/
		
		String startingDirectory="D:/test";
		String keyword="chen";
		final int FILE_QUEUE_SIZE=10;
		final int SEARCH_THREADS=100;
		BlockingQueue<File> queues = new ArrayBlockingQueue<File>(FILE_QUEUE_SIZE) ;
		FileEnumerationTask fTask = new FileEnumerationTask(queues, new File(startingDirectory));
		new Thread(fTask).start();
		for(int i=1;i<=SEARCH_THREADS;i++){
			new Thread(new SearchTask(queues,keyword)).start();
		}
	}

}

 

 

十一.旧的线程安全集合

 JDK1.0开始,Vector和Hashtable都是线程安全的集合，但是JDK1.2被弃用，代替的是ArrayList和HashMap。但是他们不是线程安全的。不过他们可以使用同步包装器(synchornization wapper)包装成线程安全的：

List synList = Collections.synchronizedList(new ArrayList<E>());

Map synMap = Collections.synchronizedMap(new HashMap());

经过同步包装的Collection对象被一个锁保护，如果你想迭代这个Collection，需要使用同步块来保护。

 

十二.Callable和Future(对于这一块，个人理解也不是非常清楚)

Runnable封装的是异步运行的的任务，您可以把它想象成没有任何参数和返回值的异步方法。Callable和Runnable相似，但是它有参数类型和返回值，只有一个方法call()

interface Callable<V>{

   V call() throws Exception;

}

Future对象保存异步计算的结果。Future对象的持有者在结果计算完毕后能得到它。

public interface Future<V>{

  V get();                                        //被阻塞，直至计算完成。

  V get(long timeout,Timeunit unit);     //如果在计算完成前超时，抛出异常，如果运行时线程被中断，抛中断异常，如果计算完成，返回结果。

  void cancel(boolean myInterrupt);     //如果计算还没有开始，那么将永远不会开始，如果已经在计算之中，那么被中断

  boolean isCancel();

  boolean isDone();

}

 

十三.线程池

1.执行器(Executor)

下面有几种构建线程池的静态工厂方法：

--------------------------------------ExecutorService接口-----------------------------

newCachedThreadPool-------------------在需要的时候创建新线程，空闲的时候保存60秒

newFixedThreadPool---------------------池中保存一定数量的线程，空闲的时候也保存

newSingleThreadExecutor----------------只有一个线程的池，这个线程顺寻执行提交的任务

--------------------------------ScheduledExecutorService接口-----------------------

newScheduledThreadPool----------------为预定执行而构建的固定线程池

newSingleThreadScheduledThreadPool----为预定执行而构建的单线程的池

 

下面看一个例子来解释：

(1).为使用线程池的概念的核心代码(这个例子会产生很多子线程)

 

/**********************************/
MatchCounter mc = new MatchCounter(new File(directory), keyword); //一个实现了Callable的类，有个Call方法计算文件个数
FutureTask<Integer> task = new FutureTask<Integer>(mc); //最后结果存在FutureTask
	Thread t = new Thread(task);
	t.start();

 

 (2).使用线程池的概念

 

ExecutorService pool = Executors.newCachedThreadPool();
		MatchCounter mc = new MatchCounter(new File(directory), keyword,pool);
		Future<Integer> task = pool.submit(mc);

MatchCounter的代码如下：

 

package thread.threadpool;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.Scanner;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

public class MatchCounter implements Callable<Integer> {
	private File directory;
	private String keyword;
	private ExecutorService pool;
	private int count;
	
	public MatchCounter(File directory, String keyword,ExecutorService pool) {
		super();
		this.directory = directory;
		this.keyword = keyword;
		this.pool = pool;
	}

	
	
	@Override
	public Integer call() throws Exception {
		// TODO Auto-generated method stub
		int count=0;
		
		File[] files = directory.listFiles();
		
		ArrayList<Future<Integer>> results = new ArrayList<Future<Integer>>();
		
		for(File file:files){
			if(file.isDirectory()){
				MatchCounter counter = new MatchCounter(file, keyword,pool);
			/*	FutureTask<Integer> task = new FutureTask<Integer>(counter);
				results.add(task);
				Thread t = new Thread(task);
				t.start();*/
				Future<Integer> result = pool.submit(counter);
				results.add(result);
			}else{
				if(search(file)){
					count++;
				}
			}
			
			for(Future<Integer> f :results){
				count+=f.get();
			}
		}
		
		return count;
	}
	
	
	public boolean search(File file){
		
		try {
			Scanner in = new Scanner(new FileInputStream(file));
			boolean founds= false;
			while(!founds&&in.hasNextLine()){
				String line = in.nextLine();
				if(line.contains(keyword)){
					founds =true;
				}
			}
			in.close();
			return founds;
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			return false;
		}
	}
}

 

 

 

那么，线程池所要做的事情如下：

(1).调用Executors的静态工厂方法newCachedThreadPool或者newFixedThreadPool。

(2).调用submit来提交一个Runnable或者Callable对象

(3).如果能够希望取消任务或者如果提交的是Callable对象，那么保存好返回的Future对象

(4).当不想再提交任何任务时调用shutdown

 

接着看看预定义执行接口的实现：

SchduledExecutorService具有为预定或重复执行任务而设计的方法，可以预定Runnable或者Callable预定只运行一次或者周期性运行。newScheduledThreadPool，newSingleThreadScheduledThreadPool返回的是SchduledExecutorService的对象。

ScheduledFuture schedule(Callable<V> callable, long delay, TimeUnit unit)----预定在给定的时间执行任务

 scheduleAtFixedRate(Runnable command, long initialDelay, long period,TimeUnit unit)---

创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期；也就是将在 initialDelay 后开始执行，然后在 initialDelay+period 后执行，接着在 initialDelay + 2 * period 后执行，依此类推。如果任务的任何一个执行遇到异常，则后续执行都会被取消。否则，只能通过执行程序的取消或终止方法来终止该任务。如果此任务的任何一个执行要花费比其周期更长的时间，则将推迟后续执行，但不会同时执行。

scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit)

创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。如果任务的任一执行遇到异常，就会取消后续执行。否则，只能通过执行程序的取消或终止方法来终止该任务。

 

十四.控制线程组(ExecutorCompletionService)

ExecutorService    invokeAny(Collection<? extends Callable<T>> tasks)----执行给定的任务，如果某个任务已成功完成（也就是未抛出异常），则返回其结果。一旦正常或异常返回后，则取消尚未完成的任务。如果此操作正在进行时修改了给定的 collection，则此方法的结果是不确定的。

List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout,TimeUnit unit)
执行给定的任务，当所有任务完成时，返回保持任务状态和结果的 Future 列表。返回列表的所有元素的 Future.isDone() 为 true。注意，可以正常地或通过抛出异常来终止已完成 任务。如果正在进行此操作时修改了给定的 collection，则此方法的结果是不确定的。

缺点：如果有一个线程运行花很多的时间，你不得不去等待。

ExecutorCompletionService    new ExecutorCompletionService(Executor e)

submit(Callable<V> task)  ---------提交要执行的值返回任务，并返回表示挂起的任务结果的 Future。在完成时，可能会提取或轮询此任务。

submit(Runnable task,V result)----提交要执行的 Runnable 任务，并返回一个表示任务完成的 Future，可以提取或轮询此任务.

 

十五.障栅(CyclicBarrier)

当考虑大量线程分别计算不同部分时。所有部分计算完毕后，需要整合所有部分时，可以用到障栅。它的原理是，当某一个线程运行完它的部分时，就在障栅出等待，知道所有部分完成后，障栅结束，等待线程释放。继续运行。

CyclicBarrier barrier = new CyclicBarrier(nThread)   // 参与运行的线程数

public void run(){

   doWork();    //每个线程都要做的工作

  barrier.await();

}

障栅是循环的，它可以在所有等待线程释放后被重用。

 

十六.倒计时门拴(CountDownLatch)

一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。

用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。

CountDownLatch 是一个通用同步工具，它有很多用途。将计数 1 初始化的 CountDownLatch 用作一个简单的开/关锁存器，或入口：在通过调用 countDown() 的线程打开入口前，所有调用 await 的线程都一直在入口处等待。用 N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待，或者使其在某项操作完成 N 次之前一直等待。

CountDownLatch 的一个有用特性是，它不要求调用 countDown 方法的线程等到计数到达零时才继续，而在所有线程都能通过之前，它只是阻止任何线程继续通过一个 await。

另一种典型用法是，将一个问题分成 N 个部分，用执行每个部分并让锁存器倒计数的 Runnable 来描述每个部分，然后将所有 Runnable 加入到 Executor 队列。当所有的子部分完成后，协调线程就能够通过 await。（当线程必须用这种方法反复倒计数时，可改为使用 CyclicBarrier。）

 

倒计时门拴和障栅区别：

(1)不是所有线程都等到门拴打开为止

(2)门拴可以由外部事件打开

(3)倒计时门拴是一次性的，一旦计数到达0，不可重用

 

十七.交换器(Exchanger)

当两个线程工作一个一个数据缓冲区的两个实例上，可以使用交换器，一个线程向数据缓冲区注入数据，另一个线程读取数据，当完成后，它们相互交换缓冲。

 

十八.同步列(SynchronousQueue)

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

 

十九.信号量(Semaphore)(不是很理解)

一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。