Java中线程通信方式七种

朴灿烈づ我的快乐病毒、 2023-03-04 11:27 58阅读 0赞

### 文章目录 ###

*  一：volatile
 *  二：synchronized 临界区方式
 *  ReentrantLock/Condition 消息队列方式
 *  三：使用JUC工具类 CountDownLatch
 *  四：基本LockSupport实现线程间的阻塞和唤醒
 *  五：通过Socket网络通信
 *  六：信号量机制(Semaphore)：包括无名线程信号量和命名线程信号量
 *  七：管道通信就是使用java.io.PipedInputStream 和 java.io.PipedOutputStream进行通信

这篇博客是是我在学习了多线程并发和操作系统后，针对Java中的情况，为保证线程安全和线程的并发运行进行总结的博客。  
看了我的博客后如果哪里我有理解不到位地方欢迎大家评论区给我留言，感谢支持。

*  引言  
    Java支持多个线程同时访问一个对象或者对象的成员变量，由于每个线程可以拥有这个变量的拷贝（虽然对象以及成员变量分配的内存是在共享内存（堆）中的，但是每个执行的线程还是可以拥有一份拷贝，这样做的目的是加速程序的执行，这是现代多核处理器的一个显著特性），所以程序在执行过程中，一个线程看到的变量并不一定是最新的。所以在多个线程访问同一个数据时会将变量拷贝到自己线程独有的栈中进行数据的操作，在操作完后在把数据放回共享内存。这时就会有多线程并发的安全问题出现。保证多线程的并发运行，我们就需要多线程相互进行配合，那么这时就引出了线程间的通信。这里线程和进程概念不相同，因为Java的程序是运行于JVM之上，所以进程概念大家可以自行去了解。这篇博客主要讲Java里多线程之间的通信。
 *  首先，要短信线程间通信的模型有两种：**共享内存和消息传递**，以下方式都是基本这两种模型来实现的。
 *  题目：  
    有两个线程A、B，A线程向一个集合里面依次添加元素"abc"字符串，一共添加十次，当添加到第五次的时候，希望B线程能够收到A线程的通知，然后B线程执行相关的业务操作。

# 一：volatile #

关键字volatile可以用来修饰字段（成员变量），就是告知程序任何对该变量的访问均需要从共享内存中获取，而对它的改变必须同步刷新回共享内存，它能保证所有线程对变量访问的可见性。

使用while进行轮询在这种方式下，线程A不断地改变条件，线程ThreadB不停地通过while语句检测这个条件(list.size==5)是否成立 ，从而实现了线程间的通信。但是这种方式会浪费CPU资源。之所以说它浪费资源，是因为JVM调度器将CPU交给线程B执行时，它没做啥“有用”的工作，只是在不断地测试某个条件是否成立。

public class TestSync {
        
        // 定义一个共享变量来实现通信，它需要是volatile修饰，否则线程不能及时感知
        static volatile boolean notice = false;
    
