LinkedBlockingQueue 例子
import java.util.Random;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.LinkedBlockingQueue;import java.util.concurrent.TimeUnit;import java.util.concurrent.atomic.AtomicLong;import org.apache.commons.lang.RandomStringUtils;/** * @author zhaoqilong * @version 创建时间：2012-6-7 上午9:16:56 * */public class Test { private static LinkedBlockingQueue<String> queue =new LinkedBlockingQueue<String>(); // 线程控制开关 private final CountDownLatch latch = new CountDownLatch(1); //的线程池 private final ExecutorService pool; //AtomicLong 计数 生产数量 private final AtomicLong output = new AtomicLong(0); //AtomicLong 计数 销售数量 private final AtomicLong sales = new AtomicLong(0); //是否停止线程 private final boolean clear; public Test(boolean clear){ this.pool = Executors.newCachedThreadPool(); this.clear=clear; } public void service() throws InterruptedException{ Saler a=new Saler(queue, sales, latch, clear); pool.submit(a); Worker w=new Worker(queue, output, latch); pool.submit(w); latch.countDown(); } public static void main(String[] args) { Test t=new Test(false); try {t.service();} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} } class Saler implements Runnable{ private final LinkedBlockingQueue<String> queue; private final AtomicLong sales; private final CountDownLatch latch; private final boolean clear; public Saler(LinkedBlockingQueue<String> queue, AtomicLong sales, CountDownLatch latch, boolean clear){ this.queue = queue; this.sales = sales; this.latch = latch; this.clear = clear; }public void run() {try { latch.await(); // 放闸之前老实的等待着 for (;;) { sale(); Thread.sleep(500); } }catch (InterruptedException e) { if(clear) { // 响应中断请求后,如果有要求则销售完队列的产品后再终止线程 cleanWarehouse(); } else { System.out.println("Seller Thread will be interrupted..."); } } }public void sale(){System.out.println("==取take=");try {String item = queue.poll(50, TimeUnit.MILLISECONDS);System.out.println(item);if(item!=null){sales.incrementAndGet(); // 可以声明long型的参数获得返回值,作为日志的参数}} catch (InterruptedException e) {e.printStackTrace();} } /** * 销售完队列剩余的产品 */ private void cleanWarehouse() { try { while (queue.size() > 0) { sale(); } } catch (Exception ex) { System.out.println("Seller Thread will be interrupted..."); } } } /** * 生产者 * @author Administrator * */ class Worker implements Runnable{ private LinkedBlockingQueue<String> queue; private CountDownLatch latch; private AtomicLong output; public Worker(){ } public Worker(LinkedBlockingQueue<String> queue, AtomicLong output,CountDownLatch latch){ this.queue=queue; this.latch=latch; this.output=output; } public void run() { try { latch.await(); // 线程等待 for (;;) { work(); Thread.sleep(100); } }catch (InterruptedException e) { System.out.println("Worker thread will be interrupted..."); } } /** * 工作 */ public void work(){ try { String product=RandomStringUtils.randomAscii(3); boolean success=queue.offer(product, 100, TimeUnit.MILLISECONDS); if(success){ output.incrementAndGet();// 可以声明long型的参数获得返回值,作为日志的参数 } } catch (InterruptedException e) { e.printStackTrace(); } } }}