java-并发集合-并发队列 ConcurrentLinkedQueue 演示

package me.grass.demo.concuronte;

import java.util.Date;

/**
* 同步队列 java.util.concurrent.ConcurrentLinkedQueue 使用
*
*/
public class ConcurrentLinkedQueueDemo {
/**
* 生产者快，消费者慢
* @param args
* @throws InterruptedException
*  @author xxj
*/
public static void main(String[] args) throws InterruptedException {
LinkedBlockingQueueDemo._lCountDownLatch.countDown();
Date before=new Date();
int pTotalThread =100; //最大线程数（生产者）
int cTotalThread =5; //活动线程数（消费者）
int pActivities=50; //最大线程数（生产者）
int cActivities=5; //活动线程数（消费者）
_lCountDownLatch = new CountDownLatch(pTotalThread+cTotalThread);

startProducer(pActivities,pTotalThread);
startConsumer(cActivities,cTotalThread);

_lCountDownLatch.await();//等待所有线程完成
Date after = new Date();

System.out.println("队列为空："+_queue.isEmpty());
System.out.println("耗时："+((after.getTime()-before.getTime())/1000));
System.out.println("同步队列："+_lCountDownLatch.getCount());
}
static java.util.concurrent.CountDownLatch _lCountDownLatch;
static java.util.concurrent.ConcurrentLinkedQueue<Integer> _queue =
new ConcurrentLinkedQueue<Integer>();

static void startProducer(int active,int totalThread) throws InterruptedException{
java.util.concurrent.ExecutorService pool = Executors.newFixedThreadPool(active);
int size =1024*1024*10;//产生 3 M 数据
Thread thread ;
for(int i=0;i<totalThread;i++){
thread = new Thread(new producer(i,size));
pool.execute(thread);
}
}
static void startConsumer(int active,int totalThread) throws InterruptedException{
java.util.concurrent.ExecutorService pool = Executors.newFixedThreadPool(active);
Thread thread ;
//启动x个消费者
for(int i=0;i<totalThread;i++){
thread = new Thread(new consumer());
pool.execute(thread);
}
}
/**
* 生产者
* @author xxj
*
*/
static class producer implements Runnable{
public producer(int key,int size){
_size=size;
_key=key;
}
int _key;
int _size;
public void run() {
ConcurrentLinkedQueueDemo._queue.add(_key);//生产
System.out.println("已创建："+_key);
ConcurrentLinkedQueueDemo._lCountDownLatch.countDown();//线程同步递减
}
}
/**
* 消费者
* @author xxj
*
*/
static class consumer implements Runnable{
public consumer(){
}
public void run() {
//循环消费，直到队列内容为空
while(!ConcurrentLinkedQueueDemo._queue.isEmpty()){
Integer nInteger = ConcurrentLinkedQueueDemo._queue.poll();//消费
System.err.println("消费："+nInteger);
try {
Thread.sleep(200);//每次消费等一会儿
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
ConcurrentLinkedQueueDemo._lCountDownLatch.countDown();//线程同步递减
}
}
}