多线程生产消费阻塞队列（亲测可用）_线程池消费阻塞队列数据

1.创建队列

package com.xinan.dataservice.util;






import org.apache.poi.ss.formula.functions.T;

import java.util.LinkedList;
public class MyStackThread {
    //初始化链表
    LinkedList<Object> values = new LinkedList<Object>();

    //设计push方法，当列表中的元素大于200时，push线程等待（生产者），否则唤醒全部
    public synchronized void push(Object t){
        while(values.size()>=20){
            try{
                System.out.println("放满了！！！！");
                this.wait();
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
        this.notifyAll();
        values.addLast(t);
    }

    //设计pull方法，当列表中的元素为空时，pull线程等待（消费者），否则唤醒全部
    public synchronized Object pull(){
        while (values.isEmpty()){
            try{
                System.out.println("没有了！！！！");
                this.wait();
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
        this.notifyAll();
        return values.removeLast();
    }

    //获取栈顶元素
    public Object peek(){
        return values.getLast();
    }

}

2.生产者

package com.xinan.dataservice.util;

import org.apache.poi.ss.formula.functions.T;


public class ProducerThread extends Thread{
    //私有属性
    private MyStackThread stack;
    //构造方法
    public ProducerThread(MyStackThread stack,String name){
        super(name);
        this.stack = stack;
    }

    @Override
    public void run(){

//        for(int i=0;i<5;i++){
//            Object c = RandomChar();
//            System.out.println(this.getName() + " push: " + c);
//            stack.push(c);
//            try{
//                Thread.sleep(100);
//            }catch (InterruptedException e){
//                e.printStackTrace();
//            }
//        }
        while(true){
            Object c = RandomChar();
            System.out.println(this.getName() + " push: " + c);
            stack.push(c);
            try{
                Thread.sleep(1000);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }


    //随机生产一个大写字符A-Z，60-95
    public Object RandomChar(){
        Object a ;
        a = (char)(Math.random()*('Z'+1 -'A') + 'A');
        return a;
    }

}

3.消费者

package com.xinan.dataservice.util;


public class ConsumerThread extends Thread{

    //设置私有属性
    private MyStackThread stack;

    //设置带参构造方法
    public ConsumerThread(MyStackThread stack,String name){
        super(name);
        this.stack = stack;
    }

    //线程的启动函数
    @Override
    public void run(){
        while (true){
            Object c = stack.pull();
            System.out.println(this.getName() + " pull: " + c);
            try{
                sleep(100);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
    }
}

4.测试

package com.xinan.dataservice.util;

public class Test {
    public static void main(String[] args) {
        MyStackThread stack = new MyStackThread();
        new ProducerThread(stack,"Producer1").start();
        new ProducerThread(stack,"Producer2").start();
        new ConsumerThread(stack,"Consumer1").start();
        new ConsumerThread(stack,"Consumer2").start();
//        new ConsumerThread(stack,"Consumer3").start();
    }

    public void start(){

    }
}