Java并发编程---并发类容器(ConcurrentMap容器)-蒲公英云

Java并发编程---并发类容器(ConcurrentMap容器)

一.背景前奏

jdk5.0以后提供了多种并发类容器来替代同步类容器从而改善性能.同步类容器的状态都是串行化的.他们虽然实现了线程安全.但是严重降低了并发性,在多线程环境时,严重降低了应用程序的吞吐量.

并发类容器是专门针对并发设计的使用ConcurrentHashMap来代替给予散列的传统的HashTable,而且在ConcurrentHashMap中,添加了一些常见复合操作的支持.以及使用了CopyOnWriteArrayList代替Vector,并发的CopyOnWriteArraySet,以及并发地Queue,ConcurrentLinkedQueue和LinkedBlockingQueue,前者是高性能的队列,后者是以阻塞形式的队列,具体实现Queue还有很多,例如ArrayBlockingQueue,PriorityBlockingQueue,synchronousQueue等.

二.ConcurrentMap概念

ConcurrentMap接口下有两个重要的实现:

ConcurrentHashMap

ConcurrentSkipListMap(支持并发排序功能,补充ConcurrentHashMap)

ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分,每一段其实就是一个小的HashTable,它们有自己的锁.只要多个修改操作发生在不同的段上,它们就可以并发进行.把一个整体分成16个段(Segment).也就是最高支持16个线程的并发修改操作.这也是多线程场景时减少锁的粒度从而降低锁竞争的一种方案,并且代码中大多共享变量使用Volatile关键字声明,目的是第一时间获取修改的内容,性能非常好.

三.代码示例

private ConcurrentHashMap> map = new ConcurrentHashMap>();

package com.thread.message;
import java.util.LinkedList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class MyQueue {
    //1.需要一个承载元素的集合
    LinkedList<Object> list = new LinkedList<Object>();
    //2.需要一个计数器(统计linkedList的个数)
    private AtomicInteger count = new AtomicInteger(0);
    //应该设置ConcurrentHashMap容器的上限,如10个(根据该数据结构,适当修改业务逻辑)
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Integer>> map = new ConcurrentHashMap<String,ConcurrentHashMap<String, Integer>>();
    //3.需要制约上限和下限
    private final int minSize = 0;
    private int maxSize;
    //4.构造方法
    public MyQueue(int size) {
        this.maxSize = maxSize;
    }
    //5.初始化一个对象,用于加锁
    private final Object lock = new Object();
    //put(anObject):把anObject加到BlockingQueue里,如果BlockingQueue没有空间,则调用此方法的线程被阻断,直到BlockingQueue里面有空间再继续
    public void put(Object obj){
        synchronized (lock) {
            while(count.get() == this.maxSize){
                try {
                    lock.wait();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            //1.加入元素
            list.add(obj);
            //2.计数器累加
            count.incrementAndGet();    //加1后notify
            //通知另外一个线程(唤醒)
            lock.notify();
            System.out.println("新加入的元素为: " + obj);
        }
    }
    //take: 取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态,直到BlockingQueue有新的数据被加入
    public Object take(){
        Object ret = null;
        synchronized (lock) {
            while(count.get() == this.maxSize){
                try {
                    lock.wait();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    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) {
        // TODO Auto-generated method stub
        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() {
                // TODO Auto-generated method stub
                mq.put("f");
                mq.put("g");
            }
        },"t1");    
        t1.start();
        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                // TODO Auto-generated method stub
                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) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        t2.start();
    }
}