同步容器是什么:
JDK提供给了很多容器,其中有list,set,queue,map等。
这里我们挑出List单讲。
众所周知,很多书上,我们看到Arraylist并不是线程安全的,Vector是线程安全的。
那就从源码上分析一下:
ArrayList中,add方法如下:
public boolean add(E e) {
ensureCapacityInternal(size + 1); // Increments modCount!!
elementData[size++] = e;
return true;
}
复制代码
Vector中,add方法如下:
public synchronized boolean add(E e) {
modCount++;
ensureCapacityHelper(elementCount + 1);
elementData[elementCount++] = e;
return true;
}
复制代码
对比发现,Vector之所以是线程安全的,是因为Vector对所有的方法使用synchronized进行了修饰。
不安全的同步容器:
public class SynchornizedVector {
public static void main(String[] agrs){
Vector vector = new Vector();
for(int i =0 ; i<10; i++){
vector.add(i,i);
}
new Thread(){
@Override
public void run() {
//vector共有10个元素,index对应0-9
//第一步:线程1执行到j=8,暂停;
for(int j = 0; j < vector.size(); j++){
//第三部,线程1继续执行,要获取vector.get(8)的时候出错,因为vector的元素已经被线程2清空
if(j == 8){
try {
Thread.currentThread().sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(vector.get(j));
}
}
}.start();
new Thread(){
@Override
public void run() {
//第二步:线程2获得时间片,立即执行,删除掉vector中所有的元素
for(int i = 0; i < vector.size(); i++){
vector.remove(i);
}
}
}.start();
}
}
复制代码
需要对size()的地方进行同步互斥,才能确保容器是安全的,举例如下:
第39行和第17行
public class SynchornizedVector {
public static void main(String[] agrs) {
Vector vector = new Vector();
for (int i = 0; i < 10; i++) {
vector.add(i, i);
}
new Thread() {
@Override
public void run() {
//vector共有10个元素,index对应0-9
//第一步:线程1执行到j=8,暂停;
synchronized (SynchornizedVector.class) {
for (int j = 0; j < vector.size(); j++) {
//第三部,线程1继续执行,要获取vector.get(8)的时候出错,因为vector的元素已经被线程2清空
if (j == 8) {
try {
Thread.currentThread().sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(vector.get(j));
}
}
}
}.start();
new Thread() {
@Override
public void run() {
//第二步:线程2获得时间片,立即执行,删除掉vector中所有的元素
synchronized (SynchornizedVector.class) {
for (int i = 0; i < vector.size(); i++) {
vector.remove(i);
}
}
}
}.start();
}
}
复制代码
工程中大量使用的同步容器ConcurrentHashMap
众所周知,hashMap是根据散列值分段存储的,同步Map在同步的时候锁住了所有的段(粗粒度的锁)
而ConcurrentHashMap根据散列值锁定了散列值对应的段,提高了并发性能(细粒度的锁)
其数据结构如下:
根据图中的数据结构:
每次对key寻找到相应的位置需要两次定位:1.定位到Segment。2.定位到元素所在Segment中的具体链表的头部。
对读操作不加锁,对写操作的锁的粒度细化到每个Segment
支持的最大并发数就是Segment的数量
static final class Segment<K,V> extends ReentrantLock implements Serializable {
transient volatile int count;
transient int modCount;
transient int threshold;
transient volatile HashEntry<K,V>[] table;
final float loadFactor;
}
复制代码
count:Segment中元素的数量
modCount:对table的大小造成影响的操作的数量,比如put(),remove()
threshold:扩容阈值
table:数组中每一个元素代表了一个链表的头部
loadFactor:用于确定threshold
get过程
static final class HashEntry<K,V> {
final K key;
final int hash;
volatile V value;
final HashEntry<K,V> next;
}
复制代码
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
// Find power-of-two sizes best matching arguments
int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
segmentShift = 32 - sshift;
segmentMask = ssize - 1;
this.segments = Segment.newArray(ssize);
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = 1;
while (cap < c)
cap <<= 1;
for (int i = 0; i < this.segments.length; ++i)
this.segments[i] = new Segment<K,V>(cap, loadFactor);
}
复制代码
initialCapcity:初始的容量
loadFactor:负载参数
concurrentLevel:Segment的数量,一旦设定不可改变,如果map容量不够,需要扩容,则增加Segment数组的大小,而不增加Segment的数量,这样就不需要对Map做rehash,只要对Segment中的元素做rehash
整个ConcurrentHashMap的初始化方法还是非常简单的,先是根据concurrentLevel来new出Segment,这里Segment的数量是不大于concurrentLevel的最大的2的指数,就是说Segment的数量永远是2的指数个,这样的好处是方便采用移位操作来进行hash,加快hash的过程。接下来就是根据intialCapacity确定Segment的容量的大小,每一个Segment的容量大小也是2的指数,同样使为了加快hash的过程。
public V get(Object key) {
int hash = hash(key.hashCode());
return segmentFor(hash).get(key, hash);
}
复制代码
第三行的作用是:把key对应的segment找出来
final Segment<K,V> segmentFor(int hash) {
return segments[(hash >>> segmentShift) & segmentMask];
}
复制代码
采用移位的方式操作,可以加快计算速度
确定了具体的segment之后,就要确定segment中具体的链表位置
HashEntry<K,V> getFirst(int hash) {
HashEntry<K,V>[] tab = table;
return tab[hash & (tab.length - 1)];
}
复制代码
V get(Object key, int hash) {
if (count != 0) { // read-volatile
HashEntry<K,V> e = getFirst(hash);
while (e != null) {
if (e.hash == hash && key.equals(e.key)) {
V v = e.value;
if (v != null)
return v;
return readValueUnderLock(e); // recheck
}
e = e.next;
}
}
return null;
}
复制代码
put过程:
V put(K key, int hash, V value, boolean onlyIfAbsent) {
lock();
try {
int c = count;
if (c++ > threshold) // ensure capacity
rehash();
HashEntry<K,V>[] tab = table;
int index = hash & (tab.length - 1);
HashEntry<K,V> first = tab[index];
HashEntry<K,V> e = first;
while (e != null && (e.hash != hash || !key.equals(e.key)))
e = e.next;
V oldValue;
if (e != null) {
oldValue = e.value;
if (!onlyIfAbsent)
e.value = value;
}
else {
oldValue = null;
++modCount;
tab[index] = new HashEntry<K,V>(key, hash, first, value);
count = c; // write-volatile
}
return oldValue;
} finally {
unlock();
}
}
复制代码
如果Segment中元素的数量超过了threshold就要进行rehash,如有key存在,则更新value值,否则新生成一个HashEntry加入到整个Segment的头部
注意:
ConcurrentHashMap 的 get 的操作在大多数情况下都是不加锁的,只有当找到的 HashEntry 的 value 是 null 时,才会再进行一次加锁的读操作,以保障读操作的一致性。通常这种情况发生在你找到的 HashEntry 恰是另一个线程在做 put 操作时创建的,且 value 恰好没有设置完成。这种情况不太容易发生。所以,对于 ConcurrentHashMap 来说,发生在同一个 Segment 的一个写和多个读操作是并不互斥的,所以 Segment 也就没有继承读写锁了,而且这种设计要比读写锁的并发能力更高
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/101366.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...