线程中级篇(一)：基本并发容器

传统的容器并没有实现线程安全，所以在多线程的情况下，并不管用。本文首先自己手写一个并发Queue引出全文：

import java.util.LinkedList; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class MyQueue { //1 需要一个承装元素的集合 private LinkedList<Object> list = new LinkedList<Object>(); //2 需要一个计数器 private AtomicInteger count = new AtomicInteger(0); //3 需要制定上限和下限 private final int minSize = 0; private final int maxSize ; //4 构造方法 public MyQueue(int size){ this.maxSize = size; } //5 初始化一个对象 用于加锁 private final Object lock = new Object(); //put(anObject): 把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断，直到BlockingQueue里面有空间再继续. public void put(Object obj){ synchronized (lock) { while(count.get() == this.maxSize){ try { lock.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } //1 加入元素 list.add(obj); //2 计数器累加 count.incrementAndGet(); //3 通知另外一个线程（唤醒） lock.notify(); System.out.println("新加入的元素为:" + obj); } } //take: 取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入. public Object take(){ Object ret = null; synchronized (lock) { while(count.get() == this.minSize){ try { lock.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } //1 做移除元素操作 ret = list.removeFirst(); //2 计数器递减 count.decrementAndGet(); //3 唤醒另外一个线程 lock.notify(); } return ret; } public int getSize(){ return this.count.get(); } public static void main(String[] args) { final MyQueue mq = new MyQueue(5); mq.put("a"); mq.put("b"); mq.put("c"); mq.put("d"); mq.put("e"); System.out.println("当前容器的长度:" + mq.getSize()); Thread t1 = new Thread(new Runnable() { @Override public void run() { mq.put("f"); mq.put("g"); } },"t1"); t1.start(); Thread t2 = new Thread(new Runnable() { @Override public void run() { Object o1 = mq.take(); System.out.println("移除的元素为:" + o1); Object o2 = mq.take(); System.out.println("移除的元素为:" + o2); } },"t2"); try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } t2.start(); } } 

并发类容器是java.util.concurrent包中提供的用于并发编程的容器，主要有以下大类：

package com.xushu.multi; import java.util.Vector; public class Ticket { public static void main(String[] args) { final Vector<String> tickets = new Vector<String>(); for(int i = 1; i < 1000; i++){ tickets.add("火车票" + i); } for(int i = 1; i <= 10; i++){ new Thread("线程" + i){ public void run() { while(true){ if(tickets.isEmpty()) break; System.out.println(Thread.currentThread().getName() + "---" + tickets.remove(0)); } } }.start(); } } } 

DelayQueue的演示
WangMing.java

package com.xushu.multi; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; public class WangMing implements Delayed{ private String name; //身份证 private String id; //截止时间 private long endTime; //定义时间工具类 private TimeUnit timeUnit = TimeUnit.SECONDS; public WangMing(String name, String id, Long endTime){ this.name = name; this.id = id; this.endTime = endTime; } public String getName(){ return this.name; } public String getId(){ return this.id; } @Override public int compareTo(Delayed o) { WangMing w = (WangMing)o; return this.getDelay(this.timeUnit) - w.getDelay(this.timeUnit) > 0 ? 1 : 0; } /** * 用来判断是否到了截止时间 */ @Override public long getDelay(TimeUnit unit) { return endTime - System.currentTimeMillis(); } } 

WangBa.java

package com.xushu.multi; import java.util.concurrent.DelayQueue; public class WangBa implements Runnable{ private DelayQueue<WangMing> queue = new DelayQueue<WangMing>(); public boolean yinye = true; public void shangji(String name,String id,int money){ WangMing man = new WangMing(name, id, 1000 * money + System.currentTimeMillis()); System.out.println("网名"+man.getName()+" 身份证"+man.getId()+"交钱"+money+"块,开始上机..."); this.queue.add(man); } public void xiaji(WangMing man){ System.out.println("网名"+man.getName()+" 身份证"+man.getId()+"时间到下机..."); } @Override public void run() { while(yinye){ WangMing man; try { man = queue.take(); xiaji(man); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } public static void main(String[] args) { try{ System.out.println("网吧开始营业"); WangBa speed = new WangBa(); Thread shangwang = new Thread(speed); shangwang.start(); speed.shangji("路人甲", "123", 1); speed.shangji("路人乙", "234", 10); speed.shangji("路人丙", "345", 5); } catch(Exception e){ e.printStackTrace(); } } }