Java线程之间通信方式

秒速五厘米 2024-03-04 02:07 95阅读 0赞

#### 目录 ####

*  1 线程之间的通信方式主要有以下几种
 *  2 共享变量
 *  3 锁机制
 *  4 条件变量
 *  5 信号量
 *  6 管道
 *  6 join方法的实现原理
 *  7 join方法实现顺序性原理

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

## 1 线程之间的通信方式主要有以下几种 ##

![在这里插入图片描述][52145529bf67457184361639bc3da6a2.png]

在实际开发时，一个进程中往往有很多个线程，大多数线程之间往往不是绝对独立的，比如说我们需要将A和B 两个线程的执行结果收集在一起然后显示在界面上，又或者比较典型的消费者-生产者模式，在这些场景下，线程间通信成了我们必须使用的手段，那么线程之间怎么通信呢？

线程间通信方式，从实现本质来讲，主要可以分为两大类共享内存和消息传递。

相信大家还记得，在内存模型一节，我们提到多线程并发情况下的三大特性，原子性，有序性，可见性，其所对应的解决方案就可以用来实现线程间通信，这些解决方案的本质就是共享内存。

对于消息传递而言，最经典的实现就是我们的Handler机制，在子线程使用主线程的Handler对象将一些信息发送到主线程以便进行处理。

下面我们来看一些线程间通信的典型实现

## 2 共享变量 ##

共享变量：线程之间可以通过共享变量来进行通信。不同的线程可以共享同一个变量，并在变量上进行读写操作。需要注意的是，共享变量可能会引发[线程安全问题][Link 1]，需要通过同步机制来确保线程安全。

public class SharedData {
        
        private int value;
        public synchronized int getValue() {
         
            return value; 
        }
        public synchronized void setValue(int value) {
         
            this.value = value; 
        }
    }

在这个示例中，定义了一个共享数据类 `SharedData`，其中包含一个整型变量 `value` 和两个同步方法 `getValue()` 和 `setValue()`，用于获取和设置变量的值。由于这两个方法都是同步的，因此多个线程可以安全地访问该变量。

public class SharedDataExample {
        
