线程的基础知识（五）之线程通信

深碍√TFBOYSˉ_ 2021-12-10 10:43 314阅读 0赞

## 传统的线程通信 ##

当线程在系统内运行时，线程的调度具有一定的透明性，程序通常无法准确控制线程的轮换执行，单Java也提供了一些机制来保证线程协调运行。为了实现这一机制，可以借助Object类提供的wait(),notify()和notifyAll()三个方法：

1.  wait() ：导致当前线程等待，直到其他线程调用该同步监视器的notify()方法或notifyAll()方法来唤醒该线程。调用wait()方法的当前线程会释放该同步监视器的锁定；
2.  notify()：唤醒在此同步监视器上等待的单个线程，若有多个等待线程，则会随机唤醒其中一个；
3.  notify() ：唤醒在此同步监视器上等待的所有线程。  
    演示代码如下：

**Account类：**

package com.dalingjia.thead.threadCommunication.tradition;
    
    public class Account {
    
        //账号编号
        private String accountNo;
        //账号余额
        private Double balance;
    
        public Account(String accountNo, Double balance) {
            this.accountNo = accountNo;
            this.balance = balance;
        }
    
        public synchronized void draw(double drawAmount) {
            try {
                //账户没钱，等待
                if (!(balance > 0)) {
                    wait();
                } else {
                    balance -= drawAmount;
                    System.out.println(Thread.currentThread().getName() + "<-取走：" + drawAmount + " ,余额：" + balance);
                    notifyAll();
                }
    
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        //存钱
        public synchronized void deposit(double depositAmount) {
            try {
                //账户有钱，等待
                if (balance > 0) {
                    wait();
                } else {
                    balance += depositAmount;
                    System.out.println(Thread.currentThread().getName() + "->存进：" + depositAmount + " ,余额：" + balance);
                    notifyAll();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

**取钱线程：**

package com.dalingjia.thead.threadCommunication.tradition;
    
    public class DrawThread extends Thread {
    
        private Account account;
    
        private Double amount;
    
        public DrawThread(String name, Account account, Double amount) {
            super(name);
            this.account = account;
            this.amount = amount;
        }
    
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                account.draw(amount);
            }
        }
    }

**存钱线程：**

package com.dalingjia.thead.threadCommunication.tradition;
    
    /**
     * 存钱线程
     */
    public class DepositThread extends Thread {
    
        private Account account;
    
        //存钱数量
        private Double amount;
    
        public DepositThread(String name, Account account, Double amount) {
            super(name);
            this.account = account;
            this.amount = amount;
        }
    
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                account.deposit(amount);
            }
        }
    }

**测试类：**

package com.dalingjia.thead.threadCommunication.tradition;
    
    public class Test {
    
        public static void main(String[] args) {
            //1,定义一个账户
            Account account = new Account("thq10000", 0d);
            //2，准备target
            DepositThread depositThread = new DepositThread("存",account, 100d);
            DrawThread drawThread = new DrawThread( "取",account, 100d);
            //3,开启2个线程
            depositThread.start();
            drawThread.start();
        }
    }

## 使用Condition控制线程通信： ##

如果程序不使用synchronized关键字来保证同步，而是直接使用Lock对象来保证同步，则系统中不存在隐式的同步监视器，也就是不使用wait(),notify()和notifyAll()方法进行线程通信，而是使用Condition对象来进行通信，它有如下三个方法：

1.  await()：类似于wait()，导致当前线程等待，知道其他线程调用Condition的signal()或signalAll()来唤醒该线程。
2.  signal()：类似于notify()，随机唤醒一个线程；
3.  signalAll()：唤醒在此Lock对象上等待的所有线程。  
    代码演示如下：

**Account类如下，其他类和上面代码相同：**

package com.dalingjia.thead.threadCommunication.condition;
    
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class Account {
    
        //使用同步锁来保证线程同步
        private final Lock lock = new ReentrantLock();
    
        //使用condition对象来进行线程通信
        private final Condition condition = lock.newCondition();
    
        //账号编号
        private String accountNo;
        //账号余额
        private Double balance;
    
        public Account(String accountNo, Double balance) {
            this.accountNo = accountNo;
            this.balance = balance;
        }
    
        public  void draw(double drawAmount) {
            lock.lock();
            try {
                //没钱，等待
                if(!(balance>0)) {
                    condition.await();
                }else {
                    balance -= drawAmount;
                    System.out.println(Thread.currentThread().getName() + "<-取走：" + drawAmount + " ,余额：" + balance);
                    condition.signalAll();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                //使用finally块来释放锁
                lock.unlock();
            }
    
        }
    
        //存钱
        public synchronized void deposit(double depositAmount) {
            lock.lock();
            try {
                //账户有钱，等待
                if (balance > 0) {
                    condition.await();
                } else {
                    balance += depositAmount;
                    System.out.println(Thread.currentThread().getName() + "->存进：" + depositAmount + " ,余额：" + balance);
                    condition.signalAll();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                //使用finally块来释放锁
                lock.unlock();
            }
        }
    }

## 使用阻塞队列（BlockingQueue）来控制线程通信 ##

1.  BlockingQueue作为线程同步的工具，具有一个特征：当生产者线程试图想BlockingQueue中放入元素时，如果该队列已满，则该线程阻塞；当消费者线程试图从BlockingQueue中取出元素时，如果该队列已空，则该线程被阻塞。
2.  BlockingQueue提供如下两个支持阻塞的方法：  
    put(E e): 尝试把e元素放入BlockingQueue中，如果该队列的元素已满，则阻塞该线程。  
    take(): 尝试从BlockingQueue的头部取出元素，如果该队列的元素已空，则阻塞该线程。

**BlockingQueue包含的方法之间的对应关系如下：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70]  
BlockingQueue代码演示如下：

**生成者：**

package com.dalingjia.thead.threadCommunication.BlockingQueue;
    
    import java.util.concurrent.BlockingQueue;
    
    public class Producer extends Thread {
    
        private BlockingQueue<String> bq;
    
        public Producer(String name, BlockingQueue<String> bq) {
            super(name);
            this.bq = bq;
        }
    
        @Override
        public void run() {
            String[] strings = {"Java", "Spring", "Mysql"};
            for (int i = 0; i < 10; i++) {
                try {
                    bq.put(strings[i%3]);
                    System.out.println(getName() + "生成：" + bq);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

**消费者：**

package com.dalingjia.thead.threadCommunication.BlockingQueue;
    
    import java.util.concurrent.BlockingQueue;
    
    public class Consumer extends Thread {
    
        private BlockingQueue bq ;
    
        public Consumer(String name, BlockingQueue bq) {
            super(name);
            this.bq = bq;
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    bq.take();
                    System.out.println(getName() + "消费：" + bq);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
            }
        }
    }

测试类：

package com.dalingjia.thead.threadCommunication.BlockingQueue;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    public class Test {
    
        public static void main(String[] args) {
            BlockingQueue<String> bq = new ArrayBlockingQueue<>(1);
            Producer producer = new Producer("生产者->", bq);
            Consumer cousumer = new Consumer("消费者<-", bq);
            producer.start();
            cousumer.start();
        }
    }

上述测试代码存在线程安全问题，本测试只为加深理解BlockingQueue。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70]: /images/20211210/c943828f64d34127b06864cf86ed0fb1.png