/**
 * @author fuguangli
 * @description 前沿消费者类
 * @Create date:  2017/3/7
 * @using  EXAMPLE
 */
public class Customer implements Runnable{


  /**
   *     抛出异常  特殊值    阻塞     超时
   插入    add(e)  offer(e)  put(e)  offer(e, time, unit)
   移除    remove()  poll()  take()  poll(time, unit)
   检查    element()  peek()  不可用  不可用

   */
  private BlockingQueue blockingQueue;
  private AtomicInteger count = new AtomicInteger();
  public Customer(BlockingQueue blockingQueue) {
    this.blockingQueue = blockingQueue;
  }

  /**
   * When an object implementing interface Runnable is used
   * to create a thread, starting the thread causes the object's
   * run method to be called in that separately executing
   * thread.
   * 

   * The general contract of the method run is that it may
   * take any action whatsoever.
   *
   * @see Thread#run()
   */
  @Override
  public void run() {
    System.out.println("消费者线程启动...");
    LockFlag.setCustomerRunningFlag(true);
    try {
      while (LockFlag.getProducerRunningFlag()){
        System.out.println(Thread.currentThread().getId()+"I'm Customer.Queue current size="+blockingQueue.size());
        String data = (String) blockingQueue.poll(10, TimeUnit.SECONDS);
        if(data!=null){
          System.out.println(Thread.currentThread().getId()+"*************正在消费数据 data="+data);
        }else{
          //表示超过取值时间，视为生产者不再生产数据
          System.out.println(Thread.currentThread().getId()+"队列为空无数据，请检查生产者是否阻塞");
        }
        Thread.sleep(50);
      }
      System.err.println("消费者程序执行完毕");
    } catch (InterruptedException e) {
      e.printStackTrace();
      System.err.println("消费者程序退出");
      LockFlag.setCustomerRunningFlag(false);//异常退出线程
      Thread.currentThread().interrupt();
    }
  }
}

package com.qysxy.framework.queue;

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

/**
 * @author fuguangli
 * @description 队列生产者类
 * @Create date:  2017/3/7
 * @using    EXAMPLE
 */
public class Producer implements Runnable{


  /**
   *     抛出异常  特殊值    阻塞     超时
   插入  add(e)  offer(e)  put(e)  offer(e, time, unit)
   移除  remove()  poll()  take()  poll(time, unit)
   检查  element()  peek()  不可用  不可用

   */
  private BlockingQueue blockingQueue;
  private AtomicInteger count = new AtomicInteger();
  public Producer(BlockingQueue blockingQueue) {
    this.blockingQueue = blockingQueue;
  }

  /**
   * When an object implementing interface Runnable is used
   * to create a thread, starting the thread causes the object's
   * run method to be called in that separately executing
   * thread.
   * 

   * The general contract of the method run is that it may
   * take any action whatsoever.
   *
   * @see Thread#run()
   */
  @Override
  public void run() {
    System.out.println("生产者线程启动...");
    LockFlag.setProducerRunningFlag(true);
    try {
      while (LockFlag.getProducerRunningFlag()){
        String data = "data:"+count.incrementAndGet();
        if(blockingQueue.offer(data,10, TimeUnit.SECONDS)){
          //返回true表示生产数据正确
          System.out.println("^^^^^^^^^^^^^^正在生产数据 data="+data);
        }else {
          //表示阻塞时间内还没有生产者生产数据
          System.out.println("生产者异常，无法生产数据");
        }
        Thread.sleep(50);

      }
    } catch (InterruptedException e) {
      e.printStackTrace();
      System.err.println("生产者程序退出");
      LockFlag.setProducerRunningFlag(false);//异常退出线程
      Thread.currentThread().interrupt();
    }
  }
}

package com.qysxy.framework.queue;

/**
 * @author fuguangli
 * @description 前沿生产者消费者模型的锁类
 * @Create date:  2017/3/7
 */
public class LockFlag {
  /**
   * 生产者互斥锁
   */
  private static Boolean producerRunningFlag = false;
  /**
   * 消费者互斥锁
   */
  private static Boolean customerRunningFlag = false;

  public static Boolean getProducerRunningFlag() {
    return producerRunningFlag;
  }

  public static void setProducerRunningFlag(Boolean producerRunningFlag) {
    LockFlag.producerRunningFlag = producerRunningFlag;
  }

  public static Boolean getCustomerRunningFlag() {
    return customerRunningFlag;
  }

  public static void setCustomerRunningFlag(Boolean customerRunningFlag) {
    LockFlag.customerRunningFlag = customerRunningFlag;
  }
}

package com.qysxy.framework.queue;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.Queue;
import java.util.concurrent.*;

/**
 * @author fuguangli
 * @description 前沿队列实用类，用于大量并发用户
 * @Create date:  2017/3/7
 */
public class BlockingQueueHelper {


  private static final Integer maxQueueSize = 1000;
  private static BlockingQueue blockingQueue = new LinkedBlockingQueue(maxQueueSize);
  private static ExecutorService threadPool = Executors.newCachedThreadPool();


  public static BlockingQueue getBlockingQueue() {
    if (blockingQueue == null) {
      blockingQueue = new LinkedBlockingQueue(maxQueueSize);
    }
    return blockingQueue;
  }

  /**
   * @param o 队列处理对象（包含request,response,data）
   */
  public static void requestQueue(Object o) {
    //检测当前的队列大小
    if (blockingQueue != null && blockingQueue.size() < maxQueueSize) {
      //可以正常进入队列
      if (blockingQueue.offer(o)) {
        //添加成功，检测数据处理线程是否正常
        if (LockFlag.getCustomerRunningFlag()) {
          //说明处理线程类正常运行
        } else {
          //说明处理线程类停止，此时，应重新启动线程进行数据处理
          LockFlag.setCustomerRunningFlag(true);

          //example:run
          Customer customer = new Customer(blockingQueue);
          threadPool.execute(customer);

        }

      } else {
        //进入队列失败，做出相应的处理，或者尝试重新进入队列

      }
    } else {
      //队列不正常，或队列大小已达上限，做出相应处理

    }

  }
}