Java5 Concurrent包中的锁机制

转自http://www.iteye.com/topic/333669

        JDK1.5以后加入了concurrent包，主要是为了提高多线程的开发效率，其中提供了很多支持并发的集合类，其中包括：ConcurrentHashMap。大家知道HashTable也是支持并发环境的，也就是说多线程安全的，那两者有什么区别呢？

分析

        其实简单的说是同步机制有区别，具体区别又在那里呢？

        请看HashTable的put方法：

/**
     * Maps the specified <code>key</code> to the specified
     * <code>value</code> in this hashtable. Neither the key nor the
     * value can be <code>null</code>. <p>
     *
     * The value can be retrieved by calling the <code>get</code> method
     * with a key that is equal to the original key.
     *
     * @param      key     the hashtable key
     * @param      value   the value
     * @return     the previous value of the specified key in this hashtable,
     *             or <code>null</code> if it did not have one
     * @exception  NullPointerException  if the key or value is
     *               <code>null</code>
     * @see     Object#equals(Object)
     * @see     #get(Object)
     */
    public synchronized V put(K key, V value) {
    // Make sure the value is not null
    if (value == null) {
        throw new NullPointerException();
    }

    // Makes sure the key is not already in the hashtable.
    Entry tab[] = table;
    int hash = key.hashCode();
    int index = (hash & 0x7FFFFFFF) % tab.length;
    for (Entry<K,V> e = tab[index] ; e != null ; e = e.next) {
        if ((e.hash == hash) && e.key.equals(key)) {
        V old = e.value;
        e.value = value;
        return old;
        }
    }

    modCount++;
    if (count >= threshold) {
        // Rehash the table if the threshold is exceeded
        rehash();

            tab = table;
            index = (hash & 0x7FFFFFFF) % tab.length;
    }

    // Creates the new entry.
    Entry<K,V> e = tab[index];
    tab[index] = new Entry<K,V>(hash, key, value, e);
    count++;
    return null;
    }

     代码中使用synchronized函数的方式进行同步，这个类的其他方法也是使用这个机制进行同步的。我们先来大致了解一下synchronized的机制：

     在多线程环境中，Java的对象都会隐式的包含有一个同步队列，其中类会有一个，然后每个类实例也会包含一个。如图：

class Foo {
      synchronized void doSomething(); // 把同步的线程放入类实例的同步队列
      synchronized static void doSomething(); //把同步的线程放入类的同步队列
}

 然后我们再来看看ConcurrentHashMap 的put方法：

//ConcurrentHashMap
public V put(K key, V value) {
        if (value == null)
            throw new NullPointerException();
        int hash = hash(key.hashCode());
        return segmentFor(hash).put(key, hash, value, false);
    }

//Segment内部类，继承至ReentrantLock
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();
            }
        }

ReentrantLock就是Java Concurrent包提供的锁对象，Lock的使用方法大致如下：

Lock l = ...;
l.lock();
try {
    // access the resource protected by this lock
} finally {
    l.unlock();
}

为什么使用Lock对象会比使用synchronized有更好的性能呢？我们再来看看ReentrantLock的实现：

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    /** Synchronizer providing all implementation mechanics */
    private final Sync sync;

    /**
     * Base of synchronization control for this lock. Subclassed
     * into fair and nonfair versions below. Uses AQS state to
     * represent the number of holds on the lock.
     */
    static abstract class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;
..........
...........

 我们从中看到ReentrantLock对象也维护着一个同步队列，性能差别就在于这个队列的实现上，我们再来看AbstractQueuedSynchronizer的代码：

 /**
     * Inserts node into queue, initializing if necessary. See picture above.
     * @param node the node to insert
     * @return node's predecessor
     */
    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                Node h = new Node(); // Dummy header
                h.next = node;
                node.prev = h;
                if (compareAndSetHead(h)) {
                    tail = node;
                    return h;
                }
            }
            else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

/**
     * CAS head field. Used only by enq
     */
    private final boolean compareAndSetHead(Node update) {
        return unsafe.compareAndSwapObject(this, headOffset, null, update);//使用compare and swap方式
    }

    /**
     * CAS tail field. Used only by enq
     */
    private final boolean compareAndSetTail(Node expect, Node update) {
        return unsafe.compareAndSwapObject(this, tailOffset, expect, update);//使用compare and swap方式
    }

到了Unsafe.java这里就要通过本地代码实现了，下面是kaffe里面的本地代码实现：

/**
 * Helper macro, defining a sun.misc.Unsafe compare and swap function
 * with a given NAME tail and TYPE of arguments.
 */
#define KAFFE_UNSAFE_COMPARE_AND_SWAP(NAME, TYPE)
JNIEXPORT jboolean JNICALL Java_sun_misc_Unsafe_compareAndSwap ## NAME(JNIEnv* env, jobject unsafe UNUSED, jobject obj, jlong offset, TYPE expect, TYPE update)
{
  volatile TYPE * address = getFieldAddress(env, obj, offset);
  if (sizeof(TYPE) == sizeof(gint))
    return g_atomic_int_compare_and_exchange((volatile gint *) address, (gint) expect, (gint) update);
  else if (sizeof(TYPE) == sizeof(gpointer))
    return g_atomic_pointer_compare_and_exchange((volatile gpointer *) address, (gpointer) expect, (gpointer) update);
  else
    if (*address == expect) {
      *address = update;
      return JNI_TRUE;
    }
    else
      return JNI_FALSE;
}

再看glib的代码：

gboolean g_atomic_int_compare_and_exchange (volatile gint *atomic,
                   gint           oldval,
                   gint           newval)
{
  gint result;

  __asm__ __volatile__ ("lock; cmpxchgl %2, %1"
            : "=a" (result), "=m" (*atomic)
            : "r" (newval), "m" (*atomic), "0" (oldval));

  return result == oldval;
}

 AT&T汇编，

一条指令总会是原子性的了吧

总结

     concurrent包中大量使用了新的锁机制，新的Lock机制最终归结到一个原子性操作上，所以会比使用synchronized关键字有更高的性能。