原 阻塞队列 BlockingQueue复习

版权声明：本文为博主原创文章，请尊重他人的劳动成果，转载请附上原文出处链接和本声明。
本文链接：https://www.91mszl.com/zhangwuji/article/details/1147

题目：一个线程生产，一个线程消费。5秒钟后停止。

代码如下：

package com.mszl.thread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 功能：阻塞队列复习：一个线程生产，一个线程消费。
 * 备注：更多资料请访问 http://www.91mszl.com
 * @author bobo teacher
 */
class QueueData{
	
	public volatile boolean flag=true;
	BlockingQueue<String> bk=new ArrayBlockingQueue<>(5);
	AtomicInteger ai=new AtomicInteger();
	
	// 生产
	public void produce() throws InterruptedException{
		String dt=null;
		while(flag){
			dt=String.valueOf(ai.incrementAndGet());
			boolean result=bk.offer(dt, 2, TimeUnit.SECONDS); // 插入值
			if(result){
				System.out.println(Thread.currentThread().getName() + "\t" + "插入成功" + dt);
			} else{
				System.out.println(Thread.currentThread().getName() + "\t" + "插入失败" + dt);
			}
			
			Thread.sleep(1000);
		}
		System.out.println(Thread.currentThread().getName() + "\t" + "停止了。。。");
	}
	
	// 消费
	public void consume() throws InterruptedException{
		while(flag){
			String result=bk.poll(2, TimeUnit.SECONDS); // 移除值
			if(null==result){
				flag=false;
				System.out.println(Thread.currentThread().getName() + "\t" + "超过2s没有取到值，退出");
				System.out.println();
				return;
			}
			System.out.println(Thread.currentThread().getName() + "\t" + "移除成功" + result);
		}
		System.out.println(Thread.currentThread().getName() + "\t" + "消费完成");
	}
	
	public void stop(){
		flag=false;
	}
	
}

public class BlockingQueueDemo1 {
	
	public static void main(String[] args) {
		QueueData qd=new QueueData();
		
		// t1线程
		new Thread(()->{
			try {
				System.out.println("生产者线程启动");
				qd.produce();
			} catch (Exception e) {
				e.printStackTrace();
			}
		},"produce").start();
		
		// t2线程
		new Thread(()->{
			try {
				System.out.println("消费者线程启动");
				System.out.println();
				qd.consume();
			} catch (Exception e) {
				e.printStackTrace();
			}
		},"consume").start();
		
		// 暂停5s
		try {
			Thread.sleep(5000);
			System.out.println("5s时间到。。。");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		// 停止
		qd.stop();
	}
	

}

执行结果：

生产者线程启动
消费者线程启动

produce	插入成功1
consume	移除成功1
produce	插入成功2
consume	移除成功2
produce	插入成功3
consume	移除成功3
produce	插入成功4
consume	移除成功4
produce	插入成功5
consume	移除成功5
5s时间到。。。
produce	停止了。。。
consume	超过2s没有取到值，退出