阻塞队列的使用Demo

野性酷女 2023-07-22 09:11 24阅读 0赞

谈到阻塞队列需要谈到以往的lock中的await和signel，以往线程的状态需要我们手动修改来完成线程间的调度，如wait 和 notify，此时BlockingQueue阻塞队列横空出世，即线程状态的挂起和唤醒状态都由该队列内部完成，你用即可，该队列在多线程场景下，可以很好的实现生产和消费的模式，其实该消息中间件（mq kafka等）底层用的就是该队列。

BlockingQueue核心的方法就是offer和poll，即放和拿，这两个方法不传参意为拿不到（放不进）消息我就一种等，直到拿到（放进去）消息；要是传参，即我也是有限度的，等了那么久还没放入或拿到那我就停了。

package com.reentralock;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    class MySource{//资源类
    	
    	private volatile boolean flag = true;//全网开关
    	private AtomicInteger atomicInteger =new AtomicInteger();
    	
    	private BlockingQueue<String> queue = null;//留着给别人用的队列，口子要大，传它的子类全可以
    
    	public MySource(BlockingQueue<String> queue) {
    		super();
    		this.queue = queue;
    		System.out.println(queue.getClass().getName());//日志打印细粒度
    	}
    	
    	public void myProduct() throws Exception{
    		
    		String data = "";
    		boolean result;
    		//1.判断
    		while(flag){
    			data = atomicInteger.incrementAndGet()+"";
    			result = queue.offer(data, 2, TimeUnit.SECONDS);//2s加不进去不加了
    			
    			if(result){
    				System.out.println("生产cake"+data+"成功");
    			}else{
    				System.out.println("生产cake"+data+"失败");
    			}
    			TimeUnit.SECONDS.sleep(1);//生产间隔1s
    		}
    		
    		System.out.println("boss叫停，别再生产了！");
    	}
    	
    	public void myConsumer() throws Exception{
    		
    		String data = "";
    		
    		//1.判断
    		while(flag){
    			data = queue.poll(2, TimeUnit.SECONDS);//两秒钟拿不到，我就不拿了
    			 if(null == data || "".equals(data)){
    				 System.out.println("队列中无cake可消费");
    				 flag = false;
    				 System.out.println("2s未拿到cake，停止拿");
    				 return;//跳出方法
    			 }else{
    				 System.out.println("消费cake"+data+"成功");
    			 }
    		}
    		System.out.println("boss叫停，别再生产了！");
    	}
    	
    	public void stop(){//boss让你停 你就停
    		flag = false;
    	}
    }
    /**
     * 队列高级版，阻塞队列将挂起、唤醒 内部实现，你用即可
     * 
     * 模仿场景，多线程下，只需一个标志位，控制整个消费和生产的实现，消费者和生产者的来源于去处需要队列暂存
     * 多线程下标配：Atomic、BlockQueue、CAS、Volatile =abcv
     * 
     * 多线程
     * 上联：线程操作资源类
     * 下联：判断干活再通知（注意虚假唤醒）= 针对lock而言
     * 横批：BlockQueue属顶级
     * 
     * @author 英俊
     *
     */
    public class BlockQueueDemo {
    
    	public static void main(String[] args) {
    		
    		MySource mySource =new MySource(new ArrayBlockingQueue<>(10));
    		
    		new Thread(()->{
    			System.out.println(Thread.currentThread().getName()+"\t 生产线程启动");
    			try {
    				mySource.myProduct();
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}, "product").start();
    		
    		new Thread(()->{
    			System.out.println(Thread.currentThread().getName()+"\t 消费线程启动");
    			try {
    				mySource.myConsumer();
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}, "consumer").start();
    		
    		
    		try {
    			TimeUnit.SECONDS.sleep(5);
    			System.out.println("boss 5s叫停！！");
    			mySource.stop();//5s main主线程叫停！！
    		} catch (InterruptedException e1) {
    			// TODO Auto-generated catch block
    			e1.printStackTrace();
    		}
    	}
    
    }

放心：一直poll拿不到消息，不会占用cpu资源。