Java生产者消费者模式

在Java中,负责产生数据的模块是生产者,负责使用数据的模块是消费者. 生产者消费者解决数据的平衡问题,即先有数据然后才能使用,没有数据时,消费者需要等待。

生产-消费:操作值

package com.wkcto.producerdata;

/**
 * 定义一个操作数据的类
 * 北京乐学网老崔
 */
public class ValueOP {
    private String value = "";

    //定义方法修改value字段的值
    public void setValue(){
        synchronized ( this ){
            //如果value值不是""空串就等待
            while ( !value.equalsIgnoreCase("")){
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //如果value字段值是容串, 就设置value字段的值
            String value = System.currentTimeMillis() + " - " + System.nanoTime();
            System.out.println("set设置的值是: " + value);
            this.value = value;
//            this.notify();        //在多生产者多消费者环境中,notify()不能保证是生产者唤醒消费者,如果生产者唤醒的还是生产者可能会出现假死的情况
            this.notifyAll();
        }
    }

    //定义方法读取字段值
    public void getValue(){
        synchronized (this){
            //如果value是空串就等待
            while ( value.equalsIgnoreCase("")){
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //不是空串,读取 字段值
            System.out.println("get的值是: " + this.value);
            this.value = "";
            this.notifyAll();
        }
    }
}

package com.wkcto.producerdata;

/**
 * 定义线程类模拟生产者
 * 北京乐学网老崔
 */
public class ProducerThread extends Thread {
    //生产者生产数据就是调用ValueOP类的setValue方法给value字段赋值
    private ValueOP obj;

    public ProducerThread(ValueOP obj) {
        this.obj = obj;
    }

    @Override
    public void run() {
        while (true){
            obj.setValue();
        }
    }
}

package com.wkcto.producerdata;

/**
 * 定义线程类模拟消费者
 * 北京乐学网老崔
 */
public class ConsumerThread extends Thread {
    //消费者使用数据, 就是使用ValueOP类的value字段值
    private ValueOP obj;

    public ConsumerThread(ValueOP obj) {
        this.obj = obj;
    }

    @Override
    public void run() {
        while (true){
            obj.getValue();
        }
    }
}

package com.wkcto.producerdata;

/**
 * 测试多生产,多消费的情况
 * 北京乐学网老崔
 */
public class Test2 {
    public static void main(String[] args) {
        ValueOP valueOP = new ValueOP();

        ProducerThread p1 = new ProducerThread(valueOP);
        ProducerThread p2 = new ProducerThread(valueOP);
        ProducerThread p3 = new ProducerThread(valueOP);
        ConsumerThread c1 = new ConsumerThread(valueOP);
        ConsumerThread c2 = new ConsumerThread(valueOP);
        ConsumerThread c3 = new ConsumerThread(valueOP);

        p1.start();
        p2.start();
        p3.start();
        c1.start();
        c2.start();
        c3.start();
    }
}

操作栈

使生产者把数据存储到List集合中, 消费者从List集合中取数据,使用List集合模拟栈。

package com.wkcto.produerstack;

import java.util.ArrayList;
import java.util.List;

/**
 * 模拟栈
 * 北京乐学网老崔
 */
public class MyStack {
    private List list = new ArrayList();        //定义集合模拟栈
    private static final  int MAX = 3;      //集合的最大容量

    //定义方法模拟入栈
    public synchronized void push(){
        //当栈中的数据已满 就等待
        while ( list.size() >= MAX ){
            System.out.println(Thread.currentThread().getName() + " begin  wait....");
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        String data = "data--" + Math.random();
        System.out.println( Thread.currentThread().getName() + "添加了数据: " + data);
        list.add(data);
//        this.notify();            //当多个生产者多个消费者时,使用notify()可能会出现假死的情况
        this.notifyAll();
    }

    //定义方法模拟出栈
    public synchronized void pop(){
        //如果没有数据就等待
        while ( list.size() == 0 ){
            try {
                System.out.println(Thread.currentThread().getName() + " begin  wait....");
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println( Thread.currentThread().getName() + "出栈数据:" + list.remove(0) );
        this.notifyAll();
    }
}

package com.wkcto.produerstack;

/**
 * 生产者线程
 * 北京乐学网老崔
 */
public class ProduerThread extends  Thread {
    private  MyStack stack;

    public ProduerThread(MyStack stack) {
        this.stack = stack;
    }

    @Override
    public void run() {
        while (true){
            stack.push();
        }
    }
}

package com.wkcto.produerstack;

/**
 * 消费线程
 * 北京乐学网老崔
 */
public class ConsumerThread extends  Thread {
    private  MyStack stack;

    public ConsumerThread(MyStack stack) {
        this.stack = stack;
    }

    @Override
    public void run() {
        while (true){
            stack.pop();
        }
    }
}

package com.wkcto.produerstack;


/**
 * 测试多生产多消费的情况
 * 北京乐学网老崔
 */
public class Test02 {
    public static void main(String[] args) {
        MyStack stack = new MyStack();

        ProduerThread  p = new ProduerThread(stack);
        ProduerThread  p2 = new ProduerThread(stack);
        ProduerThread  p3 = new ProduerThread(stack);
        ConsumerThread c1 = new ConsumerThread(stack);
        ConsumerThread c2 = new ConsumerThread(stack);
        ConsumerThread c3 = new ConsumerThread(stack);
        p.setName("生产者1号");
        p2.setName("生产者2号");
        p3.setName("生产者3号");
        c1.setName("消费者1号");
        c2.setName("消费者2号");
        c3.setName("消费者3号");

        p.start();
        p2.start();
        p3.start();
        c1.start();
        c2.start();
        c3.start();

    }
}

Java CAS多线程 Java中利用管道实现线程间的通讯