多线程-线程间通信

1，等待/通知机制

线程通信：线程通信就是当多个线程共同操作共享的资源时，互相告知自己的状态以避免资源争夺。

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

例：现在有A，B两个线程，A线程负责将数据添加进集合中，B线程通过不断判断，当集合中的元素的个数达到5个的时候，B线程将通过异常结束。此时会出现一个问题，就是B线程通过不断while循环来检测一个条件会特别浪费CPU的资源；如果轮询的时间间隔太大的话，可能会导致其他线程得不到其想要的数据。为了解决这个问题，我们就可以使用wait/notify来实现线程之间的通信。

class MyList{
    private List list = new ArrayList();
    public void add(){
        list.add("拾忆最初的理想");
    }
    public int size(){
        return list.size();
    }
}

class ThreadA extends Thread{
    private MyList list;
    public ThreadA(MyList list){
        this.list = list;
    }

    @Override
    public void run() {
        try{
            for (int i = 0; i < 10; i++) {
                list.add();
                System.out.println("添加了" + (i+1) +"个元素");
                Thread.sleep(1000);
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

class ThreadB extends Thread{
    private MyList list;
    public ThreadB(MyList list){
        this.list = list;
    }

    @Override
    public void run() {
        try{
            while (true){
                if(list.size() == 5){
                    System.out.println("==5,线程b要退出了");
                    throw new InterruptedException();
                }
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}
public class Test1 {
    public static void main(String[] args) {
        MyList service = new MyList();
        ThreadA a = new ThreadA(service);
        a.setName("a");
        a.start();
        ThreadB b = new ThreadB(service);
        b.setName("b");
        b.start();
    }
}

这里需要说明的就是即使不使用等待/通知机制也可以实现线程之间的通信，但是这种方法非常消耗资源，同时也可能导致数据的丢失，因此这多个线程都是使用的主动的方式来读取资源。因此这里才会使用wait/notify来实现线程之间的通信，这种机制就是当一个线程获取到资源的时候，就会使用notify方法唤醒该线程，这个线程就可以进行工作，而得不到资源的线程会调用wait方法进行休眠，直到该线程得到资源的时候才会再次调用notify方法将这个线程唤醒。

总结：wait使线程停止，notify使停止的线程继续运行。

public class Test2 {
    public static void main(String[] args) throws InterruptedException {
        String newString = new String("");
        newString.wait();
    }
}

如果在没有对象锁的线程中执行就会报IllegalMonitorStateException的错误。

public class Test2 {
    public static void main(String[] args) throws InterruptedException {
        String newString = new String("");
        System.out.println("aaa");
        synchronized (newString){
            System.out.println("bbb");
            newString.wait();
            System.out.println("ccc");
        }
        System.out.println("ddd");
    }
}

这样就可以执行了，程序只会输出aaa和bbb，从而会一直阻塞在这一行，除非有其他的线程获取这个对象锁并执行notify方法唤醒这个线程。这里需要注意的是一定要在主线程中创建一个共有的对象给两个线程，如果是两个对象的话，一个线程的notify方法是无法影响到另一个线程的wait方法的。

class ThreadC extends Thread{
    private String newString;

    public ThreadC(String newString){
        this.newString = newString;
    }
    @Override
    public void run() {
        System.out.println("aaa");
        synchronized (newString){
            try {
                System.out.println("bbb");
                long begin = System.currentTimeMillis();
                newString.wait();
                long end = System.currentTimeMillis();
                System.out.println("ccc");
                System.out.println("线程等待了"+(end - begin)/1000 + "秒");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("ddd");
    }
}

class ThreadD extends Thread{
    private String newString;

    public ThreadD(String newString){
        this.newString = newString;
    }
    @Override
    public void run() {
        try {
            Thread.sleep(10000);
            synchronized (newString){
                newString.notify();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
public class Test3 {
    public static void main(String[] args) {
        String newString = new String();
        ThreadC c = new ThreadC(newString);
        c.setName("c");
        c.start();
        ThreadD d = new ThreadD(newString);
        d.setName("d");
        d.start();
    }
}

注意：我们在调用wait方法的时候该线程会释放掉这个对象的锁，调用sleep方法的时候是不会释放锁的，调用notify方法的时候也是不会释放锁的，只有当notify所对应的线程执行完synchronized代码以后才会释放锁。

当interrupt方法遇到wait方法的时候

程序在执行同步代码块的时候，一般都是执行完以后才会释放锁；或者同步代码块在执行的过程中被打断了以后同步代码块就会停止执行，同时也会释放锁。

另外，notify方法会随机唤醒一个线程，而notifyAll方法会将所有休眠的线程都唤醒。

class Service{
    public void testMethod(Object lock){
        try{
            System.out.println("begin wait...");
            lock.wait();
            System.out.println("end wait...");
        }catch (InterruptedException e){
            e.printStackTrace();
            System.out.println("出现异常了，因为呈wait状态的线程被interrupt了");
        }
    }
}

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

    @Override
    public void run() {
        Service service = new Service();
        service.testMethod(lock);
    }
}
public class Test4 {
    public static void main(String[] args) {
        try {
            Object lock = new Object();
            ThreadE e = new ThreadE(lock);
            e.start();
            ThreadE.sleep(5000);
            e.interrupt();
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

wait(超时时间)方法：在等待的过程中，执行这个方法，如果在没有达到超时时间的时候，该线程会等待notify方法将其唤醒，如果达到超时时间了还有被唤醒，那么该线程就会自动唤醒。

如下面的例子，本来应该是等待10秒被notify唤醒的，但是因为wait设置了5秒的超时时间，因此在5秒的时候，该线程就直接被唤醒了

class ThreadC extends Thread{
    private String newString;

    public ThreadC(String newString){
        this.newString = newString;
    }
    @Override
    public void run() {
        System.out.println("aaa");
        synchronized (newString){
            try {
                System.out.println("bbb");
                long begin = System.currentTimeMillis();
                newString.wait(5000);
                long end = System.currentTimeMillis();
                System.out.println("ccc");
                System.out.println("线程等待了"+(end - begin)/1000 + "秒");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("ddd");
    }
}

class ThreadD extends Thread{
    private String newString;

    public ThreadD(String newString){
        this.newString = newString;
    }
    @Override
    public void run() {
        try {
            Thread.sleep(10000);
            synchronized (newString){
                newString.notify();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
public class Test3 {
    public static void main(String[] args) {
        String newString = new String();
        ThreadC c = new ThreadC(newString);
        c.setName("c");
        c.start();
        ThreadD d = new ThreadD(newString);
        d.setName("d");
        d.start();
    }
}

2，生产者/消费者模式实现

等待/通知模式最经典的案例就是“生产者/消费者”模式，但此模式在使用上有几种变形，还有一些小的注意事项，但原理都是基于wait/notify的。

变形1：一生产者一消费者

下面就是一对一的生产者消费者模式，set和get交替打印，并且下面的一个get和它上面的set的值一定是一样的。

//创建存储值对象
class ValueObject{
    public static String value = "";
}

//生产者类
class P{
    private String lock;
    public P(String lock){
        this.lock = lock;
    }
    public void setValue(){
        try{
            synchronized (lock){
                //有值的话就不生产
                if(!ValueObject.value.equals("")){
                    lock.wait();
                }
                String value = System.currentTimeMillis() + "_" + System.nanoTime();
                System.out.println("set的值是" + value);
                ValueObject.value = value;
                lock.notify();
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

//消费者类
class C{
    private String lock;
    public C(String lock){
        this.lock = lock;
    }
    public void getValue(){
        try{
            synchronized (lock){
                //没有值的话就生产
                if(ValueObject.value.equals("")){
                    lock.wait();
                }
                System.out.println("get的值是" + ValueObject.value);
                ValueObject.value = "";
                lock.notify();
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

//生产者线程
class ThreadP extends Thread{
    private P p;
    public ThreadP(P p){
        this.p = p;
    }

    @Override
    public void run() {
        while (true){
            p.setValue();
        }
    }
}

//消费者线程
class ThreadC extends Thread{
    private C c;
    public ThreadC(C c){
        this.c = c;
    }

    @Override
    public void run() {
        while (true){
            c.getValue();
        }
    }
}

public class Test5 {
    public static void main(String[] args) {
        String lock = new String("");
        P p = new P(lock);
        C c = new C(lock);
        new ThreadP(p).start();
        new ThreadC(c).start();
    }
}

变形2：多生产者多消费者

这种情况下会出现假死的情况，因为可能出现消费者唤醒的是消费者，生产者唤醒的是生产者的情况。

对上面的代码做一点小的修改

//创建存储值对象
class ValueObject{
    public static String value = "";
}

//生产者类
class P{
    private String lock;
    public P(String lock){
        this.lock = lock;
    }
    public void setValue(){
        try{
            synchronized (lock){
                //有值的话就不生产
                if(!ValueObject.value.equals("")){
                    System.out.println("生产者" + Thread.currentThread().getName()+"WAITING");
                    lock.wait();
                }
                System.out.println("生产者" + Thread.currentThread().getName()+"RUNNING");
                String value = System.currentTimeMillis() + "_" + System.nanoTime();
                System.out.println("set的值是" + value);
                ValueObject.value = value;
                lock.notify();
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

//消费者类
class C{
    private String lock;
    public C(String lock){
        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()+"RUNNING");
                System.out.println("get的值是" + ValueObject.value);
                ValueObject.value = "";
                lock.notify();
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

//生产者线程
class ThreadP extends Thread{
    private P p;
    public ThreadP(P p){
        this.p = p;
    }

    @Override
    public void run() {
        while (true){
            p.setValue();
        }
    }
}

//消费者线程
class ThreadC extends Thread{
    private C c;
    public ThreadC(C c){
        this.c = c;
    }

    @Override
    public void run() {
        while (true){
            c.getValue();
        }
    }
}

public class Test5 {
    public static void main(String[] args) {
        String lock = new String("");
        P p = new P(lock);
        C c = new C(lock);
        ThreadP[] pThread = new ThreadP[2];
        ThreadC[] cThread = new ThreadC[2];
        for (int i = 0; i < 2; i++) {
            pThread[i] = new ThreadP(p);
            pThread[i].setName("生产者"+(i+1));
            cThread[i] = new ThreadC(c);
            cThread[i].setName("消费者"+(i+1));
            pThread[i].start();
            cThread[i].start();
        }
    }
}

3，通过管道进行通信

在java中提供了各种各样的输入输出流，这些流使我们很方便地对数据进行操作，其中管道流流可以实现在不同的线程之间传输数据。一个线程发送数据到输出管道，另一个线程从输入管道中读取数据。

jdk提供了4个类来使线程间可以进行通信PipedInputStream和PipedOutputStream，PipedReader和PipedWriter

第一种：使用字节流

class WriteData{
    public void writeMethod(PipedOutputStream out){
        try{
            System.out.println("Write:");
            for (int i = 0; i < 300; i++) {
                String outData = "" + (i + 1);
                out.write(outData.getBytes());
                System.out.print(outData);
            }
            System.out.println();
            out.close();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
}

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();
        }
    }
}

class ThreadWrite extends Thread{
    private WriteData write;
    private PipedOutputStream out;

    public ThreadWrite(WriteData write, PipedOutputStream out) {
        this.write = write;
        this.out = out;
    }

    @Override
    public void run() {
        write.writeMethod(out);
    }
}

class ThreadRead extends Thread{
    private ReadData read;
    private PipedInputStream input;

    public ThreadRead(ReadData read, PipedInputStream input) {
        this.read = read;
        this.input = input;
    }

    @Override
    public void run() {
        read.readMethod(input);
    }
}
public class Test6 {
    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 | InterruptedException e){
            e.printStackTrace();
        }
    }
}

第二种：使用字符流

class WriteData1{
    public void writeMethod(PipedWriter out){
        try{
            System.out.println("Write:");
            for (int i = 0; i < 300; i++) {
                String outData = "" + (i + 1);
                out.write(outData);
                System.out.print(outData);
            }
            System.out.println();
            out.close();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
}

class ReadData1{
    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();
        }
    }
}

class ThreadWrite1 extends Thread{
    private WriteData1 write1;
    private PipedWriter out1;

    public ThreadWrite1(WriteData1 write1, PipedWriter out1) {
        this.write1 = write1;
        this.out1 = out1;
    }

    @Override
    public void run() {
        write1.writeMethod(out1);
    }
}

class ThreadRead1 extends Thread{
    private ReadData1 read1;
    private PipedReader input1;

    public ThreadRead1(ReadData1 read1, PipedReader input1) {
        this.read1 = read1;
        this.input1 = input1;
    }

    @Override
    public void run() {
        read1.readMethod(input1);
    }
}
public class Test7 {
    public static void main(String[] args) {
        try{
            WriteData1 writeData1 = new WriteData1();
            ReadData1 readData1 = new ReadData1();
            PipedReader pipedReader = new PipedReader();
            PipedWriter pipedWriter = new PipedWriter();
            //inputStream.connect(outputStream);
            pipedWriter.connect(pipedReader);
            ThreadRead1 threadRead = new ThreadRead1(readData1, pipedReader);
            threadRead.start();
            Thread.sleep(2000);
            ThreadWrite1 threadWrite = new ThreadWrite1(writeData1, pipedWriter);
            threadWrite.start();
        }catch (IOException | InterruptedException e){
            e.printStackTrace();
        }
    }
}

4，join方法的使用

在很多情况下，主线程创建并启动子线程，如果子线程需要大量的耗时的计算，主线程结束早于子线程，那么如果这个时候主线程需要子线程得到的结果的话，就需要在主线程中面调用子线程对象的join方法，只有有子线程执行完后主线程才能执行，这样主线程的结束就晚于子线程，这样就可以实现需求了。

在下面的例子中，如果我们把下面的这一行注释掉，那么主线程一定在前面执行完成，但是加入了注释的一行以后，主线程会阻塞在那一行，直到该线程结束以后主线程才会继续执行。

class MyThread extends Thread{

    @Override
    public void run() {
        try {
            System.out.println("我要开始执行了");
            Thread.sleep(10000);
            System.out.println("我执行结束啦");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
public class Test8 {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("主线程开始执行");
        MyThread myThread = new MyThread();
        myThread.start();
        //myThread.join();
        System.out.println("主线程执行结束了");
    }
}

注意：join方法和interrupt方法如果彼此遇到的话就会出现异常

join(超时时间)方法：如果不加时间的话，就相当于只有当子线程执行完了以后主线程才会执行，如果加了时间，就表示只会等待这么长的时间，不论子线程是否执行完，主线程都不会再等了。

class MyThread extends Thread{

    @Override
    public void run() {
        try {
            long begin = System.currentTimeMillis();
            System.out.println("我要开始执行了");
            Thread.sleep(10000);
            System.out.println("我执行结束啦");
            long end = System.currentTimeMillis();
            System.out.println("子线程的用时为" + (end - begin) / 1000 +"秒");
            System.out.println();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
public class Test8 {
    public static void main(String[] args) throws InterruptedException {
        long begin = System.currentTimeMillis();
        System.out.println("主线程开始执行");
        MyThread myThread = new MyThread();
        myThread.start();
        myThread.join(5000);
        long end = System.currentTimeMillis();
        System.out.println("主线程的用时为" + (end - begin) / 1000 +"秒");
        System.out.println("主线程执行结束了");
    }
}

5，类ThreadLocal的使用

变量值的共享可以使用public static变量的形式，所有的线程都使用同一个public static变量。如果想要实现每一个线程都有自己的共享变量，就可以使用ThreadLocal类，这个类的作用就是每个线程都可以绑定自己的值。

在使用ThreadLocal的时候，我们需要注意的是需要先设置值，然后再从里面拿值，这样的话才不会出现空指针异常。

class Tools{
    public static ThreadLocal t1 = new ThreadLocal<>();
}

class A extends Thread{
    @Override
    public void run() {
        try {
            for (int i = 0; i < 20; i++) {
                Tools.t1.set(100);
                System.out.println("A==>" + Tools.t1.get());
                Thread.sleep(100);
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

class B extends Thread{
    @Override
    public void run() {
        try {
            for (int i = 0; i < 20; i++) {
                Tools.t1.set(200);
                System.out.println("B==>" + Tools.t1.get());
                Thread.sleep(100);
            }
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

public class Test9 {
    public static void main(String[] args) {
        try {
            new A().start();
            Thread.sleep(1000);
            new B().start();
        }catch (InterruptedException e){
            e.printStackTrace();
        }
    }
}

多线程-线程间通信

Java相关栏目本月热门文章