生产消费模式

1、生产者消费者模型

      经典的多线程模型；

      通常有两类线程：若干生产者线程，若干消费者线程；

      生产者线程：提交用户请求；

      消费者线程：处理生产者提交的任务，在生产和消费之间通过共享内存缓存区进行数据通信；

 

2、流程图

 

 

3、代码示例

（1）生产者

package net.oschina.tkj.mulitcoding.providerconsumermodel;

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/*
 * 生产消费模型中的生产者
 * 
 * 负责生成数据
 */
public class Provider implements Runnable {

	// 阻塞队列，存储任务数据,使用有界的高并发阻塞队列
	private final ArrayBlockingQueue<Data> queue;

	private static AtomicInteger count = new AtomicInteger(0);

	private final Random r = new Random();

	private boolean isRuning = true;

	public Provider(ArrayBlockingQueue<Data> queue) {
		this.queue = queue;
	}

	@Override
	public void run() {
		if (null == queue) {
			System.out.println("null point exception...");
			return;
		}
		Data d = null;
		while (isRuning) {

			d = new Data();
			d.setId(count.incrementAndGet());
			d.setName(r.nextInt() + " 数据");

			System.out.println("生产者线程：" + Thread.currentThread().getName()
					+ " 生产数据：id=" + d.getId() + "  数据=" + d.getName());

			try {
				if (!queue.offer(d, 2, TimeUnit.SECONDS)) {
					System.out.println("缓存数据的队列已满，等待队列中元素消费后再放入...");
				}
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}

	}

	public boolean stop() {
		isRuning = false;
		return isRuning;
	}

}

 

（2）消费者

package net.oschina.tkj.mulitcoding.providerconsumermodel;

import java.util.concurrent.ArrayBlockingQueue;

/**
 * 生产消费模型中的消费者
 * 
 * 消费数据
 * 
 * @author Freedom
 * 
 */
public class Consumer implements Runnable {

	private final ArrayBlockingQueue<Data> queue;

	public Consumer(ArrayBlockingQueue<Data> queue) {
		this.queue = queue;
	}

	private boolean run = true;

	@Override
	public void run() {

		if (null == queue) {
			System.out.println("queue is null,null point exception ...");
			return;
		}
		Data d = null;
		while (run) {
			try {
				if (0 != queue.size()) {
					// take方法为阻塞方法，如果队列为空则wait等待数据
					d = queue.take();

					System.out.println("当前线程消费数据："
							+ Thread.currentThread().getName() + " 数据id="
							+ d.getId() + " 数据=" + d.getName());
				} else {
					// == 0
					run = false;
				}
				Thread.sleep(500);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

}

 

（3）数据实体类

package net.oschina.tkj.mulitcoding.providerconsumermodel;

/**
 * 数据对象，供生产消费者使用
 * 
 * @author Freedom
 * 
 */
public class Data {

	private int id;
	private String name;

	public int getId() {
		return id;
	}

	public void setId(int id) {
		this.id = id;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

}

 

（4）Main

package net.oschina.tkj.mulitcoding.providerconsumermodel;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Main {

	public static void main(String[] args) throws InterruptedException {

		ArrayBlockingQueue<Data> queue = new ArrayBlockingQueue<>(10); // 长度为10固定容量队列
		// 生产者
		Provider p1 = new Provider(queue);
		Provider p2 = new Provider(queue);
		Provider p3 = new Provider(queue);
		// 消费者
		Consumer c1 = new Consumer(queue);
		Consumer c2 = new Consumer(queue);
		Consumer c3 = new Consumer(queue);

		ExecutorService pool = Executors.newCachedThreadPool();
		pool.execute(p1);
		pool.execute(p2);
		pool.execute(p3);
		pool.execute(c1);
		pool.execute(c2);
		pool.execute(c3);

		TimeUnit.SECONDS.sleep(2);
		p1.stop();
		p2.stop();
		p3.stop();

		pool.shutdown();

	}

}