第3章线程间通信

﹏ヽ暗。殇╰゛Y 2022-06-16 04:09 181阅读 0赞

# 第3章 线程间通信 #

标签： Java多线程编程

> 《[Java多线程编程核心技术][Java]》 个人笔记

--------------------

*  第3章 线程间通信
    
     *  等待通知机制
        
         *  不使用等待通知机制实现线程间通信
         *  什么是等待通知机制
         *  等待通知机制的实现
         *  方法wait锁释放与notify锁不释放
         *  当interrupt方法遇到wait方法
         *  只通知一个线程
         *  唤醒所有线程
         *  方法waitlong的使用
         *  通知过早
         *  等待wait的条件发生变化
         *  生产者消费者模式实现
         *  多生产与多消费操作值
         *  一生产与多消费操作栈解决wait条件改变与假死
         *  多生产与一消费操作栈
         *  多生产与多消费操作栈
         *  通过管道进行线程间通信字节流
         *  通过管道进行线程间通信字符流
         *  实战等待通知之交叉备份
     *  方法join的使用
        
         *  学习方法join前的铺垫
         *  用join方法类解决
         *  方法join与异常
         *  方法joinlong的使用
         *  方法joinlong和sleeplong的区别
         *  方法join后面的代码提前运行解释意外
     *  类ThreadLocal的使用
        
         *  方法get与null
         *  验证线程变量的隔离性
         *  解决get返回Null问题
         *  再次验证线程变量的隔离性
         *  类InhreitableThreadLocal的使用
         *  值继承再修改

--------------------

本章需要着重掌握的重点：  
1. 使用wait/notify实现线程间通信  
2. 生产者/消费者模式的实现  
3. 方法join的使用  
4. ThreadLocal类的使用

--------------------

## 等待/通知机制 ##

### 不使用等待/通知机制实现线程间通信 ###

### 什么是等待/通知机制 ###

### 等待/通知机制的实现 ###

*  方法wait()的作用是使当前执行代码的线程进行等待，wait()方法是Object类的方法，该方法用来将当前线程置入“预执行队列”中，并且在wait()所在的代码行处停止执行，直到接到通知或被中断为止。
 *  在调用wait()之前，线程必须获得该对象的对象级别锁，即只能在同步方法或同步块中调用wait()方法。
 *  在执行wait()方法后，当前线程释放锁 。在从wait()返回前，线程与其他线程竞争重新获得锁。
 *  如果调用wait()时没有持有适当的锁，则抛出IllegalMonitorStateException，它是RunTimeException的一个子类，因此，不需要try catch语句进行捕捉异常
 *  方法notify()也要在同步方法或同步块中调用，即在调用前，线程也必须获得该对象的对象级别锁。如果调用notify()时没有适当的锁，也会抛出IllegalMonitorStateException。
 *  该方法用来通知那些可能在等待该对象的对象锁的其他线程，如果有多个线程等待，则由线程规划器随机挑选其中**一个wait状态的线程**，对其发出通知notify，并使它等待获取该对象的对象锁。
 *  需要说明的是，在执行notify()方法后，当前线程不会马上释放该对象锁，呈wait状态的线程也不能马上获取该对象锁，要等到执行notify()方法的线程将程序执行完，也就是退出synchronized代码块后，当前线程才会释放，而呈wait状态所在的线程才可以获得该对象锁。
 *  用一句话总结：wait使线程停止运行，而notify使停止的线程继续运行
 *  notifyAll()方法可以使所有正在等待队列中等待同一共享资源的“全部”线程从等待状态退出，进入可运行状态。  
    ![此处输入图片的描述][20140828202610671]

### 方法wait()锁释放与notify()锁不释放 ###

### 当interrupt()方法遇到wait()方法 ###

*  当线程呈wait状态时，调用线程对象 的interrupt()方法会出现InterruptedException异常

