Java 线程通信线程组线程异常处理机制

柔情只为你懂 2022-02-23 12:16 314阅读 0赞

## 线程通信 ##

### 传统的线程通信 ###

方法概述：

*  wait方法：导致当前线程等待，直到其他线程调用该同步监视器的notify()方法或notifyAll()方法来唤醒该线程。
    
        wait()方法有三种形式——无时间参数的wait()方法（一直等待，直到其他线程通知）；
        带毫秒参数的wait()方法、带毫秒、毫微妙参数的wait()方法，这2种方法都是等待指定时间后自动苏醒
        调用wait()方法的当前线程会释放对该同步监视器的锁定
 *  notify：唤醒在此同步监视器上等待的单个线程。如果所有线程都在此同步监视器上等待，则会随机选择唤醒其中一个线程。只有当前线程放弃对该同步监视器的锁定后（用wait()方法），才可以执行被唤醒的线程
 *  notifyAll：唤醒在此同步监视器上等待的所有线程。只有当前线程放弃对该同步监视器的锁定后，才能执行唤醒的线程

这三个方法属于Object类，必须由同步监视器对象来调用，可分成以下两种情况：

*  对于使用synchronize修饰的同步方法，因为该类的默认实例（this）就是同步监视器，所以可以在同步方法中直接调用这三个方法
 *  对于使用synchronized修改的同步代码块，同步监视器是synchronized后可括号中的对象，所以必须使用括号中的对象调用这3个方法

public class Account
    {
        // 封装账户编号、账户余额的两个成员变量
        private String accountNo;
        private double balance;
        // 标识账户中是否已有存款的旗标
        private boolean flag = false;
    
        public Account(){}
        // 构造器
        public Account(String accountNo , double balance)
        {
            this.accountNo = accountNo;
            this.balance = balance;
        }
    
        // accountNo的setter和getter方法
        public void setAccountNo(String accountNo)
        {
            this.accountNo = accountNo;
        }
        public String getAccountNo()
        {
            return this.accountNo;
        }
        // 因此账户余额不允许随便修改，所以只为balance提供getter方法，
        public double getBalance()
        {
            return this.balance;
        }
    
        public synchronized void draw(double drawAmount)
        {
            try
            {
                // 如果flag为假，表明账户中还没有人存钱进去，取钱方法阻塞
                if (!flag)
                {
                    wait();
                }
                else
                {
                    // 执行取钱
                    System.out.println(Thread.currentThread().getName()
                        + " 取钱:" +  drawAmount);
                    balance -= drawAmount;
                    System.out.println("账户余额为：" + balance);
                    // 将标识账户是否已有存款的旗标设为false。
                    flag = false;
                    // 唤醒其他线程
                    notifyAll();
                }
            }
            catch (InterruptedException ex)
            {
                ex.printStackTrace();
            }
        }
        public synchronized void deposit(double depositAmount)
        {
            try
            {
                // 如果flag为真，表明账户中已有人存钱进去，则存钱方法阻塞
                if (flag)             //①
                {
                    wait();
                }
                else
                {
                    // 执行存款
                    System.out.println(Thread.currentThread().getName()
                        + " 存款:" +  depositAmount);
                    balance += depositAmount;
                    System.out.println("账户余额为：" + balance);
                    // 将表示账户是否已有存款的旗标设为true
                    flag = true;
                    // 唤醒其他线程
                    notifyAll();
                }
            }
            catch (InterruptedException ex)
            {
                ex.printStackTrace();
            }
        }
    
        // 下面两个方法根据accountNo来重写hashCode()和equals()方法
        public int hashCode()
        {
            return accountNo.hashCode();
        }
        public boolean equals(Object obj)
        {
            if(this == obj)
                return true;
            if (obj !=null && obj.getClass() == Account.class)
            {
                Account target = (Account)obj;
                return target.getAccountNo().equals(accountNo);
            }
            return false;
        }
    }

### 使用Condition控制线程通信 ###

直接使用Lock对象来保证同步，则系统中不存在隐式的同步监视器，不能使用wait()、notify()、notifyAll()方法进行线程通信

当使用Lock对象来保证同步时，Java提供一个Condition类来保持协调，使用Condition可以让那些已经得到Lock对象却无法继续执行的线程释放Lock对象，Condition对象也可以唤醒其他处于等待的线程

Condition将同步监视器方法（wait、notify、notifyAll）分解成截然不同的对象，以便通过将这些对象与Lock对象组合使用，为每个对象提供了多个等待集（wait-set），这种情况下，Lock替代了同步方法和同步代码块，Condition替代同步监视器的功能

Condition实例被绑定在一个Lock对象上，要获得特定的Lock实例的Condition实例，调用Lock对象的newCondition()即可

Condition类方法介绍：

