生产者和消费者模型

 三个生产者向一个内存容器中产生数据，另外三个消费者从容器中消费数据。

public class Main {

public static void main(String[] args) throws Exception {
//内存缓冲区
        BlockingQueue<Data> queue = new LinkedBlockingQueue<Data>(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);

//创建线程池运行,这是一个缓存的线程池，可以创建无穷大的线程，没有任务的时候不创建线程。空闲线程存活时间为60s（默认值）
        ExecutorService cachePool = Executors.newCachedThreadPool();
        cachePool.execute(p1);
        cachePool.execute(p2);
        cachePool.execute(p3);
        cachePool.execute(c1);
        cachePool.execute(c2);
        cachePool.execute(c3);

try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        p1.stop();
        p2.stop();
        p3.stop();

try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }        
//        cachePool.shutdown(); 
//        cachePool.shutdownNow();
    }

}

 

public class Provider implements Runnable{

//共享缓存区
private BlockingQueue<Data> queue;

//多线程间是否启动变量，有强制从主内存中刷新的功能。即时返回线程的状态
private volatile boolean isRunning = true;

//id生成器,因为多个producer共享这个变量
private static AtomicInteger count = new AtomicInteger();

//随机对象
private static Random r = new Random(); 

public Provider(BlockingQueue queue){
this.queue = queue;
    }

public void run() {
while(isRunning){
try {
//随机休眠0 - 1000 毫秒 表示获取数据(产生数据的耗时) 
                Thread.sleep(r.nextInt(1000));
//获取的数据进行累计...
int id = count.incrementAndGet();
//比如通过一个getData方法获取了
                Data data = new Data(Integer.toString(id), "数据" + id);
                System.out.println("当前线程:" + Thread.currentThread().getName() + ", 获取了数据，id为:" + id + ", 进行装载到公共缓冲区中...");
if(!this.queue.offer(data, 2, TimeUnit.SECONDS)){
                    System.out.println("提交缓冲区数据失败....");
//do something... 比如重新提交
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

public void stop(){
this.isRunning = false;
    }

}

public class Consumer implements Runnable{

private BlockingQueue<Data> queue;

public Consumer(BlockingQueue queue){
this.queue = queue;
    }

//随机对象
private static Random r = new Random(); 

public void run() {
while(true){
try {
//获取数据
                Data data = this.queue.take();
//进行数据处理。休眠0 - 1000毫秒模拟耗时
                Thread.sleep(r.nextInt(1000));
                System.out.println("当前消费线程：" + Thread.currentThread().getName() + "， 消费成功，消费数据为id: " + data.getId());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public final class Data {

private String id;
private String name;

public Data(String id, String name){
this.id = id;
this.name = name;
    }

public String getId() {
return id;
    }

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

public String getName() {
return name;
    }

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

    @Override
public String toString(){
return "{id: " + id + ", name: " + name + "}";
    }

}