浅谈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 只对可用许可的号码进行计数，并采取相应的行动。