首页 > 代码库 > 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基础系列之(二) - 线程