        public static void main(String[] args) {
        
            List<String>  list = new ArrayList<>();
            // 实现线程A
            Thread threadA = new Thread(() -> {
        
                for (int i = 1; i <= 10; i++) {
        
                    list.add("abc");
                    System.out.println("线程A向list中添加一个元素，此时list中的元素个数为：" + list.size());
                    try {
        
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                    if (list.size() == 5)
                        notice = true;
                }
            });
            // 实现线程B
            Thread threadB = new Thread(() -> {
        
                while (true) {
        
                    if (notice) {
        
                        System.out.println("线程B收到通知，开始执行自己的业务...");
                        break;
                    }
                }
            });
            //　需要先启动线程B
            threadB.start();
            try {
        
                Thread.sleep(1000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            // 再启动线程A
            threadA.start();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70]

# 二：synchronized 临界区方式 #

synchronized和ReentrantLock实现原理上有所不同，但是针对线程之间的通信的思想相同，都是保证多个线程互斥访问临界区（共享存储），来实现线程间的通信。

*  synchronized  
    根据synchronized原理使用Object类提供了线程间通信的方法：wait()、notify()、notifyaAl()方法来实现多个线程互斥访问临界区资源，Object类这几个方法必须配合synchronized来进行使用。

public class TestSync {
        
        public static void main(String[] args) {
        
            // 定义一个锁对象
            Object lock = new Object();
            List<String>  list = new ArrayList<>();
            // 实现线程A
            Thread threadA = new Thread(() -> {
        
                synchronized (lock) {
        
                    for (int i = 1; i <= 10; i++) {
        
                        list.add("abc");
                        System.out.println("线程A向list中添加一个元素，此时list中的元素个数为：" + list.size());
                        try {
        
                            Thread.sleep(500);
                        } catch (InterruptedException e) {
        
                            e.printStackTrace();
                        }
                        if (list.size() == 5)
                            lock.notify();// 唤醒B线程
                    }
                }
            });
            // 实现线程B
            Thread threadB = new Thread(() -> {
        
                while (true) {
        
                    synchronized (lock) {
        
                        if (list.size() != 5) {
        
                            try {
        
                                lock.wait();
                            } catch (InterruptedException e) {
        
                                e.printStackTrace();
                            }
                        }
                        System.out.println("线程B收到通知，开始执行自己的业务...");
                    }
                }
            });
            //　需要先启动线程B
            threadB.start();
            try {
        
                Thread.sleep(1000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            // 再启动线程A
            threadA.start();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70 1]

# ReentrantLock/Condition 消息队列方式 #

*  ReentrantLock/Condition  
    ReentrantLock锁的实现原理虽然和synchronized不用，但是它和synchronized一样都是通过保证线程间的互斥访问临界区，来保证线程安全，实现线程间的通信。相比于synchronized使用Object类的三个方法来实现线程的阻塞和运行两个状态的切换，ReentrantLock使用Condition阻塞队列的await()、signal()、signalAll()三个方法来实现线程阻塞和运行两个状态的切换，进而实现线程间的通信。

public class TestSync {
        
        public static void main(String[] args) {
        
            ReentrantLock lock = new ReentrantLock();
            Condition condition = lock.newCondition();
    
            List<String> list = new ArrayList<>();
            // 实现线程A
            Thread threadA = new Thread(() -> {
        
                lock.lock();
                for (int i = 1; i <= 10; i++) {
        
                    list.add("abc");
                    System.out.println("线程A向list中添加一个元素，此时list中的元素个数为：" + list.size());
                    try {
        
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                    if (list.size() == 5)
                        condition.signal();
    
                }
                lock.unlock();
            });
            // 实现线程B
            Thread threadB = new Thread(() -> {
        
                lock.lock();
                if (list.size() != 5) {
        
                    try {
        
                        condition.await();
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
                System.out.println("线程B收到通知，开始执行自己的业务...");
                lock.unlock();
            });
            threadB.start();
            try {
        
                Thread.sleep(1000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            threadA.start();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70 2]  
显然这种方式使用起来并不是很好，代码编写复杂，而且线程B在被A唤醒之后由于没有获取锁还是不能立即执行，也就是说，A在唤醒操作之后，并不释放锁。这种方法跟 Object 的 wait() 和 notify() 一样。

# 三：使用JUC工具类 CountDownLatch #

jdk1.5之后在java.util.concurrent包下提供了很多并发编程相关的工具类，简化了我们的并发编程代码的书写，\*\*\*CountDownLatch\*\*\*基于AQS框架，相当于也是维护了一个线程间共享变量state

public class TestSync {
        
        public static void main(String[] args) {
        
            CountDownLatch countDownLatch = new CountDownLatch(1);
            List<String>  list = new ArrayList<>();
            // 实现线程A
            Thread threadA = new Thread(() -> {
        
                for (int i = 1; i <= 10; i++) {
        
                    list.add("abc");
                    System.out.println("线程A向list中添加一个元素，此时list中的元素个数为：" + list.size());
                    try {
        
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                    if (list.size() == 5)
                        countDownLatch.countDown();
                }
            });
            // 实现线程B
            Thread threadB = new Thread(() -> {
        
                while (true) {
        
                    if (list.size() != 5) {
        
                        try {
        
                            countDownLatch.await();
                        } catch (InterruptedException e) {
        
                            e.printStackTrace();
                        }
                    }
                    System.out.println("线程B收到通知，开始执行自己的业务...");
                    break;
                }
            });
            //　需要先启动线程B
            threadB.start();
            try {
        
                Thread.sleep(1000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            // 再启动线程A
            threadA.start();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70 3]

# 四：基本LockSupport实现线程间的阻塞和唤醒 #

LockSupport 是一种非常灵活的实现线程间阻塞和唤醒的工具，使用它不用关注是等待线程先进行还是唤醒线程先运行，但是得知道线程的名字。

public class TestSync {
        
        public static void main(String[] args) {
        
            List<String> list = new ArrayList<>();
            // 实现线程B
            final Thread threadB = new Thread(() -> {
        
                if (list.size() != 5) {
        
                    LockSupport.park();
                }
                System.out.println("线程B收到通知，开始执行自己的业务...");
            });
            // 实现线程A
            Thread threadA = new Thread(() -> {
        
                for (int i = 1; i <= 10; i++) {
        
                    list.add("abc");
                    System.out.println("线程A向list中添加一个元素，此时list中的元素个数为：" + list.size());
                    try {
        
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                    if (list.size() == 5)
                        LockSupport.unpark(threadB);
                }
            });
            threadA.start();
            threadB.start();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70 4]

# 五：通过Socket网络通信 #

网络通信例如我们的NIO，BIO通过Socket套接字来进行通信  
这里代码在我的另外一篇博客中写到[NIO/BIO的Java实现][NIO_BIO_Java]  
当然还有我们的各种网络通信框架例如netty等等。都是通过网络通信来实现多线程的通信。

网络通信和前面的不同，他是基于Socket实现的，所以不是牵扯前面所说的共享内存之类的概念，当然因为是网络通信，它可以实现不同设备间的通信。

# 六：信号量机制(Semaphore)：包括无名线程信号量和命名线程信号量 #

这个大家可以自行进行查询

# 七：管道通信就是使用java.io.PipedInputStream 和 java.io.PipedOutputStream进行通信 #

管道通信这里不做具体阐述，感兴趣朋友可自行了解。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70]: /images/20230208/46ba06a818314373babd1591d5d8e8c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70 1]: /images/20230208/3894a6ffba25451eb3151976ccee6b7f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70 2]: /images/20230208/3c12d7bf0de149848ff31cef174b5664.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70 3]: /images/20230208/af092698b7724a6e96d4800b6043fd0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNDExMjE0_size_16_color_FFFFFF_t_70 4]: /images/20230208/c39bc5cf75e44fc2967f84c9b5f76e0b.png
[NIO_BIO_Java]: https://blog.csdn.net/qq_42411214/article/details/104640148