        public static void main(String[] args) throws InterruptedException {
        
            SharedData sharedData = new SharedData();
            Thread thread1 = new Thread(() -> {
        
                for (int i = 0; i < 10; i++) {
        
                    sharedData.setValue(i);
                    System.out.println(Thread.currentThread().getName() + " write " + sharedData.getValue());
                    try {
        
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            });
            Thread thread2 = new Thread(() -> {
        
                for (int i = 0; i < 10; i++) {
        
                    System.out.println(Thread.currentThread().getName() + " read " + sharedData.getValue());
                    try {
        
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            });
            thread1.start();
            thread2.start();
            thread1.join();
            thread2.join();
        }
    }

在这个示例中，创建了两个线程分别用于读写共享数据 `SharedData`，多次执行该示例可以看到控制台输出表明两个线程在安全地访问共享变量。

结果如图：  
![在这里插入图片描述][72f8bad25b5e45ffa612751cf86cb0ae.png]

## 3 锁机制 ##

[锁机制][Link 2]：锁机制是一种常用的线程同步机制，可以保证在同一时间只有一个线程能够访问共享资源。Java提供了多种锁类型，如 synchronized 关键字、ReentrantLock 类等。

public class LockExample {
        
        private static Lock lock = new ReentrantLock();
        private static int count = 0;
        private static void increase() {
        
            lock.lock();
            try {
        
                count++;
            } finally {
        
                lock.unlock();
            }
        }
        public static void main(String[] args) throws InterruptedException {
        
            Thread thread1 = new Thread(() -> {
        
                for (int i = 0; i < 10000; i++) {
        
                    increase();
                }
            });
            Thread thread2 = new Thread(() -> {
        
                for (int i = 0; i < 10000; i++) {
        
                    increase();
                }
            });
            thread1.start();
            thread2.start();
            thread1.join();
            thread2.join();
            System.out.println(count);
        }
    }

在这个示例中，使用了 Lock 接口和 ReentrantLock 类来对计数器进行同步，多次执行该示例可以看到最终输出的计数器值为 20000。

结果如图：  
![在这里插入图片描述][cd575b8ee66344eb8642b61b53bece99.png]

## 4 条件变量 ##

条件变量：条件变量是一种线程间通信机制，它用于在一个共享资源上等待某个条件的成立。Java 提供了 Condition 接口来支持条件变量的实现，在使用 Condition 时需要先获取锁，然后调用 await() 方法等待条件成立，当条件成立时可以通过 signal() 或 signalAll() 方法唤醒等待该条件的线程。

public class ConditionExample {
        
        private static Lock lock = new ReentrantLock();
        private static Condition condition = lock.newCondition();
        private static int count = 0;
        private static void await() throws InterruptedException {
        
            lock.lock();
            try {
        
                condition.await();
            } finally {
        
                lock.unlock();
            }
        }
        private static void signal() {
        
            lock.lock();
            try {
        
                condition.signalAll();
            } finally {
        
                lock.unlock();
            }
        }
        public static void main(String[] args) throws InterruptedException {
        
            Thread thread1 = new Thread(() -> {
        
                for (int i = 0; i < 10; i++) {
        
                    try {
        
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                    count++;
                    System.out.println(Thread.currentThread().getName() + " increase count to " + count);
                    if (count == 5) {
        
                        signal();
                    }
                }
            });
            Thread thread2 = new Thread(() -> {
        
                try {
        
                    await();
                    System.out.println(Thread.currentThread().getName() + " receive signal, count is " + count);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            });
            thread1.start();
            thread2.start();
            thread1.join();
            thread2.join();
        }
    }

在这个示例中，使用了 Lock 接口和 Condition 接口来定义了一个计数器，线程1每次增加计数器的值并判断是否达到条件，当计数器达到条件时调用 signal() 方法通知线程2，线程2等待条件成立后执行相应的操作。

![在这里插入图片描述][5d0456c35795464c9d63f629c2eae312.png]

## 5 信号量 ##

信号量：信号量是一种常见的[线程同步][Link 3]机制，可用于控制多个线程对共享资源的访问。Java 提供了 Semaphore 类来实现信号量，Semaphore 类有两个常用的方法 acquire() 和 release()，分别用于获取和释放信号量。

public class SemaphoreExample {
        
        private static Semaphore semaphore = new Semaphore(2);
        private static void doWork() throws InterruptedException {
        
            semaphore.acquire();
            System.out.println(Thread.currentThread().getName() + " start working");
            Thread.sleep(1000);
            System.out.println(Thread.currentThread().getName() + " finish working");
            semaphore.release();
        }
        public static void main(String[] args) throws InterruptedException {
        
            Thread thread1 = new Thread(() -> {
        
                try {
        
                    doWork();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            });
            Thread thread2 = new Thread(() -> {
        
                try {
        
                    doWork();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            });
            Thread thread3 = new Thread(() -> {
        
                try {
        
                    doWork();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            });
            thread1.start();
            thread2.start();
            thread3.start();
            thread1.join();
            thread2.join();
            thread3.join();
        }
    }

在这个示例中，使用了 Semaphore 类来定义了一个信号量，线程1、线程2、线程3都需要获取信号量才能进行工作，每次执行 doWork() 方法需要占用资源，执行完毕后释放信号量。

![在这里插入图片描述][fde0f7911de6429f93ea1e4dd00040fc.png]

## 6 管道 ##

管道：管道是一种用于线程间通信的高级机制，它可以实现一个线程向另一个线程传送数据。Java 提供了 PipedInputStream 和 PipedOutputStream 两个类来支持管道的实现，其中 PipedInputStream 用于读取数据，PipedOutputStream 用于写入数据。

public class PipeExample {
        
        static class WriterThread extends Thread {
        
            private PipedOutputStream output;
            WriterThread(PipedOutputStream output) {
        
                this.output = output;
            }
            @Override
            public void run() {
        
                try {
        
                    for(int i=1;i<=10;i++) {
        
                        output.write(i);
                        System.out.println("写入数据：" + i);
                        Thread.sleep(1000);
                    }
                } catch(Exception e) {
        
                    e.printStackTrace();
                } finally {
        
                    try {
        
                        output.close();
                    } catch(Exception e) {
        
                        e.printStackTrace();
                    }
                }
            }
        }
        static class ReaderThread extends Thread {
        
            private PipedInputStream input;
            ReaderThread(PipedInputStream input) {
        
                this.input = input;
            }
            @Override
            public void run() {
        
                try {
        
                    int value;
                    while((value=input.read()) != -1) {
        
                        System.out.println("读取数据：" + value);
                    }
                } catch(Exception e) {
        
                    e.printStackTrace();
                } finally {
        
                    try {
        
                        input.close();
                    } catch(Exception e) {
        
                        e.printStackTrace();
                    }
                }
            }
        }
        public static void main(String[] args) throws IOException {
        
            PipedOutputStream output = new PipedOutputStream();
            PipedInputStream input = new PipedInputStream(output);
            Thread thread1 = new WriterThread(output);
            Thread thread2 = new ReaderThread(input);
            thread1.start();
            thread2.start();
            try {
        
                thread1.join();
                thread2.join();
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }

在这个示例中，使用了 PipedOutputStream 类和 PipedInputStream 类来定义了一个管道，线程1向管道中写入数据，线程2从管道中读取数据，通过管道来实现两个线程之间的通信。

运行结果：![在这里插入图片描述][9b6ced74339143edafa43dbd5922b18c.png]

需要注意的是，以上通信方式都需要在多线程程序中谨慎使用，需要考虑线程安全和性能等方面的问题。为了确保程序正确、高效地运行，需要根据具体情况选择合适的线程通信方式，并进行相应的测试和优化。

## 6 join方法的实现原理 ##

说了这么多，也举了这么多例子了，join方法是怎么实现线程等待的呢？我点进去join方法内部可以看到，其实它的内部也就是调用了wait()方法，

![在这里插入图片描述][a5b4b326f4ef41129eb46969011fe129.png]

只不过它多做了一步，用了一个循环来判断线程是否还活着，**isAlive()方法**就是用来判断线程是否还活着；

![在这里插入图片描述][dfcab7094d274e1c9b119f9c263e9e4c.png]

## 7 join方法实现顺序性原理 ##

那么这个时候肯定有的同学就有疑问了，只看到join()方法调用了wait()方法，但是没看到有调用notify() 或者 notifyAll() 系列的唤醒方法，那它是怎么唤醒的呢？**如果不唤醒，那不就一直等待下去了吗？**

***\*原来啊，在java中，Thread类线程执行完run()方法后，一定会自动执行notifyAll()方法\****

这是notifyAll()非常重要的隐藏知识点，这个细节隐藏在Java的Native方法中，所以一般不会被人发现。我们观察C/C++源码，如下：

oid JavaThread::exit(booldestory_vm, ExitTypeexit_type);
    static void ensure_join(JavaThread*thread) {
        
    	Handle threadObj(thread, thread -> threadObj());
    	ObjectLocker lock(threadObj, thread);
    	thread -> clear_pending_exception();
    	java_lang_Thread::set_thread_status(threadObj(), java_lang_Thread::TERMINATED);
    	java_lang_Thread::set_thread(threadObj(), NULL);
         //下行执行了notifyAll()操作
    	lock.notify_all(thread);
    	thread -> clear_pending_exception();
    	}

其中`ensure_join`就是执行join()方法，等方法执行结束时，此行代码`lock.notify_all(thread);`意思是通过`notifyAll()`唤醒了所有等待线程。所以在使用线程的时候，要特别注意

[52145529bf67457184361639bc3da6a2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/42013f4b59974a78a9ec9e161e736027.png
[Link 1]: https://so.csdn.net/so/search?q=%E7%BA%BF%E7%A8%8B%E5%AE%89%E5%85%A8%E9%97%AE%E9%A2%98&spm=1001.2101.3001.7020
[72f8bad25b5e45ffa612751cf86cb0ae.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/e840b5d374d549598f451438b52027cd.png
[Link 2]: https://so.csdn.net/so/search?q=%E9%94%81%E6%9C%BA%E5%88%B6&spm=1001.2101.3001.7020
[cd575b8ee66344eb8642b61b53bece99.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/b13fb1717b9c430a90d46f3fce58397f.png
[5d0456c35795464c9d63f629c2eae312.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/9b6403cd7119488faefb78e1ec44deb0.png
[Link 3]: https://so.csdn.net/so/search?q=%E7%BA%BF%E7%A8%8B%E5%90%8C%E6%AD%A5&spm=1001.2101.3001.7020
[fde0f7911de6429f93ea1e4dd00040fc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/51fc8b69d58e43b88ca980207833219f.png
[9b6ced74339143edafa43dbd5922b18c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/f25bdcd66b714d8090cba335f01161cd.png
[a5b4b326f4ef41129eb46969011fe129.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/25bc9a36fce44f4d8bea3363eac61c9d.png
[dfcab7094d274e1c9b119f9c263e9e4c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/04/41bf365487d94c8a945602fc50f5250f.png