Detailed explanation of Java thread communication

Thread communication is used to ensure the coordinated operation of threads. Generally, thread communication issues need to be considered when doing thread synchronization.

1. Traditional thread communication

Usually use the three methods provided by the Objeclt class:

wait() causes the current thread to wait and releases the lock of the synchronization monitor , until other threads call the notify() or notifyAll() method of the synchronization monitor to wake up the thread.

notify(), wake up the threads waiting on this synchronization monitor, if there are multiple, one will be randomly selected to wake up

notifyAll(), wake up all threads waiting on this synchronization monitor , after these threads compete for resources through scheduling, a thread acquires the lock of this synchronization monitor and then runs.

These three methods must be called by the synchronization monitor object and are divided into two situations:

When synchronizing the method, since the synchronization monitor is this object, these three methods can be called directly .

The example is as follows:

public class SyncMethodThreadCommunication {
  static class DataWrap{
    int data = 0;
    boolean flag = false;
     
    public synchronized void addThreadA(){
      if (flag) {
        try {
          wait();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      } 
       
      data++;
      System.out.println(Thread.currentThread().getName() + " " + data);
      flag = true;
      notify();
    }
     
    public synchronized void addThreadB() {
      if (!flag) {
        try {
          wait();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      } 
       
      data++;
      System.out.println(Thread.currentThread().getName() + " " + data);
      flag = false;
      notify();
    }
  }
   
  static class ThreadA extends Thread {
    private DataWrap data;
     
    public ThreadA(DataWrap dataWrap) {
      this.data = dataWrap;
    }
     
    @Override
    public void run() {
      for (int i = 0; i < 10; i++) {
        data.addThreadA();
      }
    }
  }
   
  static class ThreadB extends Thread {
    private DataWrap data;
     
    public ThreadB(DataWrap dataWrap) {
      this.data = dataWrap;
    }
     
    @Override
    public void run() {
      for (int i = 0; i < 10; i++) {
        data.addThreadB();
      }
    }
  }
   
  public static void main(String[] args) {
    //实现两个线程轮流对数据进行加一操作
    DataWrap dataWrap = new DataWrap();
     
    new ThreadA(dataWrap).start();
    new ThreadB(dataWrap).start();
  }
 
}

When synchronizing the code block, you need to use the monitor object to call these three methods.

Examples are as follows:

public class SyncBlockThreadComminication {
  static class DataWrap{
    boolean flag;
    int data;
  }
   
  static class ThreadA extends Thread{
    DataWrap dataWrap;
     
    public ThreadA(DataWrap dataWrap){
      this.dataWrap = dataWrap;
    }
     
    @Override
    public void run() {
      for(int i = 0 ; i < 10; i++) {
        synchronized (dataWrap) {
          if (dataWrap.flag) {
            try {
              dataWrap.wait();
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
          }
           
          dataWrap.data++;
          System.out.println(getName() + " " + dataWrap.data);
          dataWrap.flag = true;
          dataWrap.notify();
        }  
      }
    }
  }
   
  static class ThreadB extends Thread{
    DataWrap dataWrap;
     
    public ThreadB(DataWrap dataWrap){
      this.dataWrap = dataWrap;
    }
     
    @Override
    public void run() {
      for (int i = 0; i < 10; i++) {
          synchronized (dataWrap) {
            if (!dataWrap.flag) {
              try {
                dataWrap.wait();
              } catch (InterruptedException e) {
                e.printStackTrace();
              }
            }
             
            dataWrap.data++;
            System.out.println(getName() + " " + dataWrap.data);
            dataWrap.flag = false;
            dataWrap.notify();
          }
        }  
      }
       
  }
  public static void main(String[] args) {
    //实现两个线程轮流对数据进行加一操作
     
    DataWrap dataWrap = new DataWrap();
    new ThreadA(dataWrap).start();
    new ThreadB(dataWrap).start();
  }
 
}

2. Use Condition to control thread communication

When using the Lock object to ensure synchronization, use the Condition object to ensure coordination.

Examples are as follows:

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
 
import com.sun.media.sound.RIFFInvalidDataException;
 
import javafx.scene.chart.PieChart.Data;
 
public class SyncLockThreadCommunication {
  static class DataWrap {
    int data;
    boolean flag;
     
    private final Lock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();
     
