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

2022-06-02 10:57 85阅读 0赞

**一.背景前奏**

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<String,ConcurrentHashMap<String, Integer>> map = new ConcurrentHashMap<String,ConcurrentHashMap<String, Integer>>();**

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();
    	}
    
    }