首页 > 代码库 > Java基础系列之(二) - 线程
Java基础系列之(二) - 线程
一.线程的实现方式
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 只对可用许可的号码进行计数，并采取相应的行动。
Java基础系列之(二) - 线程