java 多线程（生产消费者模式）

备注说明：

该内容拷贝自网络

Introduce BlockingQueue:

 

BlockingQueue only support java 1.5 or higher,and it's defined as below.

 

public interface BlockingQueue<E>
extends Queue<E>

 

As the definition,the parameters E is the type of elements helding in this collection. A Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.

 

Summay of BlockingQueue methods:(as the figure)

 

 

	Throws exception	Special value	Blocks	Times out
Insert	add(e)	offer(e)	put(e)	offer(e, time, unit)
Remove	remove()	poll()	take()	poll(time, unit)
Examine	element()	peek()	not applicable	not applicable

Pls Note: A BlociingQueue does not accept null elements. Implementations throw nullPointerException on attempts to add, put or offer a null. A null is used as a sentinel value to indicate failure of poll operations.

 

Primarily designed of BlockingQueue:

 

BlockingQueue implementations are designed to be used primarily for producer-comsumer queues, but addtionally support the Collection interface. So,for example, it is possiable to remove an arbitrary element from a queue using remove(i). However, such operations are in general not performed very efficiently, and are intended for only occasional use, such as when a queued message is cancelled.

 

Of course BlockingQueue implementations are thread-safe. All queeing methods achieve their effects atomically using internal locks or other form of concurrency control. However, the bulk Collection operations addAll, containsAll, retainAll and removeAll are not necessarily performed atomically unless specified otherwise in an implementation. So it is possible, for example for addAll(c) to fail (throw an exception) after adding only some of the elements in c.

 

Sample of use BlockingQueue to implement Producer-Consumer Question :

 

1. Create a producer.

Producer.java

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

    private BlockingQueue queue;

    // put producer in queue
    public Producer(BlockingQueue q) {
        this.queue = q;
    }

    public void run() {
        try {
            while (true) {
                // wait for a random time
                Thread.sleep((int) (Math.random() * 3000));
                queue.put(produce());
            }
        } catch (InterruptedException ex) {
            // catch exception
        }
    }

    public String produce() {
        List<String> product = new ArrayList<String>();
        String producProduct = null;
        product.add("Car");
        product.add("Phone");
        product.add("Plane");
        product.add("Computer");
        boolean flag = true;
        int getProduct = (int) (Math.random() * 3);
        while (flag) {
            if (getProduct <= product.size()) {
                producProduct = product.get(getProduct);
                flag = false;
            }
        }
        System.out.println("Producer producing " + producProduct);
        return producProduct;
    }

}

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

	private BlockingQueue queue;

	// put producer in queue
	public Producer(BlockingQueue q) {
		this.queue = q;
	}

	public void run() {
		try {
			while (true) {
				// wait for a random time
				Thread.sleep((int) (Math.random() * 3000));
				queue.put(produce());
			}
		} catch (InterruptedException ex) {
			// catch exception
		}
	}

	public String produce() {
		List<String> product = new ArrayList<String>();
		String producProduct = null;
		product.add("Car");
		product.add("Phone");
		product.add("Plane");
		product.add("Computer");
		boolean flag = true;
		int getProduct = (int) (Math.random() * 3);
		while (flag) {
			if (getProduct <= product.size()) {
				producProduct = product.get(getProduct);
				flag = false;
			}
		}
		System.out.println("Producer producing " + producProduct);
		return producProduct;
	}

}

 

2.Create a consumer.

Consumer.java

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {

    private  BlockingQueue queue;

    public Consumer(BlockingQueue q){
        this.queue=q;
    }
    @Override
    public void run() {
        // TODO Auto-generated method stub
        try {
            while (true) {
                // wait for a random time
                Thread.sleep((int) (Math.random() * 3000));
                consume(queue.take());
            }
        } catch (InterruptedException ex) {
        }

    }

    public void consume(Object x) {
        System.out.println("Consumer buy product "+x.toString());
    }
}

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {

	private  BlockingQueue queue;

	public Consumer(BlockingQueue q){
		this.queue=q;
	}
	@Override
	public void run() {
		// TODO Auto-generated method stub
		try {
			while (true) {
				// wait for a random time
				Thread.sleep((int) (Math.random() * 3000));
				consume(queue.take());
			}
		} catch (InterruptedException ex) {
		}

	}

	public void consume(Object x) {
		System.out.println("Consumer buy product "+x.toString());
	}
}

 

3. Test Client

TestClient.java

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;


public class TestClient {

    /**
     * @param args
     */
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        new TestClient();
    }

    public TestClient() {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(1);
        Thread producerThread = new Thread(new Producer(queue));
        Thread consumerThread = new Thread(new Consumer(queue));
        producerThread.start();
        consumerThread.start();
    }
}

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;


public class TestClient {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		new TestClient();
	}

	public TestClient() {
		BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(1);
		Thread producerThread = new Thread(new Producer(queue));
		Thread consumerThread = new Thread(new Consumer(queue));
		producerThread.start();
		consumerThread.start();
	}
}

 

 

Summay:

This sample introduces BlockingQueue Class and using it to implement the Producer-Consumer Question.