*  await()：类似于隐式同步监视器上的wait方法，导致当前程序等待，直到其他线程调用该Condition的signal()方法和signalAll()方法来唤醒该线程。该await方法有跟多获取变体：long awaitNanos(long nanosTimeout)、void awaitUninterruptibly()、awaitUntil(Date daadline)等
 *  signal()：唤醒在此Lock对象上等待的单个线程，如果所有的线程都在该Lock对象上等待，则会选择随机唤醒其中一个线程。只有当前线程放弃对该Lock对象的锁定后（使用await()方法），才可以唤醒在执行的线程
 *  signalAll()：唤醒在此Lock对象上等待的所有线程。只有当前线程放弃对该Lock对象的锁定后，才可以执行被唤醒的线程

import java.util.concurrent.*;
    import java.util.concurrent.locks.*;
    public class Account
    {
        // 显式定义Lock对象
        private final Lock lock = new ReentrantLock();
        // 获得指定Lock对象对应的Condition
        private final Condition cond  = lock.newCondition();
        // 封装账户编号、账户余额的两个成员变量
        private String accountNo;
        private double balance;
        // 标识账户中是否已有存款的旗标
        private boolean flag = false;
    
        public Account(){}
        // 构造器
        public Account(String accountNo , double balance)
        {
            this.accountNo = accountNo;
            this.balance = balance;
        }
    
        // accountNo的setter和getter方法
        public void setAccountNo(String accountNo)
        {
            this.accountNo = accountNo;
        }
        public String getAccountNo()
        {
            return this.accountNo;
        }
        // 因此账户余额不允许随便修改，所以只为balance提供getter方法，
        public double getBalance()
        {
            return this.balance;
        }
    
        public void draw(double drawAmount)
        {
            // 加锁
            lock.lock();
            try
            {
                // 如果flag为假，表明账户中还没有人存钱进去，取钱方法阻塞
                if (!flag)
                {
                    cond.await();
                }
                else
                {
                    // 执行取钱
                    System.out.println(Thread.currentThread().getName() + " 取钱:" +  drawAmount);
                    balance -= drawAmount;
                    System.out.println("账户余额为：" + balance);
                    // 将标识账户是否已有存款的旗标设为false。
                    flag = false;
                    // 唤醒其他线程
                    cond.signalAll();
                }
            }
            catch (InterruptedException ex)
            {
                ex.printStackTrace();
            }
            // 使用finally块来释放锁
            finally
            {
                lock.unlock();
            }
        }
        public void deposit(double depositAmount)
        {
            lock.lock();
            try
            {
                // 如果flag为真，表明账户中已有人存钱进去，则存钱方法阻塞
                if (flag)             // ①
                {
                    cond.await();
                }
                else
                {
                    // 执行存款
                    System.out.println(Thread.currentThread().getName() + " 存款:" +  depositAmount);
                    balance += depositAmount;
                    System.out.println("账户余额为：" + balance);
                    // 将表示账户是否已有存款的旗标设为true
                    flag = true;
                    // 唤醒其他线程
                    cond.signalAll();
                }
            }
            catch (InterruptedException ex)
            {
                ex.printStackTrace();
            }
            // 使用finally块来释放锁
            finally
            {
                lock.unlock();
            }
        }
    
        // 下面两个方法根据accountNo来重写hashCode()和equals()方法
        public int hashCode()
        {
            return accountNo.hashCode();
        }
        public boolean equals(Object obj)
        {
            if(this == obj)
                return true;
            if (obj !=null
                && obj.getClass() == Account.class)
            {
                Account target = (Account)obj;
                return target.getAccountNo().equals(accountNo);
            }
            return false;
        }
    }

### 使用阻塞队列（BlockingQueue）控制线程通信 ###

BlockingQueue具有一个特征：当生产者线程试图向BlockingQueue中放入元素时，如果该队列已满，则线程被阻塞；但消费者线程试图从BlockingQueue中取出元素时，如果队列已空，则该线程阻塞

程序的两个线程通过交替向BlockingQueue中放入元素、取出元素，即可很好地控制线程的通信

BlockingQueue提供如下两个支持阻塞的方法：

*  put(E e)：尝试把E元素放入BlockingQueue中，如果该队列的元素已满，则阻塞该线程
 *  take()：尝试从BlockingQueue的头部取出元素，如果该队列的元素已空，则阻塞该线程

BlockingQueue继承了Queue接口，当然也可以使用Queue接口中的方法，这些方法归纳起来可以分为如下三组：

*  在队列尾部插入元素，包括add(E e)、offer(E e)和put(E e)方法，当该队列已满时，这三个方法分别会抛出异常、返回false、阻塞队列
 *  在队列头部删除并返回删除的元素。包括remove()、poll()和take()方法，当该队列已空时，这三个方法分别会抛出异常、返回false、阻塞队列
 *  在队列头部取出但不删除元素。包括element()和peek()方法，当队列已空时，这两个方法分别抛出异常、返回false

