首页 > 代码库 > 线程通信
线程通信
线程通信
一 使用Synchronized的线程
1.当线程在系统内运行时，线程的调度具有一定的透明性，程序通常无法准确控制线程的轮换执行，但java也提供了一些机制来保证线程协调运行。
Object类提供了wait(),notify()和notifyAll()三个方法，这三个方法属于Object类，但是必须由同步监视器来调用，可以分为以下两种情况：
(1)对于使用synchronized修饰的同步方法，因为该类的默认实例this就是同步监视器，所以可以在同步方法中直接调用这三个方法
(2)对于使用synchronized修饰的同步代码块，同步监视器是synchronized后括号里的对象，所以必须使用该对象调用这三个方法
2.wait():导致当前线程等待，直到其他线程调用该同步监视器的notify()方法或notifyAll()方法来唤醒该线程。无时间参数的wait会一直等待，直到其他线程通知，带毫秒的wait会等到指定时间后自动苏醒。调用wait()方法的当前线程会释放对该同步监视器的锁定。
notify():唤醒在此同步监视器上等待的单个线程。如果所有线程都是在此同步监视器上等待，则会选择唤醒其中一个线程，选择是任意的。
notifyAll():唤醒在此同步监视器上等待的所有线程。
3.假设在系统中有两个线程，分别代表存款者和取钱者，系统要求存款者和取钱者不断重复存钱、取钱的动作，而且要求每当存款者将钱存入指定账户后，取钱者就立即取出该笔钱，不允许存款者连续两次存钱，也不允许取钱者连续两次取钱。
程序中可以通过一个旗标来标识账户中是否已有存款，当旗标为false时，账户中没有存款，存款者线程可以向下执行，当存款者存款后，旗标为true，并调用notify()或notifyAll()方法来唤醒其他线程，并调用wait()让存款者线程等待，取款者的操作类似。
Account.java
public class Account { private String accountNo; private double balance; private boolean flag=false; public Account(String accountNo,double balance){ this.accountNo=accountNo; this.balance=balance; } //因为账户余额不可以随便更改，所以只为balance提供getter方法 public double getBalance() { return balance; } public String getAccountNo() { return accountNo; } public void setAccountNo(String accountNo) { this.accountNo = accountNo; } @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; Account account = (Account) o; return accountNo.equals(account.accountNo); } @Override public int hashCode() { return accountNo.hashCode(); } public synchronized void draw(double drawAmount){ try{ if(!flag){ wait(); }else{ System.out.println(Thread.currentThread().getName()+"取钱："+drawAmount); balance-=drawAmount; System.out.println("账户余额为："+balance); flag=false; notifyAll(); } }catch(InterruptedException ex){ ex.printStackTrace(); } } public synchronized void depoist(double depoistAmount){ try{ if(flag){ wait(); }else{ System.out.println(Thread.currentThread().getName()+"存款："+depoistAmount); balance+=depoistAmount; System.out.println("账户余额为："+balance); flag=true; notifyAll(); } }catch(InterruptedException ex){ ex.printStackTrace(); } } }
DrawThread.java
public class DrawThread extends Thread {
private Account account;
private double drawAmount;
public DrawThread(String name, Account account, double drawAmount) {
super(name);
this.account = account;
this.drawAmount = drawAmount;
}
public void run(){
for(int i=0;i<100;i++) {
account.draw(drawAmount);
}
}
}
DepositThread.java
public class DepositThread extends Thread { private Account account; private double depositAmount; public DepositThread(String name, Account account, double depositAmount) { super(name); this.account = account; this.depositAmount = depositAmount; } public void run(){ for(int i=0;i<100;i++){ account.depoist(depositAmount); } } }
DrawTest.java
public class DrawTest { public static void main(String[] args){ Account acct=new Account("1234567",0); new DrawThread("取钱者",acct,800).start(); new DepositThread("存款者甲",acct,800).start(); new DepositThread("存钱者乙",acct,800).start(); new DepositThread("存钱者丙",acct,800).start(); } }
结果：
存款者甲存款：800.0
账户余额为：800.0
取钱者取钱：800.0
账户余额为：0.0
从结果可以发现，三个存款者线程随机的向账户中存款，只有一个取款者取钱。但是程序最后被阻塞无法继续向下执行，这是因为三个存款者线程共有300此存款操作，而一个取钱者线程只有100次取钱操作，程序最后被阻塞，如下：
存钱者丙存款：800.0
账户余额为：800.0
这里的阻塞并不是死锁，只是取钱者线程已经执行完毕，而存款者线程还在等待其他线程来取钱而已，并不是等待其他线程释放同步监视器。
二 使用Condition控制线程通信
1.如果程序中直接使用Lock对象来保证同步，则系统中不存在隐式的同步监视器，就不能使用wait()，notify()，notifyAll()进行线程通信了。Java提供了一个Condition来保持协调，使用Condition可以让那些已经得到Lock对象却无法继续执行的线程释放Lock锁，也可以唤醒其他处于等待的线程。Condition实例被绑定在了一个Lock对象上，只要调用Lock对象的newCondition()方法即可。
2.Condition提供了如下三个方法：
await():同wait()方法，但是该方法有更多的变体，如long awaitNanos(long nanosTimeout)，void awaitUninterruptibly()，awaitUtil(Date deadline)等
signal():同notify()
signalAll():同notifyAll()
3.Account.java
public class Account { private final Lock lock=new ReentrantLock(); private final Condition cond=lock.newCondition(); private String accountNo; private double balance; private boolean flag=false; public Account(String accountNo,double balance){ this.accountNo=accountNo; this.balance=balance; } //因为账户余额不可以随便更改，所以只为balance提供getter方法 public double getBalance() { return balance; } public String getAccountNo() { return accountNo; } public void setAccountNo(String accountNo) { this.accountNo = accountNo; } @Override public boolean equals(Object o) { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; Account account = (Account) o; return accountNo.equals(account.accountNo); } @Override public int hashCode() { return accountNo.hashCode(); } public synchronized void draw(double drawAmount){ lock.lock(); try{ if(!flag){ cond.await(); }else{ System.out.println(Thread.currentThread().getName()+"取钱："+drawAmount); balance-=drawAmount; System.out.println("账户余额为："+balance); flag=false; cond.signalAll(); } }catch(InterruptedException ex){ ex.printStackTrace(); }finally{ lock.unlock(); } } public synchronized void deposit(double depositAmount){ lock.lock(); try{ if(flag){ cond.await(); }else{ System.out.println(Thread.currentThread().getName()+"存款："+depositAmount); balance+=depositAmount; System.out.println("账户余额为："+balance); flag=true; cond.signalAll(); } }catch(InterruptedException ex){ ex.printStackTrace(); }finally{ lock.unlock(); } } }
其他的几个java文件不变，运行结果是一样的。
三 使用阻塞队列(BlockingQueue)控制线程通信
1.BlockingQueue是Queue的子接口，但它的主要用途并不是作为容器，而是作为线程同步的工具。它的一个特征是：当生产者线程试图向BlockingQueue中放入元素时，如果该队列已满，则该线程被阻塞；当消费者线程试图从BlockingQueue中取出元素时，如果该队列已空，则该线程被阻塞。程序的两个线程交替向BlockingQueue中放入、取出元素，即可很好的控制线程的通信。
2.提供了两个支持阻塞的方法：
put(E e):试图把E元素放入BlockingQueue中，如果队列的元素已满，则阻塞该线程；
take():尝试从BlockingQueue的头部取出元素，如果队列的元素已空，则阻塞该线程；
3.BlockingQueue继承了Queue接口，可以使用Queue接口中的方法，归纳起来分为三组：
(1)在队列尾部插入元素，包括add(E e),offer(E e),put(E e)，当队列已满时，分别会抛出异常、返回false、阻塞队列；
(2)在队列头部删除并返回删除的元素，包括remove(),poll()和take()，当队列已空时，分别会抛出异常、返回false、阻塞队列；
(3)在队列头部取出但不删除元素，包括element()和peek()方法，当队列已空时，着三个方法分别抛出异常、返回false
4.Java7之后又新增了一些阻塞队列，包含5个实现类：
(1)ArrayBlockingQueue:基于数组实现的BlockingQueue队列；
(2)LinkedBlockingQueue:基于链表实现的BlockingQueue队列；
(3)PriorityBlockingQueue:它并不是标准的阻塞队列，当调用remove()扥方法取出元素时，并不是取出队列中存在时间最长的元素，而是队列中最小的元素。判断元素的大小可根据元素(实现Comparable接口)的本身大小来自然排序，也可以使用Comparator进行定制排序。
(4)SynchronousQueue:同步队列，对该队列的存取操作必须交替进行；
(5)DelayQueue:它是一个特殊的BlockingQueue，底层基于PriorityBlockingQueue实现，并要求所有的集合元素都要实现Delay接口。
5.举例
public class BlockingQueueTest{ public static void main(String[] args) throws Exception{ BlockingQueue<String> bq=new ArrayBlockingQueue<String>(2); bq.put("java"); bq.put("study"); bq.put("javaee"); //阻塞线程 } }
四 线程池
1.系统启动一个新线程的成本是比较高的，因为涉及到与操作系统的交互，使用线程池可以很好的提高性能，尤其是当程序中需要创建大量生存期很短的线程时，更要考虑使用线程池。线程池在系统启动时就创建大量的空闲的线程，程序将一个Runnable对象或Callable对象传给线程池，线程池就会启动一个线程来执行它的run()或call()方法，当方法结束后，该线程并不会死亡，而是再次返回线程池中成为空闲状态，等待下一个Runnable对象的run()或call()方法。使用线程池还可以控制系统中并发线程的数量，当系统中包含大量并发线程时，会导致系统性能剧烈下降，甚至导致JVM崩溃。
2.Java5之后支持内建线程池，通过一个Executors工厂类来产生线程池，该工厂类包含以下几个静态工厂方法来创建线程池：
(1)newCachedThreadPool():创建一个具有缓存功能的线程池
(2)newFixedThreadPool( int nThreads):创建一个可重用的、具有固定线程数的线程池
(3)newSingleThreadPool():创建一个只有单线程的线程池
(4)newScheduledThreadPool(int corePoolSize):创建具有指定线程数的线程池，它可以在指定延迟后执行线程任务。
(6)newSingleThreadScheduledExecutor():创建只有一个线程的线程池，它可以在指定延迟后执行线程任务。
(7)ExecutorService newWorkStealingPool(int parallelism)：创建持有足够的线程的线程池来支持给定的并行级别，该方法还会使用多个队列来减少竞争
(8)ExecutorService newWorkStealingPool():上一个方法的简化版本，如果当前机器有4个CPU，则目标并行级别被设置为4，也就是相当于前一个方法传入4作为参数
注意：前三个方法返回一个ExecutorService对象，该对象代表一个线程池，可以执行Runnable或Callable对象所代表的线程；中间两个方法返回ScheduledExecutorService线程池，它可以在指定延迟后执行线程任务；后两个是Java8新增的，充分利用了多CPU并行的能力，这两个方法生成的work stealing池，相当于后台线程池，如果所有的后台线程都死亡了，work stealing池中的线程会自动死亡。
3.ExecutorService里提供了如下三个方法：
(1)Future<?> submit(Runnable task): 将一个Runnable对象提交给指定的线程池，线程池将在有空闲线程时执行Runnable对象代表的任务。这里的Future对象代表Runnable任务的返回值，但run()方法无返回值，所以在run()方法执行结束后返回null。但可以调用Future的isDone(),isCancelled()来获得Runnable对象的执行状态。
(2)<T> Future<T> submit(Runnable task, T result): 这里的result显式指定线程执行结束的返回值，所以Future对象将在run()方法执行结束后返回result
(3)<T> Future<T> submit(Callable<T> task): 这里提交的是Callable对象
4.ScheduledExecutorService提供了4个方法：
(1)ScheduledFuture<V> schedule(Callable<V> callable,long delay,TimeUnit unit):指定callable任务将在delay延迟后执行。
(2)ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit):指定command任务将在delay延迟后执行。
(3)ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit): 指定command任务将在delay延迟后执行，而且以设定频率重复执行，也就是说，在initialDelay后开始执行，一次在initialDelay+period,initialDelay+2*period...处重复执行。
(4)ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit):
创建并执行一个在给定初始延迟后首次启用的定期操作，随后在每一次执行终止和下一次执行开始之间都会存在给定的延迟。如果任务在某次执行时遇到异常，就会取消后续执行；否则，只会通过程序来显示取消或终止该任务。
5.用完一个线程池后，应该调用该线程池的shutdown()方法，该方法将启动线程池的关闭序列，线程池不再接收新任务，但会将所有已提交任务执行完成；也可以调用shutdownNow()方法来关闭线程池，该方法试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表。
6.举例
public class ThreadPoolTest { public static void main(String[] args) throws Exception{ ExecutorService pool= Executors.newFixedThreadPool(6); Runnable target= ()->{ for(int i=0;i<100;i++){ System.out.println(Thread.currentThread().getPriority()+"的i值为："+i); } }; //向线程池中提交两个线程 pool.submit(target) pool.submit(target); pool.shutdown(); } }
五 ThreadLocal类
1.java为ThreadLocal类增加了泛型支持，使用ThreadLocal类可以简化多线程编程时的并发访问，使用这个工具类可以很简捷的隔离多线程程序的竞争资源。
2.ThreadLocal，是Thread Local Variable (线程局部变量) 的意思，它的功能很简单，就是为每一个使用该变量的线程都提供一个变量值的副本，使每一个线程都可以独立的改变自己的副本，而不会与其他线程的副本冲突。从线程的角度看，就好像每一个线程都完全拥有该变量一样。
3.ThreadLocal类只提供了三个public方法：
T get():返回此线程局部变量中当前线程副本中的值
void remove():删除此线程局部变量中当前线程的值
void set(T value):设置此线程局部变量中当前线程副本中的值
4.ThreadLocalTest.java
class Account{ //定义一个ThreadLocal类型的变量，该变量将是一个线程局部变量，每个线程都会保留该变量的一个副本 private ThreadLocal<String> name=new ThreadLocal<>(); public Account(String str){ this.name.set(str); //用于访问当前线程的name的副本 System.out.println("---"+this.name.get()); } public String getName(){ return name.get(); } public void setName(String str){ this.name.set(str); } } class MyTest extends Thread{ private Account account; public MyTest(String name, Account account) { super(name); this.account = account; } public void run(){ for(int i=0;i<10;i++){ if(i==6) { account.setName(getName()); } System.out.println(account.getName()+"账户的i值："+i); } } } public class ThreadLocalTest { public static void main(String[] args){ //虽然两个线程共享同一个账户，即只有一个账户名，但由于账户名是ThreadLocal类型的，所以每个线程都完全拥有各自的账户名副本，因此在i==6之后，将看到两个线程访问同一个账户时出现不同的账户名 Account at=new Account("初始名"); new MyTest("线程甲",at).start(); new MyTest("线程乙",at).start(); } }
结果：
---初始名
null账户的i值：0
null账户的i值：0
null账户的i值：1
null账户的i值：1
null账户的i值：2
null账户的i值：2
null账户的i值：3
null账户的i值：3
null账户的i值：4
null账户的i值：4
null账户的i值：5
null账户的i值：5
线程甲账户的i值：6
线程乙账户的i值：6
线程甲账户的i值：7
线程乙账户的i值：7
线程甲账户的i值：8
线程乙账户的i值：8
线程甲账户的i值：9
线程乙账户的i值：9
实际上账户名有三个副本，主线程一个，另外启动的两个线程各一个，它们的值互不干扰，每个线程都完全拥有自己的ThreadLocal变量。ThreadLocal将需要并发访问的资源复制多份，每个线程拥有一份资源，每个线程都拥有自己的线程副本，从而也就没有必要对该变量进行同步了。ThreadLocal提供了线程安全的共享对象，在编写多线程代码时，可以把不安全的整个变量封装进ThreadLocal，或者把该对象与线程相关的状态使用ThreadLocal保存。
5.ThreadLocal并不能代替同步机制，两者面向的问题领域不同，同步机制是为了同步多个线程对相同资源的并发访问，是多个线程之间进行通信的有效方式，而ThreadLocal是为了隔离多个线程的数据共享，从根本上避免多个线程之间对共享资源的竞争。
线程通信