java生产消费者算法

 

今天的db2培训仍然没有太多的实践，围绕考试来的，闲着也是闲着，想到我们后续业务中使用多线程的地方比较多。于是想起那个经典的案例：生产者和消费者

关于这个案例的原理，就不多说了。主要涉及到临界资源互斥锁的使用、wait和notify操作，还有就是线程sleep。关于几个操作的区别，我会写在代码的注释中。这和我的工作习惯有关系，不喜欢写文档（敏捷开发认为代码是最好的文档，^_^，我的代码没有重构，完成了功能就贴上来了，当然不符合敏捷的要求了^_^，见谅）

请看代码：

1、生产和消费的产品抽象类

//预留
public abstract class Product {

    public String name;
    public abstract String toString();

}

2、一个具体的产品类

public class AProduct extends Product {

    public AProduct(String pname) {
        super();
        name = pname;
        // TODO Auto-generated constructor stub
    }

    public String toString() {
        // TODO Auto-generated method stub
        return name;
    }
}

 

3、容器类（仓库）

import java.util.ArrayList;

/*

 * 存放生产者和消费者的产品队列

 * */
public class Container {

    private ArrayList arrList = new ArrayList();
    private int LENGTH = 10;

    public boolean isFull() {
        return arrList.size() == LENGTH;
    }

    public boolean isEmpty() {
        return arrList.isEmpty();
    }

    /* 如果此处不加synchronized锁，那么也可以再调用push的地方加锁
    * 既然此处加了锁，那么再别的地方可以不加锁
    */
    public synchronized void push(Object o) {
        arrList.add(o);
    }

    // 如果此处不加synchronized锁，那么也可以再调用push的地方加锁
    public synchronized Object pop() {
        Object lastOne = arrList.get(arrList.size() - 1);
        arrList.remove(arrList.size() - 1);
        return lastOne;
    }
}

 

4、休息一会，生产者和消费者都要休息，因此作为抽象基类

public abstract class Sleep {

    public void haveASleep() throws InterruptedException {
        Thread.sleep((long) (Math.random() * 3000));
    }

}

 

// sleep让出cpu时间给其他线程执行.也许你会问，既然我已经wait了，为什么还要sleep?
 // wait()的作用就是阻塞当前线程并且释放对象的锁给别人（一般是等待队列中的第一个线程）
 // Thead.sleep()也是阻塞当前线程，但不释放锁。
 // sleep()方法是使线程停止一段时间的方法。
 // 在sleep 时间间隔期满后，线程不一定立即恢复执行。
 // 这是因为在那个时刻，其它线程可能正在运行而且没有被调度为放弃执行，除非 (a)“醒来”的线程具有更高的优先级。
 // (b)正在运行的线程因为其它原因而阻塞。 wait()是线程交互时，如果线程对一个同步对象x
 // 发出一个wait()调用，该线程会暂停执行，被调对象进入等待状态，直到被唤醒或等待时间到。
 // 当调用wait()后，线程会释放掉它所占有的“锁标志”，从而使线程所在对象中的其它synchronized数据可被别的线程使用。
 // waite()和notify()因为会对对象的“锁标志”进行操作，所以它们必须在synchronized函数或synchronized

 

/*
 * 消费者线程
 **/

public class Consumer extends Sleep implements Runnable {

    private Container contain = null;

    public Consumer(Container contain) {
        super();
        this.contain = contain;
    }

    public void run() {
        // TODO Auto-generated method stub
        while (true) {
            synchronized (contain) {
                while (contain.isEmpty()) {
                    try {
                        contain.wait();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
            consume();//消费
            try {
                haveASleep();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            synchronized (contain) {
                contain.notify();
            }
        }
    }

    private void consume() {
        Product a = (AProduct) contain.pop();
        System.out.println("消费了一个产品" + a.toString());
    }
}

 

/*

 * 生产者线程

 * */

public class Producator extends Sleep implements Runnable {

    private Container contain = null;

    public Producator(Container contain) {
        super();
        this.contain = contain;
    }

    public void run() {
        // TODO Auto-generated method stub
        while (true) {
            synchronized (contain) {
                while (contain.isFull()) {
                    try {
                        contain.wait();// 阻塞当前线程,当前线程进入等待队列。这个时候只有等待别的线程来唤醒自己了。
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
            producer();// 生产一个产品
            try {
                haveASleep();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            synchronized (contain) {
                contain.notify();// 唤醒等待队列中正在等待的第一个线程，让其执行。
            }
        }
    }

    public void producer() {
        Product aProduct = new AProduct("pp:" + String.valueOf(Math.random()));
        System.out.println("生产了一个产品：" + aProduct.toString());
        contain.push(aProduct);
    }
}

 

 

 

5、写一个测试吧：

public class TestMain {

    /**
     * @param args
     */

    public static void main(String[] args) {
        // TODO Auto-generated method stub
        Container contain = new Container();
        Producator p = new Producator(contain);
        Consumer c = new Consumer(contain);
        Thread pt = new Thread(p);
        Thread ct = new Thread(c);
        pt.start();
        ct.start();
    }
}

 

好了，看看执行结果吧：

生产了一个产品：pp:0.5010546528255287

生产了一个产品：pp:0.08173509855014827

消费了一个产品pp:0.5010546528255287

生产了一个产品：pp:0.7022996270428004