public class Service { 
        public void testMethod(Object lock) {
            try {
                synchronized (lock) {
                    System.out.println("begin wait()");
                    lock.wait();        //wait状态
                    System.out.println(" end wait()");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                System.out.println("出现异常了，因为呈wait状态的线程被interrupt了！");
            }
        }
    }
    //------------线程类---------
    public class ThreadA extends Thread { 
        private Object lock;
        public ThreadA(Object lock) {
            super();
            this.lock = lock;
        }
        @Override
        public void run() {
            Service service = new Service();
            service.testMethod(lock);
        }
    }
    //-------------主函数-----------
    public class Test { 
        public static void main(String[] args) {
            try {
                Object lock = new Object();
                ThreadA a = new ThreadA(lock);
                a.start();
                Thread.sleep(5000);
                a.interrupt();          //调用中断
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    //--------结果----------
    begin wait()
    java.lang.InterruptedException
    出现异常了，因为呈wait状态的线程被interrupt了！
        at java.lang.Object.wait(Native Method)
        at java.lang.Object.wait(Unknown Source)
        at service.Service.testMethod(Service.java:9)
        at extthread.ThreadA.run(ThreadA.java:17)

*  结论：  
    
    
    1.  执行完同步代码块就会释放对象的锁
    2.  在执行同步代码块的过程中，遇到异常而导致线程终止，锁也会释放
    3.  在执行同步代码块的过程中，执行了锁所属对象的wait()方法，这个线程会释放对象锁，而此线程对象会进入线程等待池中，等待被唤醒

### 只通知一个线程 ###

*  调用notify()一次只随机通知一个线程

### 唤醒所有线程 ###

*  notifyAll()唤醒全部线程

### 方法wait(long)的使用 ###

*  带参数的wait(long)方法的功能是等待某一时间内是否有线程对锁进行唤醒，如果超过这个时间则自动唤醒

### 通知过早 ###

*  如果通知过早，则会打乱 程序正常的运行逻辑
 *  如果先通知了，则wait()方法也没有必要执行了，可以更改代码：

public class MyRun {
        private String lock = new String("");
        private boolean isFirstRunB = false;    //是否已经通知过
    
        private Runnable runnableA = new Runnable() {
            @Override
            public void run() {
                try {
                    synchronized (lock) {
                        while (isFirstRunB == false) {  //如果已经通知过，则不用wait
                            System.out.println("begin wait");
                            lock.wait();
                            System.out.println("end wait");
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        private Runnable runnableB = new Runnable() {
            @Override
            public void run() {
                synchronized (lock) {
                    System.out.println("begin notify");
                    lock.notify();      //通知
                    System.out.println("end notify");
                    isFirstRunB = true;
                }
            }
        };
    //-----------主函数------------
        public static void main(String[] args) throws InterruptedException {
            MyRun run = new MyRun();
            Thread a = new Thread(run.runnableA);   //先wait
            a.start();
            Thread.sleep(100);
            Thread b = new Thread(run.runnableB);  //再通知
            b.start();
        }
    }
    //--------------运行结果--------------
    begin wait
    begin notify
    end notify
    end wait
    //---------------主函数------------
    public static void main(String[] args) throws InterruptedException {
            MyRun run = new MyRun();
            Thread b = new Thread(run.runnableB);   //先通知
            b.start();
            Thread.sleep(100);
            Thread a = new Thread(run.runnableA);   //再wait
            a.start();
        }
    //--------------运行结果--------------
    begin notify
    end notify

### 等待wait的条件发生变化 ###

*  在使用wait/notify模式时，还需要注意另外一种情况，也就是wait等待的条件发生了变化，也容易造成程序逻辑的混乱

public void subtract() {
            try {
                synchronized (lock) {
                    while (ValueObject.list.size() == 0) {  //用while循环，而不是if判断
                        System.out.println("wait begin ThreadName="
                                + Thread.currentThread().getName());
                        lock.wait();
                        System.out.println("wait end ThreadName="
                                + Thread.currentThread().getName());
                    }
                    ValueObject.list.remove(0);
                    System.out.println("list size=" + ValueObject.list.size());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

*  用while循环，而不是if判断，当多个线程执行这段代码时，当一个被唤醒，会继续循环判断当前list.size()是否为0，若为0则继续等待，防止list被其他线程清空了导致list.remove(0)报异常

### 生产者/消费者模式实现 ###

1.  一生产者与一消费者
2.  多生产与多消费：操作值假死  
    
    
     *  ‘假死’的现象其实就是线程进入WAITING等待状态。如果全部线程都进入WAITING状态，则程序就不再执行任务业务功能了，整个项目呈停止状态。这在生产者与消费者模式时经常遇到。
     *  假死的原因：notify()方法是随机唤醒的，当连续唤醒了同类时，譬如：有两个生产者，两个消费者，当第一个生产者唤醒另一个生产者，则这时仍然有生产未被消费，所以两个生产者都进入wait状态，接着，第一个消费者消费了生产的东西，唤醒另一个消费者，进入wait状态，此时被唤醒的消费者又没有生产可以消费，进入wait状态，全部线程都处于wait状态

//生产的资源类
    public class ValueObject { 
        public static String value = "";
    }
    //--------生产者------
    public class P { 
        private String lock;    //锁
        public P(String lock) {
            super();
            this.lock = lock;
        }
        public void setValue() {
            try {
                synchronized (lock) {
                    while (!ValueObject.value.equals("")) {
       //如果资源非空，等待
                        System.out.println("生产者 "
                                + Thread.currentThread().getName() + " WAITING了★");
                        lock.wait();
                    }
                    System.out.println("生产者 " + Thread.currentThread().getName()
                            + " RUNNABLE了");
                    String value = System.currentTimeMillis() + "_"
                            + System.nanoTime();
                    ValueObject.value = value;  //生产
                    lock.notify();  //生产后再通知
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    //------------消费者----------
    public class C { 
        private String lock;
        public C(String lock) {
            super();
            this.lock = lock;
        }
        public void getValue() {
            try {
                synchronized (lock) {
                    while (ValueObject.value.equals("")) { //没有资源可消费，等待
                        System.out.println("消费者 "
                                + Thread.currentThread().getName() + " WAITING了☆");
                        lock.wait();
                    }
                    System.out.println("消费者 " + Thread.currentThread().getName()
                            + " RUNNABLE了");
                    ValueObject.value = "";     //消费
                    lock.notify();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    //----------生产者线程-------
    public class ThreadP extends Thread { 
        private P p;
        public ThreadP(P p) {
            super();
            this.p = p;
        }
        @Override
        public void run() {
            while (true) {
                p.setValue();
            }
        }
    }
    //----------消费者线程-----------
    public class ThreadC extends Thread { 
        private C r;
        public ThreadC(C r) {
            super();
            this.r = r;
        }
        @Override
        public void run() {
            while (true) {
                r.getValue();
            }
        }
    }
    //-------------主函数----------
    public class Run { 
    
        public static void main(String[] args) throws InterruptedException {
            String lock = new String("");
            P p = new P(lock);
            C r = new C(lock);
            ThreadP[] pThread = new ThreadP[2];
            ThreadC[] rThread = new ThreadC[2];
            for (int i = 0; i < 2; i++) {
                pThread[i] = new ThreadP(p);
                pThread[i].setName("生产者" + (i + 1));
    
                rThread[i] = new ThreadC(r);
                rThread[i].setName("消费者" + (i + 1));
    
                pThread[i].start();
                rThread[i].start();
            }
            Thread.sleep(5000);
            Thread[] threadArray = new Thread[Thread.currentThread()
                    .getThreadGroup().activeCount()];
            Thread.currentThread().getThreadGroup().enumerate(threadArray);
    
            for (int i = 0; i < threadArray.length; i++) {
                System.out.println(threadArray[i].getName() + " "
                        + threadArray[i].getState());
            }
        }
    }

### 多生产与多消费：操作值 ###

*  解决“假死”很简单，将生产者和消费者的notify()改成notifyAll()，即不光通知同类线程，也通知异类线程。

### 一生产与多消费————操作栈：解决wait条件改变与假死 ###

### 多生产与一消费————操作栈 ###

### 多生产与多消费————操作栈 ###

### 通过管道进行线程间通信：字节流 ###

*  管道流是一种特殊的流，用于在不同线程间直接传送数据。一个线程发送数据到输出管道，另一个线程从输入管道中读取数据。通过使用管道，实现不同线程间的通信，而无需借助类似临时文件安之类的东西。
 *  JDK中提供了4各类来使线程间可以进行通信：  
    
    
    1.  PipedInputStream和PipedOutputStream
    2.  PipedReader和PipedWriter

public class ReadData { 
        public void readMethod(PipedInputStream input) {
            try {
                System.out.println("read :");
                byte[] byteArray = new byte[20];
                int readLength = input.read(byteArray);
                while (readLength != -1) {
                    String newData = new String(byteArray, 0, readLength);
                    System.out.print(newData);
                    readLength = input.read(byteArray);
                }
                System.out.println();
                input.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public class WriteData { 
        public void writeMethod(PipedOutputStream out) {
            try {
                System.out.println("write :");
                for (int i = 0; i < 30; i++) {
                    String outData = " " + (i + 1);
                    out.write(outData.getBytes());
                    System.out.print(outData);
                }
                System.out.println();
                out.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    // --------- 两个线程类------
    public class ThreadRead extends Thread { 
        private ReadData read;
        private PipedInputStream input;
    
        public ThreadRead(ReadData read, PipedInputStream input) {
            super();
            this.read = read;
            this.input = input;
        }
        @Override
        public void run() {
            read.readMethod(input);
        }
    }
    public class ThreadWrite extends Thread { 
        private WriteData write;
        private PipedOutputStream out;
    
        public ThreadWrite(WriteData write, PipedOutputStream out) {
            super();
            this.write = write;
            this.out = out;
        }
        @Override
        public void run() {
            write.writeMethod(out);
        }
    }
    //---------主函数----------
    public class Run { 
        public static void main(String[] args) {
            try {
                WriteData writeData = new WriteData();
                ReadData readData = new ReadData();
                PipedInputStream inputStream = new PipedInputStream();
                PipedOutputStream outputStream = new PipedOutputStream();
                // inputStream.connect(outputStream);
                outputStream.connect(inputStream);  //管道连接
    
                ThreadRead threadRead = new ThreadRead(readData, inputStream);
                threadRead.start();
                Thread.sleep(2000);
                ThreadWrite threadWrite = new ThreadWrite(writeData, outputStream);
                threadWrite.start();    //开始写入
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    //-------------运行结果------------
    read  :
    write :
     1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30
     1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30

### 通过管道进行线程间通信：字符流 ###

*  与字节流一样，只是把字节流管道类换成字符流管道类

public class ReadData {
        public void readMethod(PipedReader input) {
            try {
                System.out.println("read :");
                char[] byteArray = new char[20];
                int readLength = input.read(byteArray);
                while (readLength != -1) {
                    String newData = new String(byteArray, 0, readLength);
                    System.out.print(newData);
                    readLength = input.read(byteArray);
                }
                System.out.println();
                input.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public class WriteData {
        public void writeMethod(PipedWriter out) {
            try {
                System.out.println("write :");
                for (int i = 0; i < 30; i++) {
                    String outData = " " + (i + 1);
                    out.write(outData);
                    System.out.print(outData);
                }
                System.out.println();
                out.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

### 实战：等待/通知之交叉备份 ###

*  本节的目的是要继续学习等待/通知相关知识点，创建20个线程，其中10个线程是将数据备份到A数据库中，另外10个线程将数据备份到B数据库中，并且备份A数据库和B数据库是交叉进行的

public class DBTools {
    
        volatile private boolean prevIsA = false;       //公共标记
    
        synchronized public void backupA() {
            try {
                while (prevIsA == true) {
                    wait();
                }
                for (int i = 0; i < 5; i++) {
                    System.out.println("AAAAA");
                }
                prevIsA = true;
                notifyAll();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        synchronized public void backupB() {
            try {
                while (prevIsA == false) {
                    wait();
                }
                for (int i = 0; i < 5; i++) {
                    System.out.println("BBBBB");
                }
                prevIsA = false;
                notifyAll();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
        public static void main(String[] args) {
            DBTools dbtools = new DBTools();
            for (int i = 0; i < 5; i++) {
                BackupB output = new BackupB(dbtools);
                output.start();
                BackupA input = new BackupA(dbtools);
                input.start();
            }
        }

## 方法join的使用 ##

### 学习方法join前的铺垫 ###

### 用join()方法类解决 ###

*  方法join()的作用是使所属的线程对象x正常执行run()方法中的任务，而使当前线程z进行无限期的阻塞，等待线程x销毁后再继续执行线程z后面的代码
 *  方法join()具有使线程排队运行的作用，有些类似同步的运行效果。join与synchronized的区别是：join在内部使用wait()方法进行等待，而synchronized使用的是“对象监视器”原理作为同步

public class MyThread extends Thread { 
        @Override
        public void run() {
            try {
                int secondValue = (int) (Math.random() * 10000);
                System.out.println(secondValue);
                Thread.sleep(secondValue);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    //---------主函数---------
    public class Test { 
        public static void main(String[] args) {
            try {
                MyThread threadTest = new MyThread();
                threadTest.start();
                threadTest.join();   //此处当前线程被阻塞，知道threadTest执行完毕
                System.out.println("我想当threadTest对象执行完后再执行！");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    //-------运行结果----------
    2524
    我想当threadTest对象执行完后再执行！

### 方法join与异常 ###

*  在join过程中，如果当前线程对象被中断，则当前线程出现异常，但子线程仍然正常运行

### 方法join(long)的使用 ###

*  方法join(long)中的参数是设定等待的时间

### 方法join(long)和sleep(long)的区别 ###

*  方法join(long)的功能在内部是使用wait(long)方法来实现的，所以join(long)方法具有释放锁的特点。当执行完wait(long)方法后，当前线程的锁被释放，那么其他线程就可以调用此线程中的同步方法了。
 *  Thread.sleep(long)方法却不释放锁。

### 方法join()后面的代码提前运行：解释意外 ###

## 类ThreadLocal的使用 ##

*  类ThreadLocal主要解决的就是每个线程绑定自己的值，可以将ThreadLocal类比喻呈全局存放数据的盒子，盒子中可以存储每个线程的私有数据

### 方法get()与null ###

### 验证线程变量的隔离性 ###

public class Tools { 
        public static ThreadLocal tl = new ThreadLocal();
    }
    public class ThreadA extends Thread { 
        @Override
        public void run() {
            try {
                for (int i = 0; i < 100; i++) {
                    if (Tools.tl.get() == null) {
                        Tools.tl.set("ThreadA" + (i + 1));
                    } else {
                        System.out.println("ThreadA get Value=" + Tools.tl.get());
                    }
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public class ThreadB extends Thread { 
        @Override
        public void run() {
            try {
                for (int i = 0; i < 100; i++) {
                    if (Tools.tl.get() == null) {
                        Tools.tl.set("ThreadB" + (i + 1));
                    } else {
                        System.out.println("ThreadB get Value=" + Tools.tl.get());
                    }
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public class Run { 
        public static void main(String[] args) {
            try {
                ThreadA a = new ThreadA();
                ThreadB b = new ThreadB();
                a.start();
                b.start();
    
                for (int i = 0; i < 100; i++) {
                    if (Tools.tl.get() == null) {
                        Tools.tl.set("Main" + (i + 1));
                    } else {
                        System.out.println("Main get Value=" + Tools.tl.get());
                    }
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

*  虽然3个线程都向tl对象中set()数据值，但每个线程还是能取出自己的数据

### 解决get()返回Null问题 ###

*  默认值

public class ThreadLocalExt extends ThreadLocal { 
        @Override
        protected Object initialValue() {
            return "我是默认是，第一次 get 不再为 null";
        }
    }

### 再次验证线程变量的隔离性 ###

public class ThreadLocalExt extends ThreadLocal { 
        @Override
        protected Object initialValue() {
            return new Date().getTime();    //每个线程的初始值都不同
        }
    }

*  子线程和父线程各自有各自的值

### 类InhreitableThreadLocal的使用 ###

*  使用类InhreitableThreadLocal可以子线程中取得父线程继承下来的值

### 值继承再修改 ###

*  在使用InhreitableThreadLocal类需要注意一点的是，如果子线程在取得值的同时，主线程将InhreitableThreadLocal中的值进行修改，那么子线程取到的值还是旧值

[Java]: http://product.dangdang.com/23711315.html
[20140828202610671]: /images/20220616/505d4075b809445f8607c3eb526b7731.png