    public void addThreadA() {
      lock.lock();
      try {
        if (flag) {
          try {
            condition.await();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
         
        data++;
        System.out.println(Thread.currentThread().getName() + " " + data);
        flag = true;
        condition.signal();
      } finally {
        lock.unlock();
      }
    }
     
    public void addThreadB() {
      lock.lock();
      try {
        if (!flag) {
          try {
            condition.await();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        }
         
        data++;
        System.out.println(Thread.currentThread().getName() + " " + data);
        flag = false;
        condition.signal();
      } finally {
        lock.unlock();
      }
    }
  }
   
  static class ThreadA extends Thread{
    DataWrap dataWrap;
     
    public ThreadA(DataWrap dataWrap) {
      this.dataWrap = dataWrap;
    }
     
    @Override
    public void run() {
      for (int i = 0; i < 10; i++) {
        dataWrap.addThreadA();
      }
    }
  }
   
  static class ThreadB extends Thread{
    DataWrap dataWrap;
     
    public ThreadB(DataWrap dataWrap) {
      this.dataWrap = dataWrap;
    }
     
    @Override
    public void run() {
      for (int i = 0; i < 10; i++) {
        dataWrap.addThreadB();
      }
    }
  }
   
  public static void main(String[] args) {
    //实现两个线程轮流对数据进行加一操作
     
    DataWrap dataWrap = new DataWrap();
    new ThreadA(dataWrap).start();
    new ThreadB(dataWrap).start();
  }
 
}

The await(), singal(), and singalAll() of the Condition object correspond to the wait(), notify(), and notifyAll() methods respectively.

3. Use blocking queue BlockingQueue to control thread communication

BlockingQueue is a sub-interface of the Queue interface. It is mainly used for thread communication. It has a characteristic: when the producer thread tries to enter the BlockingQueue When an element is put in, if the queue is full, the thread is blocked; when the consumer thread tries to take out an element from the BlockingQueue, if the queue is empty, the thread is blocked. These two features respectively correspond to two methods that support blocking, put(E e) and take()

The examples are as follows:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
 
public class BlockingQueueThreadComminication {
  static class DataWrap{
    int data;
  }
   
  static class ThreadA extends Thread{
    private BlockingQueue<DataWrap> blockingQueue;
     
    public ThreadA(BlockingQueue<DataWrap> blockingQueue, String name) {
      super(name);
      this.blockingQueue = blockingQueue;
    }
     
    @Override
    public void run() {
      for (int i = 0; i < 100; i++) {
        try {
          DataWrap dataWrap = blockingQueue.take();
           
          dataWrap.data++;
          System.out.println(getName() + " " + dataWrap.data);
          sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }
   
  static class ThreadB extends Thread{
    private BlockingQueue<DataWrap> blockingQueue;
    private DataWrap dataWrap;
     
    public ThreadB(BlockingQueue<DataWrap> blockingQueue, DataWrap dataWrap, String name) {
      super(name);
      this.blockingQueue = blockingQueue;
      this.dataWrap = dataWrap;
    }
     
    @Override
    public void run() {
      for (int i = 0; i < 100; i++) {
        try {
          dataWrap.data++;
          System.out.println(getName() + " " + dataWrap.data);
          blockingQueue.put(dataWrap);
          sleep(1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }
  }
   
  public static void main(String[] args) {
    ///实现两个线程轮流对数据进行加一操作
     
    DataWrap dataWrap = new DataWrap();
    BlockingQueue<DataWrap> blockingQueue = new ArrayBlockingQueue<>(1);
     
    new ThreadA(blockingQueue, "Consumer").start();
    new ThreadB(blockingQueue, dataWrap, "Producer").start();
  }
 
}

BlockingQueue has five implementation classes:

ArrayBlockingQueue BlockingQueue queue implemented based on array

LinkedBlockingQueue BlockingQueue queue implemented based on linked list

The elements in PriorityBlockingQueue need to implement the Comparable interface, and the sorting of the elements is customized according to the Comparator.

SynchronousQueue Synchronous queue requires that access operations to the queue must be performed alternately.

DelayQueue The collection elements must implement the Delay interface. The elements in the queue are sorted according to the return value of the Delay interface method getDelay().

The above is the entire content of this article. I hope it will be helpful to everyone's learning. I also hope that everyone will support the PHP Chinese website.

For more detailed articles related to Java thread communication, please pay attention to the PHP Chinese website!