java multithreaded summary-producer and consumer of synchronization

1. Producer-consumer wait & notify version

Code example:

/**
 * bounded-buffer problem
 * wait&notify
 */
package com.bernardlowe.concurrent.t04;

import java.util.LinkedList;
import java.util.concurrent.TimeUnit;

public class TestContainer01<E> {

	private final LinkedList<E> list = new LinkedList<>(); // LinkedList simulates production queues
	private final int MAX = 10; // Maximum capacity
	private int count = 0;
	
	public synchronized int getCount(){
		return count;
	}
	
	// Producer
	public synchronized void put(E e){
		while(list.size() == MAX){
			try {
				this.wait();
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
		}
		
		list.add(e);
		count++;
		this.notifyAll();
	}

	// Consumer
	public synchronized E get(){
		E e = null;
		while(list.size() == 0){
			try{
				this.wait();
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
		}
		e = list.removeFirst();
		count--;
		this.notifyAll();
		return e;
	}
	
	public static void main(String[] args) {
		final TestContainer01<String> c = new TestContainer01<>();
		
		// Consumer threads consume
		for(int i = 0; i < 10; i++){
			new Thread(new Runnable() {
				@Override
				public void run() {
					for(int j = 0; j < 5; j++){
						System.out.println(c.get());
					}
				}
			}, "consumer"+i).start();
		}
		try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		// Producer threads for production
		for(int i = 0; i < 2; i++){
			new Thread(new Runnable() {
				@Override
				public void run() {
					for(int j = 0; j < 25; j++){
						c.put("container value " + j); 
					}
				}
			}, "producer"+i).start();
		}
	}
	
}

Thinking: why is producer code judged by while instead of if?
The reason is that if you use if, it will lead to a false wake-up state, that is to say, if only once judged, the Tao wait() statement may sleep and release the lock. When awakened again, the statement after wait() will continue to be executed without judging whether the queue is full. The application of wait/notify and while can avoid multi-threading concurrent judgment of logic failure. Question.

2 Producer Consumer Condition Version

The Condition version does not use wait/notify, but uses the newCondition() method of ReentrantLock to get the Condition condition. It determines when to lock or unlock the condition through the condition, such as waking up, which is more intuitive than wait/notify.

/**
 * bounded-buffer problem
 * Re-entry Lock-Conditions
 */
package com.bernardlowe.concurrent.t04;

import java.io.IOException;
import java.util.LinkedList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TestContainer02<E> {

	private final LinkedList<E> list = new LinkedList<>();
	private final int MAX = 10;
	private int count = 0;
	
	private Lock lock = new ReentrantLock();
	private Condition producer = lock.newCondition();
	private Condition consumer = lock.newCondition();
	
	public int getCount(){
		return count;
	}
	
	public void put(E e){
		lock.lock();
		try {
			while(list.size() == MAX){
				System.out.println(Thread.currentThread().getName() + " Wait for...");
				// Enter the waiting queue. Release the lock tag.
				// With the help of conditions, enter the waiting queue.
				producer.await();
			}
			System.out.println(Thread.currentThread().getName() + " put . . . ");
			list.add(e);
			count++;
			// With the help of conditions, wake up all consumers.
			consumer.signalAll();
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		} finally {
			lock.unlock();
		}
	}
	
	public E get(){
		E e = null;

		lock.lock();
		try {
			while(list.size() == 0){
				System.out.println(Thread.currentThread().getName() + " Wait for...");
				// With the help of conditions, consumers enter the waiting queue
				consumer.await();
			}
			System.out.println(Thread.currentThread().getName() + " get . . . ");
			e = list.removeFirst();
			count--;
			// Wake up all producers with the help of conditions
			producer.signalAll();
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		} finally {
			lock.unlock();
		}
		
		return e;
	}
	
	public static void main(String[] args) {
		final TestContainer02<String> c = new TestContainer02<>();
		for(int i = 0; i < 10; i++){
			new Thread(new Runnable() {
				@Override
				public void run() {
					for(int j = 0; j < 5; j++){
						System.out.println(c.get());
					}
				}
			}, "consumer"+i).start();
		}
		try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		for(int i = 0; i < 2; i++){
			new Thread(new Runnable() {
				@Override
				public void run() {
					for(int j = 0; j < 25; j++){
						c.put("container value " + j); 
					}
				}
			}, "producer"+i).start();
		}
	}
	
}

Tags: Java

Posted on Tue, 08 Oct 2019 02:33:46 -0700 by cjdesign