JAVA消费者与生产者模型的几个不同实现

       java的生产者与消费者模型对与理解JAVA的锁机制，线程安全和并发编程来说是一个比较经典的例子，下面就将我遇到过的几种不同的实现跟大家分享一下。

1.使用synchronized关键字

         synchronized来 加同步锁，保证线程安全，synchronized锁自1.6后做了很大的优化，对于一般情况下的同步，用此锁已经足已应付。

public class ProducerAndConsumer {
	public static void main(String[] args) {
		SyncStack ss = new SyncStack();
		// 生产者线程
		Thread t1 = new Thread(new Producer(ss));
		// 消费者线程
		Thread t2 = new Thread(new Consumer(ss));
		t1.start();
		t2.start();
	}
}

class Entity {
	int id;

	Entity(int id) {
		this.id = id;
	}

	public String toString() {
		return " entity :" + id;
	}
}

class SyncStack {

	int index = 0;

	Entity[] arrWT = new Entity[6];

	public synchronized void push(Entity wt) {
		if (index == arrWT.length)
			try {
				this.wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		this.notify();
		arrWT[index] = wt;
		index++;
	}

	public synchronized void pop(Entity wt) {
		if (index == 0) {
			try {
				this.wait();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		this.notify();
		index--;
	}
}

class Producer implements Runnable {
	SyncStack ss = null;

	Producer(SyncStack ss) {
		this.ss = ss;
	}

	public void run() {
		for (int i = 0; i < 20; i++) {
			Entity e = new Entity(i);
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
			ss.push(e);
			System.out.println("生产了" + e);
		}
	}
}

class Consumer implements Runnable {
	SyncStack ss = null;

	public Consumer(SyncStack ss) {
		this.ss = ss;
	}

	public void run() {
		for (int i = 0; i < 20; i++) {
			Entity wt = new Entity(i);
			try {
				// 随机休眠一段时间
				Thread.sleep((int) (Math.random() * 3000));
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			ss.pop(wt);
			System.out.println("消费了" + wt);
		}
	}
}

     运行结果 ：

   

生产了 entity :1
消费了 entity :1
生产了 entity :2
生产了 entity :3
消费了 entity :2
生产了 entity :4
生产了 entity :5
消费了 entity :3

 2.使用ReentrantLock和Condition

<!-- Generated by javadoc (build 1.6.0-beta2) on Fri Mar 09 12:53:21 CST 2007 -->

<noscript></noscript>

          ReentrantLock是可重入锁，自JDK1.5添加，  一个可重入的互斥锁 Lock，它具有与使用 synchronized 方法和语句所访问的隐式监视器锁相同的一些基本行为和语义，但功能更强大。

          Condition 条件（也称为条件队列 或条件变量 ）,它 替代了 Object 监视器方法的使用。（具体自己参考API）

  

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


public class ProducerAndConsumer {
	final Lock lock = new ReentrantLock();   //可重入锁
	final Condition notFull = lock.newCondition();    //非满条件
	final Condition notEmpty = lock.newCondition();  //非空条件
	
	final Object[] items = new Object[100];
	int putptr , takeptr , count;
	
	public void put(Object o) throws InterruptedException{
		lock.lock();      //锁定操作，保证线程安全
		try{
			while(count == items.length){
				notFull.wait();
			}
			items[putptr] = o;
			if(++putptr == items.length)
				putptr = 0;
			
			++count;
			notEmpty.signal();
			
		}finally{
				lock.unlock();
				
		}
	}
	
	public Object take() throws InterruptedException{
		lock.lock();    //
		try {
			while(count == 0)
				notEmpty.await();
			Object o = items[takeptr];
			if(takeptr == items.length)
				takeptr = 0;
			--count;
			notFull.signal();
			return o;
		} finally{
			lock.unlock();
		}
	}
	
	public static void main(String[] args) {
		final ProducerAndConsumer pac = new ProducerAndConsumer();
		//模拟消费者
                new Thread(new Runnable(){
			public void run() {
				try {
					for(;;){
						pac.take();
						System.out.println(Thread.currentThread().getName());
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}).start();
		//模拟生产者
		new Thread(new Runnable(){
			public void run() {
				try {
					for(int i=0;i<10;i++){
						Object o = new Object();
						pac.put(o);
						System.out.println(Thread.currentThread().getName());
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}).start();
	}
}

3.使用semaphore

<!-- Generated by javadoc (build 1.6.0-beta2) on Fri Mar 09 12:53:11 CST 2007 -->

<noscript></noscript>

         semaphore   一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire(),然后再获取该许可。每个relase  添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目.

import java.util.concurrent.Semaphore;

public class TestSemaphore {

	final static BoundBuffer buffer = new BoundBuffer();

	public static void main(String[] args) {
		// 各启动3个线程
		for (int i = 0; i < 3; i++) {
			new Thread(new Producer()).start();
			new Thread(new Consumer()).start();
		}
	}

	static class BoundBuffer {
		final Semaphore notFull = new Semaphore(10);

		final Semaphore notEmpty = new Semaphore(0);

		// 传入参数为1，为了模拟一个互斥锁
		final Semaphore mutex = new Semaphore(1);

		Object[] items = new Object[10];

		int putptr, takeptr;

		static int count, put, take;

		public void put(Object o) {

			try {
				// 先拿到put的许可，在拿互斥锁；同时减少一个许可
				notFull.acquire();
				// 加互斥锁，保证线程安全
				mutex.acquire();
				items[putptr] = o;
				if (++putptr == items.length) {
					putptr = 0;
				}
				++count;
				++put;
				System.out.println("共生产了 " + put + " 个,还剩 " + count + " 个");
			} catch (InterruptedException e) {
				e.printStackTrace();
			} finally {
				// 释放互斥锁
				mutex.release();
				// 注意此处是notEmpty.release()，该许可+1，表示新增了一个产品
				// 消费者线程还可以来消费
				notEmpty.release();
			}
		}

		public Object take() {
			try {
				notEmpty.acquire();
				mutex.acquire();
				Object o = items[takeptr];
				if (++takeptr == items.length)
					takeptr = 0;
				--count;
				++take;
				System.out.println("共消费了 " + take + " 个,还剩 " + count + " 个");
				return o;
			} catch (InterruptedException e) {
				e.printStackTrace();
				return null;
			} finally {
				mutex.release();
				// 此处notFull.release()，表示新增了一个put的许可，生产者可以继续put
				notFull.release();
			}
		}
	}

	static class Consumer implements Runnable {

		public void run() {
			while (true) {
				buffer.take();
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}

	static class Producer implements Runnable {

		public void run() {
			while (true) {
				Object o = new Object();
				buffer.put(o);
				try {
					Thread.sleep(500);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}

		}

	}
}

 运行结果 ：

 

共生产了 1 个,还剩 1 个
共生产了 2 个,还剩 2 个
共生产了 3 个,还剩 3 个
共消费了 1 个,还剩 2 个
共消费了 2 个,还剩 1 个
共消费了 3 个,还剩 0 个
共生产了 4 个,还剩 1 个
共生产了 5 个,还剩 2 个
.......

 

还有其他的一些实现方式，欢迎大家分享