<table> 
 <thead> 
  <tr> 
   <th>-</th> 
   <th>抛出异常</th> 
   <th>不同返回值</th> 
   <th>阻塞线程</th> 
   <th>指定超时时差</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>队尾插入元素</td> 
   <td>add(e)</td> 
   <td>offer(e)</td> 
   <td>put(e)</td> 
   <td>offer(e, time, unit)</td> 
  </tr> 
  <tr> 
   <td>队头删除元素</td> 
   <td>remove()</td> 
   <td>poll()</td> 
   <td>take()</td> 
   <td>poll(time, unit)</td> 
  </tr> 
  <tr> 
   <td>获取、不删除元素</td> 
   <td>element()</td> 
   <td>peek()</td> 
   <td>无</td> 
   <td>无</td> 
  </tr> 
 </tbody> 
</table>

![clipboard.png][]

BlockingQueue的5个实现类：

*  ArrayBlockingQueue：基于数组实现的BlockingQueue队列
 *  LinkedBlockingQueue：基于链表实现的BlockingQueue队列
 *  PriorityBlockingQueue：它并不是标准的阻塞队列，当调用remove()、poll()、take()等方法取出元素时，并不是取出队列中存在时间最长的元素，而是队列中最小的元素。判断元素的大小可根据元素(实现Comparable接口)的本身大小来自然排序，也可以使用Comparator进行定制排序
 *  SynchronousQueue：同步队列，对该队列的存、取操作必须交替进行
 *  DelayQueue：它是一个特殊的BlockingQueue，底层基于PriorityBlockingQueue实现，DelayQueue要求集合元素都实现Delay接口（该接口里只有一个long getDelay()方法），DelayQueue根据集合元素的getDelay()方法的返回值进行排序

import java.util.concurrent.*;
    
    public class BlockingQueueTest
    {
        public static void main(String[] args) throws Exception
        {
            // 定义一个长度为2的阻塞队列
            BlockingQueue<String> bq = new ArrayBlockingQueue<>(2);
            bq.put("Java"); // 与bq.add("Java"、bq.offer("Java")相同
            bq.put("Java"); // 与bq.add("Java"、bq.offer("Java")相同
            bq.put("Java"); // ① 阻塞线程。
        }
    }

利用BlockingQueue实现线程通信

import java.util.concurrent.*;
    
    class Producer extends Thread
    {
        private BlockingQueue<String> bq;
        public Producer(BlockingQueue<String> bq)
        {
            this.bq = bq;
        }
        public void run()
        {
            String[] strArr = new String[]
            {
                "Java",
                "Struts",
                "Spring"
            };
            for (int i = 0 ; i < 999999999 ; i++ )
            {
                System.out.println(getName() + "生产者准备生产集合元素！");
                try
                {
                    Thread.sleep(200);
                    // 尝试放入元素，如果队列已满，线程被阻塞
                    bq.put(strArr[i % 3]);
                }
                catch (Exception ex){ex.printStackTrace();}
                System.out.println(getName() + "生产完成：" + bq);
            }
        }
    }
    class Consumer extends Thread
    {
        private BlockingQueue<String> bq;
        public Consumer(BlockingQueue<String> bq)
        {
            this.bq = bq;
        }
        public void run()
        {
            while(true)
            {
                System.out.println(getName() + "消费者准备消费集合元素！");
                try
                {
                    Thread.sleep(200);
                    // 尝试取出元素，如果队列已空，线程被阻塞
                    bq.take();
                }
                catch (Exception ex){ex.printStackTrace();}
                System.out.println(getName() + "消费完成：" + bq);
            }
        }
    }
    public class BlockingQueueTest2
    {
        public static void main(String[] args)
        {
            // 创建一个容量为1的BlockingQueue
            BlockingQueue<String> bq = new ArrayBlockingQueue<>(1);
            // 启动3条生产者线程
            new Producer(bq).start();
            new Producer(bq).start();
            new Producer(bq).start();
            // 启动一条消费者线程
            new Consumer(bq).start();
        }
    }

程序启动3个生产者线程向BlockingQueue集合放入元素，启动1个消费者线程从BlockingQueue集合取出元素。本程序的BlockingQueue集合容量为1，因此3个生产者线程无法连续放入元素，必须等待消费者线程取出一个元素后，3个生产者线程的其中一个才能放入元素。运行结果如下：  
![clipboard.png][clipboard.png 1]

## 线程组和未处理的异常 ##

ThreadGroup表示线程组，可以对一批线程进行分类管理。对线程组的控制相当于同时控制这批线程。默认情况下，子线程和创建它的父线程属于同一个线程组

