生产者-消费者

    这学期，应聘的时候有一些是线程相关的，虽然自己对线程编程有点概念，但没有写过经典的例子。放假了有点时候，就想写多线程的例子。

    笔试的题目类似地：一个生产者一次生产10个，满了后通知消费者，然后等待。一个消费者产品有满了就消费。到空时通知生产者，然后等待。

    那时对等待/通知机制没怎么写过，那次笔试应该写的大概对（想法对），但写的wait()和notifyAll()的位置不对。现在有时间就写了这个例子。
    描述：生产者一次生产N个产品，池中达到M就等待，通知等待的消费者。消费者有产品就消费，到没有时就通知生产者，然后等待。

1.生产者：

package net.blogjava.chenlb.multithreaded;

import java.util.List;

/**
 * @author chenlb
 * 
 * 生产者.<br/>
 * 默认产品池大小M=20,产品梯阶大小N=5.在生产过程中,池的大小会超过20,但池中最大应该是M+N-1.
 */
public class Producer implements Runnable {

    /**
     * 池默认大小
     */
    public static final int DEFALUT_SIZE = 20;
    /**
     * 默认一次生产的数量
     */
    public static final int DEFALUT_STEP_SIZE = 5;
    
    private static int PRODUCER_ID = 0;    //生产者号
    
    private List<Product> pool = null;
    private int size = DEFALUT_SIZE;
    private int stepSize = DEFALUT_STEP_SIZE;
    
    private String name = "Producer_"+(++PRODUCER_ID);    //生产者名
    
    private boolean isRun = true;
    
    /**
     * 默认产品池大小20, 默认产品增长梯阶大小5
     */
    public Producer(List<Product> pool) {
        this.pool = pool;
    }

    /**
     * @param pool
     * @param size 池大小
     */
    public Producer(List<Product> pool, int size) {
        this.pool = pool;
        this.size = size;
    }
    
    
    /**
     * @param pool
     * @param size 池大小
     * @param stepSize 一次生产多少
     */
    public Producer(List<Product> pool, int size, int stepSize) {
        this.pool = pool;
        this.size = size;
        this.stepSize = stepSize;
    }

    public void run() {
        // TODO 生产者线程
        //int pi = 0;
        while(isRun) {//&& pi<10
            //pi++;
            synchronized (pool) {    //同步产品池
                if(pool.size() >= size) {
                    try {
                        System.out.println(name+" 等待!");
                        pool.wait();    //同步什么就等待什么,否则抛出java.lang.IllegalMonitorStateException
                    } catch (InterruptedException e) {
                        isRun = false;
                        System.out.println(name+" thread interrupt!");                    
                    }
                } else {
                    
                    for(int i=0; i<stepSize; i++) {    //一次生产stepSize个产品
                        pool.add(product());    //生产产品
                    }
                    System.out.println("产品池中有: "+pool.size());
                    pool.notifyAll();    //通知等待的线程(主要用来通知消费者, 但生产者线程也会通知到)
                }
            }
            
            try {
                System.out.println(name+" 休息1秒!");
                Thread.sleep(1000);    //调试用
            } catch (InterruptedException e) {
                System.out.println(name+" sleep 1s thread interrupt");
            }
        }
        System.out.println(name+" end! pool size: "+pool.size());
    }

    private static int P_ID = 0;
    /**
     * 生产产品
     * @return 产品
     */
    private Product product() {
        String name = "product_"+(++P_ID);
        System.out.println(this.name+" 生产了: "+name);
        return new Production(name);
    }
    
}

2.消费者：

package net.blogjava.chenlb.multithreaded;

import java.util.List;

/**
 * @author chenlb
 * 
 * 消费者
 */
public class Consumer implements Runnable {

    private static int C_ID = 0;    //消费者ID
    
    private List<Product> pool = null;
    private String name = "Consumer_"+(++C_ID);
    private boolean isRun = true;
    public Consumer(List<Product> pool) {
        this.pool = pool;
    }
    
    public void run() {
        // TODO 消费者线程
        //int pi = 0;
        while(isRun) {//&& pi<10
            //pi++;
            synchronized (pool) {
                if(pool.size() < 1) {
                    try {
                        System.out.println(name+" 等待!");
                        pool.notifyAll();    //通知线程(主要是生产者,但也会通知到生产者线程)
                        pool.wait();
                    } catch (InterruptedException e) {
                        isRun = false;
                        System.out.println(name+" thread interrupt!");
                    }
                } else {
                    Product p = pool.remove(0);    //消费
                    printProduct(p);
                    
                }
            }
            try {
                Thread.sleep(1000);    //调试用
            } catch (InterruptedException e) {
                
                System.out.println(name+" sleep 1s thread interrupt");
            }
        }
        System.out.println(name+" end! pool size: "+pool.size());
    }

    private void printProduct(Product p) {
        System.out.println(name+" 消费了: "+p.getName());
    }
}

3.Demo

package net.blogjava.chenlb.multithreaded;

import java.util.LinkedList;
import java.util.List;

/**
 * @author chenlb
 *
 */
public class Sale {

    
    public static void main(String[] args) {
        //链表产品池
        List<Product> pool = new LinkedList<Product>();
        //两个生产者
        Producer p1 = new Producer(pool);
        Producer p2 = new Producer(pool);
        
        Thread tp1 = new Thread(p1);
        Thread tp2 = new Thread(p2);
        
        tp1.start();
        tp2.start();
        
        //两个消费者
        Consumer c1 = new Consumer(pool);
        Consumer c2 = new Consumer(pool);
        
        Thread tc1 = new Thread(c1);
        Thread tc2 = new Thread(c2);
        
        tc1.start();
        tc2.start();
        
        

    }

}

注意：等待时候要用pool.wait()因为同步的是pool。否则会抛出java.lang.IllegalMonitorStateException

^_^

代码下载