线程并发集合实现java生成消费模型（ArrayBlockingQueue和ConcurrentMap）

最近项目需要多线程，涉及线程当然就是并发的问题，在java 1.5后有了这个java.util.concurrent 包提供了一些支持并发的集合，用起来很方便，在这里就记下最简单也是最长见的java生成消费模型，不多说，附上代码

这个生产消费者模式，区分不同的生成者和不同消费者，比喻如下：有个仓库，仓库有个门就是这个和ConcurrentMap，仓库里面有很多个箱子，不同的生产者生产不同的产品放箱子里面，消费者去不同箱子里取不同的产品。仓库的门一次只能进一个人。

容器模型：

package com.me.test;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ConcurrentMap;

public class CorrentMapTest {

private static CorrentMapTest instance=null;

private static ConcurrentMap<String, ArrayBlockingQueue<String>> map=null;

private CorrentMapTest(){

map=new ConcurrentHashMap<String, ArrayBlockingQueue<String>>();

}

public static synchronized CorrentMapTest getInstance(){

if(instance==null){

instance=new CorrentMapTest();

}

return instance;

}

public void putMessage(String name,String content){

ArrayBlockingQueue<String> qu=map.get(name);

if(qu==null){

qu=new ArrayBlockingQueue<String>(1024);

}

try {

qu.put(content);

map.put(name, qu);

//System.out.println(System.currentTimeMillis()/1000+"成产者："+map);

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

public String getMessage(String name){

String message=null;

//System.out.println("来取数据:"+map);

ArrayBlockingQueue<String> qu=map.get(name);

if(qu==null){

qu=new ArrayBlockingQueue<String>(1024);

map.put(name, qu);

}

try {

message=qu.take();

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

return message;

}

}

成产者模型：

package com.me.test;

public class proceder extends Thread{

private CorrentMapTest map=CorrentMapTest.getInstance();

private String name;

private String content;

public proceder(String name,String content) {

this.name=name;

this.content=content;

}

@Override

public void run() {

while(true){

map.putMessage(name, content);

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}

}

消费者模型：

package com.me.test;

public class Cosumer extends Thread{

private String name;

private CorrentMapTest map=CorrentMapTest.getInstance();

public Cosumer(String name) {

this.name=name;

}

@Override

public void run() {

try {

Thread.sleep(10);//经过测试，这里好像一定要延时下，具体原理暂时不清楚

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

while(true){

String message=map.getMessage(name);

System.out.println(Thread.currentThread().getName()+"----"

+message);

try {

Thread.sleep(10);//经过测试，这里好像一定要延时下，保证每个线程都有机会执行，就是保证每个在门口等的人都能进去取，不能说我出来了马上又进去

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}

}

测试类

package com.me.test;

public class Test {

/**

* @param args

*/

public static void main(String[] args) {

// TODO Auto-generated method stub

try {

new proceder("张三", "做了一个项目").start();

//Thread.sleep(2000);

new Cosumer("张三").start();

new Cosumer("张三").start();

new Cosumer("李四").start();

new Cosumer("李四").start();

Thread.sleep(0);

new proceder("李四", "来打酱油了。。。").start();

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}

具体的请看类的方法请自己看java API，这里就不多说了