一旦某个线程加入了指定线程组之后，该线程将一直属于该线程组，直到该线程死亡，线程运行中不能改变它所属的线程组

Thread类提供了如下几个构造器来设置新创建的线程属于哪个线程组：

*  Thread(ThreadGroup group, Runnable target)：以target的run()方法作为线程执行体创建新线程，属于group线程组
 *  Thread(ThreadGroup group, Runnable target, String name)：以target的run()方法作为线程执行体创建新线程，属于group线程组，且线程名为name
 *  Thread(ThreadGroup group, String name)：创建新线程，新线程名为name，属于group线程组

因为中途不可改变线程所属的线程组，所以Thread类没有setThreadGroup()方法，但提供getThreadGroup()方法来返回该线程所属的线程组，返回值是ThreadGroup对象

ThreadGroup类的两个构造器：

*  ThreadGroup(String name)：以指定的线程组名字来创建新的线程组
 *  ThreadGroup(ThreadGroup parent, String name)：以指定的名字、指定的父线程组创建新的线程组

ThreadGroup类操作整个线程组里的所有线程的几个常用方法：

*  int activeCount()：返回此线程组中活动线程的数目
 *  interrupt()：中断此线程组中的所有线程
 *  isDaemon()：判断该线程组是否是后台线程组
 *  setDaemon(boolean daemon)：把该线程组设置成后台线程组。后台线程组具有一个特征，当后台线程组的最后一个线程执行结束或最后一个线程被销毁后，后台线程组将自动销毁
 *  setMaxPriority(int pri)：设置线程组的最高优先级

class MyThread extends Thread
    {
        // 提供指定线程名的构造器
        public MyThread(String name)
        {
            super(name);
        }
        // 提供指定线程名、线程组的构造器
        public MyThread(ThreadGroup group, String name)
        {
            super(group, name);
        }
        public void run()
        {
            for (int i = 0; i < 20 ; i++ )
            {
                System.out.println(getName() + " 线程的i变量" + i);
            }
        }
    }
    public class ThreadGroupTest
    {
        public static void main(String[] args)
        {
            // 获取主线程所在的线程组，这是所有线程默认的线程组
            ThreadGroup mainGroup = Thread.currentThread().getThreadGroup();
            System.out.println("主线程组的名字：" + mainGroup.getName());
            System.out.println("主线程组是否是后台线程组：" + mainGroup.isDaemon());
            new MyThread("主线程组的线程").start();
            ThreadGroup tg = new ThreadGroup("新线程组");
            tg.setDaemon(true);
            System.out.println("tg线程组是否是后台线程组：" + tg.isDaemon());
            MyThread tt = new MyThread(tg, "tg组的线程甲");
            tt.start();
            new MyThread(tg, "tg组的线程乙").start();
        }
    }

void uncaughtException(Thread t, Throwable e)：该方法可以处理该线程组内的任意线程所抛出的未处理异常

JVM在结束线程前会自动查找是否有对应的Thread.UncaughtExceptionHandler对象(该类为是Thread的一个静态内部接口)，如果找到该处理器对象，会调用该对象的uncaughtException(Thread t, Throwable e)来处理该异常。t:表示出现异常的线程，e表示该线程抛出的异常

Thread类提供两个方法设置异常处理器：

*  static setDefaultUncaughtExceptionHandler(Thread.UncaughtExceptionHanlder eh)：为该线程类的所有线程实例设置默认的异常处理器
 *  static setUncaughtExceptionHandler(Thread.UncaughtExceptionHanlder eh)：为指定的线程实例设置异常处理器

线程组处理异常的默认流程：

*  如果该线程组有父线程组，则调用父线程组的uncaughtException()来处理异常
 *  如果该线程对象所属线程类有有默认异常处理器（由setDefaultUncaughtExceptionHandler()方法设置的异常处理器），那么就调用该异常处理器来处理该异常
 *  如果该异常对象是ThreadDeath的对象，则不做任何处理；否则，将异常跟踪栈的信息打印到System.err错误输出流，并结束该线程

class MyExHandler implements Thread.UncaughtExceptionHandler
    {
        // 实现uncaughtException方法，该方法将处理线程的未处理异常
        public void uncaughtException(Thread t, Throwable e)
        {
            System.out.println(t + " 线程出现了异常：" + e);
        }
    }
    public class ExHandler
    {
        public static void main(String[] args)
        {
            // 设置主线程的异常处理器
            Thread.currentThread().setUncaughtExceptionHandler(new MyExHandler());
            int a = 23 / 0;     // ①
            System.out.println("程序正常结束！");
        }
    }

[clipboard.png]: /images/20220223/f5eaaba602374a98900396244a986f55.png
[clipboard.png 1]: /images/20220223/d45fdd33d40d4539b39f32a453253864.png