线程间通信之wait、notify

2019-07-02 本文已影响0人九点半的马拉

简介

Object.wait() -- 暂停一个线程
Object.notify() -- 唤醒一个线程
wait方法和notify方法并不是Thread特有的方法，而是Object中的方法。
wait方法介绍
wait方法必须拥有该对象的monitor,也就是wait方法必须在同步方法中使用。
当前线程执行了该对对象的wait方法之后，就会放弃对该monitor的所有权并进入与该对象关联的wait set中。
它会使当前执行wait()方法的线程等待，在wait()所在的代码行处暂停执行，并释放锁，直到接到通知或被中断为止。
notify方法介绍
唤醒单个正在执行该对象wait方法的线程
在调用前，线程必须获得锁
在执行notify()方法后，当前线程不会马上释放该锁，呈wait状态的线程也不会马上获取该对象锁，要等到执行notify()方法的线程将程序执行完，当前线程才会释放锁，而呈wait状态的线程才可以获取该对象锁

wait/notify机制简单实现

public class Test {
    public static void main(String[] args) {
        try {
            Object lock = new Object();
            Test test = new Test();
            Test.MyThread1 t1 = test.new MyThread1(lock);
            t1.start();
            Thread.sleep(2000);
            Test.MyThread2 t2 = test.new MyThread2(lock);
            t2.start();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }


    class MyThread1 extends Thread {
        private Object lock;
        public MyThread1(Object lock) {
            this.lock = lock;
        }

        @Override
        public void run() {
            try {
                synchronized (lock) {
                    System.out.println("开始 wait");
                    lock.wait();
                    System.out.println("结束 wait");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    class MyThread2 extends Thread {
        private Object lock;
        public MyThread2(Object lock) {
            this.lock = lock;
        }

        @Override
        public void run() {
            synchronized (lock) {
                System.out.println("开始 notify");
                lock.notify();
                System.out.println("结束 notify");
            }
        }

    }
}

输出结果：
开始 wait
开始 notify
结束 notify
结束 wait

简单例子

wait/notify模式最经典的案例就是生产者/消费者模式，

一生产与一消费：操作值

public class Test {
    public static String value = "";
    public static void main(String[] args) {
            Test test = new Test();
            String lock1 = "";
            P p = test.new P(lock1);
            C c = test.new C(lock1);
            Thread t1 = new Thread(new Runnable() {
                @Override
                public void run() {
                    for(int i=0; i<5; i++){
                        p.setValue();
                    }
                }
            });
            Thread t2 = new Thread(new Runnable() {
                @Override
                public void run() {
                    for(int i=0; i<5; i++){
                        c.getValue();
                    }
                }
            });
            t1.start();
            t2.start();
    }
    class P {
        private String lock;
        public P(String lock) {
            super();
            this.lock = lock;
        }
        public void setValue() {
            try {
                synchronized (lock) {
                    if(!value.equals("")) {
                        lock.wait();
                    }
                    String value1 = System.currentTimeMillis() + "_" + System.nanoTime();
                    System.out.println("set的值是" + value1);
                    value = value1;
                    lock.notify();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    class C {
        private String lock;
        public C(String lock) {
            super();
            this.lock = lock;
        }
        public void getValue() {
            try {
                synchronized (lock) {
                    if(value.equals("")) {
                        lock.wait();
                    }
                    System.out.println("get的值是： " + value);
                    value = "";
                    lock.notify();
                }

            }catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

运行结果：
set的值是1561896621397_10607393925500
get的值是： 1561896621397_10607393925500
set的值是1561896621397_10607394242800
get的值是： 1561896621397_10607394242800
set的值是1561896621397_10607394359900
get的值是： 1561896621397_10607394359900
set的值是1561896621397_10607394414900
get的值是： 1561896621397_10607394414900
set的值是1561896621397_10607394486700
get的值是： 1561896621397_10607394486700

当在多个生产者与多个消费者的情况下,操作值可能出现假死状态，即所有的线程都是waiting状态
在代码中进行wait/notify通信时，但不能保证notify唤醒的是异类，也许是同类，如“生产者”唤醒“生产者”，“消费者”唤醒“消费者”，慢慢的，大家都在等待，都呈waiting状态，程序最后就呈“假死”状态。

一生产与多消费（解决wait条件改变与假死）

import java.sql.SQLOutput;
import java.util.ArrayList;
import java.util.List;

public class Test {
    public static String value = "";
    public static void main(String[] args) {
            Test test = new Test();
            MyStack myStack = test.new MyStack();
            P p = test.new P(myStack);
            C c1 = test.new C(myStack);
            C c2 = test.new C(myStack);
            C c3 = test.new C(myStack);
            C c4 = test.new C(myStack);
            C c5 = test.new C(myStack);
            Thread t1 = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        p.pushService();
                    }
                }
            });
            Thread ct1 = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        c1.popService();
                    }
                }
            });
        Thread ct2 = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    c2.popService();
                }
            }
        });

        Thread ct3 = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    c3.popService();
                }
            }
        });

        Thread ct4 = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    c4.popService();
                }
            }
        });

        Thread ct5 = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    c5.popService();
                }
            }
        });

            t1.start();
            ct1.start();
            ct2.start();
            ct3.start();
            ct4.start();
            ct5.start();
    }
    class P {
        private MyStack myStack;
        public P(MyStack myStack) {
            super();
            this.myStack = myStack;
        }
        public void pushService() {
            myStack.push();
     }
    }

    class C {
        private MyStack myStack;
        public C(MyStack myStack) {
            super();
            this.myStack = myStack;
        }
        public void popService() {
            System.out.println("pop=" + myStack.pop());
        }
    }


    class MyStack{
        private List list = new ArrayList();
        synchronized public void push() {
            try{
                if (list.size() == 1) {
                    this.wait();
                }
                list.add("anyString=" + Math.random());
                this.notify();
                System.out.println("push=" + list.size());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        synchronized public String pop() {
            String returnValue = "";
            try {
                if (list.size() == 0) {
                    System.out.println("pop操作中的" + Thread.currentThread().getName() + "线程呈wait状态");
                    this.wait();
                }
                returnValue = "" + list.get(0);
                list.remove(0);
                this.notify();
                System.out.println("pop=" + list.size());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return returnValue;
        }
    }

}

运行结果：
push=1
pop=0
pop=anyString=0.22470100376922753
pop操作中的Thread-5线程呈wait状态
pop操作中的Thread-4线程呈wait状态
pop操作中的Thread-3线程呈wait状态
pop操作中的Thread-2线程呈wait状态
pop操作中的Thread-1线程呈wait状态
push=1
pop=0
pop=anyString=0.9533790755608161
pop操作中的Thread-5线程呈wait状态
Exception in thread "Thread-4" java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
at java.util.ArrayList.rangeCheck(ArrayList.java:657)
at java.util.ArrayList.get(ArrayList.java:433)
at Test $MyStack.pop(Test.java:119) at Test$ C.popService(Test.java:93)
at Test$5.run(Test.java:54)
at java.lang.Thread.run(Thread.java:748)

主要是pop()方法的if判断条件，因为刚一开始只往数组里放进了一个元素，然后相继执行了5个消费者线程，第一个成功删除，数组大小又重新变成了0，剩余的消费者线程变成了wait(),然后生产者又放入了一个元素，消费者再执行删除操作，并调用了notify方法，因为前面有消费者呈wait状态，所以被唤醒，执行删除操作，但此时的数组为空，所以会报错。
只需将pop()方法中的if变成while即可。

join方法的使用

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

public class Test {

    public static void main(String[] args) {
        Object oo = new Object();
        MyThread t1 = new MyThread("线程t1--", oo);
        MyThread t2 = new MyThread("线程t2--", oo);
        t2.start();
        t1.start();
        try {
            t1.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("结束");
    }

}

class MyThread extends Thread{
    private String name;
    private Object oo;
    public MyThread(String name,Object oo){
        this.name = name;
        this.oo = oo;
    }
    @Override
    public void run() {
        synchronized (oo) {
            for(int i = 0; i < 10; i++){
                System.out.println(name + i);
            }
        }
    }
}

运行结果：
线程t2--0
线程t2--1
线程t2--2
线程t2--3
线程t2--4
线程t2--5
线程t2--6
线程t2--7
线程t2--8
线程t2--9
线程t1--0
线程t1--1
线程t1--2
线程t1--3
线程t1--4
线程t1--5
线程t1--6
线程t1--7
线程t1--8
线程t1--9
结束

主线程main执行了t2.start()和t1.start()两行代码之后，创建了t2线程和t1线程，它们竞争oo这把锁，谁拿锁谁执行。首先，t2获得了该锁，t2执行完之后t1再开始执行，再从上一层次考虑的话，主线程main获得了t1这把锁。main线程继续执行了t1.join()方法，join()方法会使当前执行的线程等待 ，即让主线程main等待，主线程等到t1线程执行完成后，再继续执行。
通俗的讲就是：若线程A(main线程)调用线程B(t1线程)的join方法，那么线程A(main调用了t1.wait()被阻塞)的运行会被暂停，直到线程B(t1线程)运行结束。
实际上，调用join方法实际上调用了wait()方法。
x.join(long)中的参数用于设定等待的时间，不管x线程是否执行完毕，时间到了重新获得了锁，则当前线程会继续向后运行。如果没有重新获得锁，则一直在尝试，直到获得锁为止。

  public final synchronized void join(long millis)
    throws InterruptedException {
        long base = System.currentTimeMillis();
        long now = 0;

        if (millis < 0) {
            throw new IllegalArgumentException("timeout value is negative");
        }

        if (millis == 0) {
            while (isAlive()) {
                wait(0);
            }
        } else {
            while (isAlive()) {
                long delay = millis - now;
                if (delay <= 0) {
                    break;
                }
                wait(delay);
                now = System.currentTimeMillis() - base;
            }
        }
    }

join(long)方法与sleep(long)方法的区别

两种方法都可以使当前线程进入阻塞状态
当执行wait(long)方法时，会使当前执行的线程被释放，等其他线程执行完成后，该线程则会被唤醒
而Thread.sleep(long)方法却不释放锁，等休眠时间过后，会自动退出阻塞状态而重新恢复运行。

一道面试题
利用java的wait、notify机制实现启动两个线程, 一个输出 1,3,5,7…99, 另一个输出 2,4,6,8…100 最后 STDOUT 中按序输出 1,2,3,4,5…100

public class ThreadTest {
    private final Object flag = new Object();
    public static void main(String[] args) {
        ThreadTest threadTest = new ThreadTest();
        ThreadA threadA = threadTest.new ThreadA();
        threadA.start();
        ThreadB threadB = threadTest.new ThreadB();
        threadB.start();
    }

    class ThreadA extends Thread {
        @Override
        public void run() {
            synchronized (flag) {
                for (int i = 0; i <= 100; i += 2) {
                    flag.notify();
                    System.out.println(i);
                    try {
                        flag.wait();
                    } catch (InterruptedException e) {
                        System.out.println("error A");
                        e.printStackTrace();
                    }
                }
            }

        }
    }

    class ThreadB extends Thread {
        @Override
        public void run() {
            synchronized (flag) {
                for (int i = 1; i < 100; i += 2) {
                    flag.notify();
                    System.out.println(i);
                    try {
                        flag.wait();
                    } catch (InterruptedException e) {
                        System.out.println("error B");
                        e.printStackTrace();
                    }
                }
                flag.notify();
            }
        }
    }
}

线程间通信之wait、notify

简介

简单例子

join方法的使用

join(long)方法与sleep(long)方法的区别

猜你喜欢

热点阅读