并发容器代码摘要

/**

*循环增加putIndex和takeIndex,如果到数组尾部,那么

*置为0

*/

final

int

inc(

int

i){

return

(++i
==items.length)?

0

:i;

}

//**

*插入一个item,需要执行线程获得了锁

*/

private

void

insert(E
x){

items[putIndex]
=x;

//累加putIndex,可能到数组尾部,那么重新指向0位置

putIndex
=inc(putIndex);

++count;

//put后,使用Condition通知正在等待take的线程可以做提取操作

notEmpty.signal();

}

/**

*获取一个元素,执行这个操作的前提是线程已经获得锁,

*内部调用

*/

private

E
extract(){

final

E[]
items=

this

.items;

E
x=items[takeIndex];

items[takeIndex]
=

null

;

//累加takeIndex,有可能到数组尾部,重新调到数组头部

takeIndex
=inc(takeIndex);

--count;

//take后,使用Condition通知正在等待插入的线程可以插入

notFull.signal();

return

x;

}

public

final

void

await()

throws

InterruptedException
{

if

(Thread.interrupted())

throw

new

InterruptedException();

//加一个新的condition等待节点

Node
node=addConditionWaiter();

//释放自己占用的锁

int

savedState
=fullyRelease(node);

int

interruptMode
=

0

;

while

(!isOnSyncQueue(node)){

//如果当前线程等待状态是CONDITION,park住当前线程,等待condition的signal来解除

LockSupport.park(

this

);

if

((interruptMode
=checkInterruptWhileWaiting(node))!=

0

)

break

;

}

if

(acquireQueued(node,
savedState)&&interruptMode!=THROW_IE)

interruptMode
=REINTERRUPT;

if

(node.nextWaiter
!=

null

)

unlinkCancelledWaiters();

if

(interruptMode
!=

0

)

reportInterruptAfterWait(interruptMode);

}

private

E
dequeue(){

Node<E>h=head;

Node<E>first=h.next;

//将原始节点的next指针指向自己,这样就能GC到自己否则虚拟机会认为这个节点仍然在用而不销毁(不知道是否理解有误)

h.next
=h;

//helpGC

head
=first;

E
x=first.item;

first.item
=

null

;

return

x;

}

//无论是offerLast,offerFirst,pollFirst,pollLast等等方法都会使用同一把锁.

public

E
pollFirst(){

final

ReentrantLock
lock=

this

.lock;

lock.lock();

try

{

return

unlinkFirst();

}

finally

{

lock.unlock();

}

}

public

E
pollLast(){

final

ReentrantLock
lock=

this

.lock;

lock.lock();

try

{

return

unlinkLast();

}

finally

{

lock.unlock();

}

}

int

sshift
=

0

;

int

ssize
=

1

;

//ssize是左移位的,也就是2,4,8,16,32增长(*2),所以你设定concurrencyLevel为10的时候,这个时候并发数最大为8.

while

(ssize
<concurrencyLevel){

++sshift;

ssize
<<=

1

;

}

每个Segment维持一个自动增长的HashEntry数组(根据一个阈值确定是否要增长长度，并不是满了才做).

int

c
=count;

//threshold一般(int)(capacity*loadFactor),

if

(c++>threshold)

rehash();

public

ConcurrentHashMap(

int

initialCapacity,

float

loadFactor,

int

concurrencyLevel){

if

(!(loadFactor
>

0

)||initialCapacity
<

0

||concurrencyLevel<=

0

)

throw

new

IllegalArgumentException();

//首先确定segment的个数,左移位,并且记录移了几次,比如conurrencyLevel为30,那么2->4->8->16,ssize为16,sshift为4

if

(concurrencyLevel
>MAX_SEGMENTS)

concurrencyLevel
=MAX_SEGMENTS;

int

sshift
=

0

;

int

ssize
=

1

;

while

(ssize
<concurrencyLevel){

++sshift;

ssize
<<=

1

;

}

//segmentShift为28

segmentShift
=

32

-sshift;

//segmentMask为15

segmentMask
=ssize-

1

;

//this.segments=new
Segment[16]

this

.segments
=Segment.newArray(ssize);

if

(initialCapacity
>MAXIMUM_CAPACITY)

initialCapacity
=MAXIMUM_CAPACITY;

//假设initialCapacity使用32,那么c=2

int

c
=initialCapacity/ssize;

if

(c
*ssize<initialCapacity)

++c;

int

cap
=

1

;

//cap为2

while

(cap
<c)

cap
<<=

1

;

//每个Segment的容量为2

for

(

int

i
=

0

;i<

this

.segments.length;++i)

this

.segments[i]
=

new

Segment<K,V>(cap,
loadFactor);

}

/***/

/**

*segmentShift为28,segmentMask为15(1111)

*因为hash值为int,所以32位的

*hash
>>>segentShift会留下最高的4位,

*再与mask
1111做&操作

*所以这个最终会产生
0-15的序列.

*/

final

Segment<K,V>segmentFor(

int

hash){

return

segments[(hash
>>>segmentShift)&segmentMask];

}

/**

*将计算的hash值补充到原始hashCode中,这是为了防止

*外部用户传进来劣质的hash值(比如重复度很高)所带来

*的危害.

*/

private

static

int

hash(

int

h){

//
Spreadbitstoregularizebothsegmentandindexlocations,

//
usingvariantofsingle-wordWang/Jenkinshash.

h
+=(h<<

15

)^

0xffffcd7d

;

h
^=(h>>>

10

);

h
+=(h<<

3

);

h
^=(h>>>

6

);

h
+=(h<<

2

)+(h<<

14

);

return

h
^(h>>>

16

);

}