java下降競爭鎖的一些方法

時間 2019-11-07

標籤 java 下降競爭一些方法欄目 Java 简体版

原文原文鏈接

序

本文介紹一下提高併發可伸縮性的一些方式：減小鎖的持有時間，下降鎖的粒度，鎖分段、避免熱點域以及採用非獨佔的鎖或非阻塞鎖來代替獨佔鎖。編程

減小鎖的持有時間

下降發生競爭可能性的一種有效方式就是儘量縮短鎖的持有時間。例如，能夠將一些與鎖無關的代碼移出同步代碼塊，尤爲是那些開銷較大的操做，以及可能被阻塞的操做，例如I/O操做。數組

優化前

@ThreadSafe
public class AttributeStore {
    @GuardedBy("this") private final Map<String, String>
            attributes = new HashMap<String, String>();

    public synchronized boolean userLocationMatches(String name,
                                                    String regexp) {
        String key = "users." + name + ".location";
        String location = attributes.get(key);
        if (location == null)
            return false;
        else
            return Pattern.matches(regexp, location);
    }
}

優化後

@ThreadSafe
public class BetterAttributeStore {
    @GuardedBy("this") private final Map<String, String>
            attributes = new HashMap<String, String>();

    public boolean userLocationMatches(String name, String regexp) {
        String key = "users." + name + ".location";
        String location;
        synchronized (this) {
            location = attributes.get(key);
        }
        if (location == null)
            return false;
        else
            return Pattern.matches(regexp, location);
    }
}

下降鎖的粒度

另外一種減少鎖的持有時間的方式是下降線程請求鎖的頻率（從而減少發生競爭的可能性）。這能夠經過鎖分解和鎖分段等技術來實現，在這些技術中將採用多個相互獨立的鎖來保護獨立的狀態變量，從而改變這些變量在以前由單個鎖來保護的狀況。這些技術能減少鎖操做的粒度，並能實現更高的可伸縮性，然而，使用的鎖越多，那麼發生死鎖的風險也就越高。緩存

優化前

@ThreadSafe
public class ServerStatusBeforeSplit {
    @GuardedBy("this") public final Set<String> users;
    @GuardedBy("this") public final Set<String> queries;

    public ServerStatusBeforeSplit() {
        users = new HashSet<String>();
        queries = new HashSet<String>();
    }

    public synchronized void addUser(String u) {
        users.add(u);
    }

    public synchronized void addQuery(String q) {
        queries.add(q);
    }

    public synchronized void removeUser(String u) {
        users.remove(u);
    }

    public synchronized void removeQuery(String q) {
        queries.remove(q);
    }
}

優化後

@ThreadSafe
public class ServerStatusAfterSplit {
    @GuardedBy("users") public final Set<String> users;
    @GuardedBy("queries") public final Set<String> queries;

    public ServerStatusAfterSplit() {
        users = new HashSet<String>();
        queries = new HashSet<String>();
    }

    public void addUser(String u) {
        synchronized (users) {
            users.add(u);
        }
    }

    public void addQuery(String q) {
        synchronized (queries) {
            queries.add(q);
        }
    }

    public void removeUser(String u) {
        synchronized (users) {
            users.remove(u);
        }
    }

    public void removeQuery(String q) {
        synchronized (users) {
            queries.remove(q);
        }
    }
}

鎖分段

在某些狀況下，能夠將鎖分解技術進一步擴展爲對一組獨立對象上的鎖進行分解，這種狀況被稱爲鎖分段。例如，在ConcurrentHashMap的實現中使用了一個包含16個鎖的數組，每一個鎖保護全部散列桶的1/16，其中第N個散列桶由第（Nmod 16）個鎖來保護。假設散列函數具備合理的分佈性，而且關鍵字可以實現均勻分佈，那麼這大約能把對於鎖的請求減小到原來的1/16。正是這項技術使得ConcurrentHashMap可以支持多達16個併發的寫入器。（要使得擁有大量處理器的系統在高訪問量的狀況下實現更高的併發性，還能夠進一步增長鎖的數量，但僅當你能證實併發寫入線程的競爭足夠激烈並須要突破這個限制時，才能將鎖分段的數量超過默認的16個。）性能優化

鎖分段的一個劣勢在於：與採用單個鎖來實現獨佔訪問相比，要獲取多個鎖來實現獨佔訪問將更加困難而且開銷更高。一般，在執行一個操做時最多隻需獲取一個鎖，但在某些狀況下須要加鎖整個容器，例如當ConcurrentHashMap須要擴展映射範圍，以及從新計算鍵值的散列值要分佈到更大的桶集合中時，就須要獲取分段所集合中全部的鎖。數據結構

@ThreadSafe
public class StripedMap {
    // Synchronization policy: buckets[n] guarded by locks[n%N_LOCKS]
    private static final int N_LOCKS = 16;
    private final Node[] buckets;
    private final Object[] locks;

    private static class Node {
        Node next;
        Object key;
        Object value;
    }

    public StripedMap(int numBuckets) {
        buckets = new Node[numBuckets];
        locks = new Object[N_LOCKS];
        for (int i = 0; i < N_LOCKS; i++)
            locks[i] = new Object();
    }

    private final int hash(Object key) {
        return Math.abs(key.hashCode() % buckets.length);
    }

    public Object get(Object key) {
        int hash = hash(key);
        synchronized (locks[hash % N_LOCKS]) {
            for (Node m = buckets[hash]; m != null; m = m.next)
                if (m.key.equals(key))
                    return m.value;
        }
        return null;
    }

    public void clear() {
        for (int i = 0; i < buckets.length; i++) {
            synchronized (locks[i % N_LOCKS]) {
                buckets[i] = null;
            }
        }
    }
}

避免熱點域

若是一個鎖保護兩個獨立變量X和Y，而且線程A想要訪問X，而線程B想要訪問Y（這相似於在ServerStatus中，一個線程調用addUser，而另外一個線程調用addQuery），那麼這兩個線程不會在任何數據上發生競爭，即便它們會在同一個鎖上發生競爭。當每一個操做都請求多個變量時，鎖的粒度將很難下降。這是在性能與可伸縮性之間相互制衡的另外一個方面，一些常見的優化措施，例如將一些反覆計算的結果緩存起來，都會引入一些「熱點域（HotField）」，而這些熱點域每每會限制可伸縮性。當實現HashMap時，你須要考慮如何在size方法中計算Map中的元素數量。最簡單的方法就是，在每次調用時都統計一次元素的數量。一種常見的優化措施是，在插入和移除元素時更新一個計數器，雖然這在put和remove等方法中略微增長了一些開銷，以確保計數器是最新的值，但這將把size方法的開銷從O（n）下降到O（l）。併發

在單線程或者採用徹底同步的實現中，使用一個獨立的計數能很好地提升相似size和isEmpty這些方法的執行速度，但卻致使更難以提高實現的可伸縮性，由於每一個修改map的操做都須要更新這個共享的計數器。即便使用鎖分段技術來實現散列鏈，那麼在對計數器的訪問進行同步時，也會從新致使在使用獨佔鎖時存在的可伸縮性問題。一個看似性能優化的措施—緩存size操做的結果，已經變成了一個可伸縮性問題。在這種狀況下，計數器也被稱爲熱點域，由於每一個致使元素數量發生變化的操做都須要訪問它。爲了不這個問題，ConcurrentHashMap中的size將對每一個分段進行枚舉並將每一個分段中的元素數量相加，而不是維護一個全局計數。爲了不枚舉每一個元素，ConcurrentHashMap爲每一個分段都維護了一個獨立的計數，並經過每一個分段的鎖來維護這個值。函數

public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }
final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }

代替獨佔鎖

第三種下降競爭鎖的影響的技術就是放棄使用獨佔鎖，從而有助於使用一種友好併發的方式來管理共享狀態。例如，使用併發容器、讀-寫鎖、不可變對象以及原子變量。ReadWriteLock能提供比獨佔鎖更高的併發性。而對於只讀的數據結構，其中包含的不變性能夠徹底不須要加鎖操做。性能

public class ReadWriteMap <K,V> {
    private final Map<K, V> map;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private final Lock r = lock.readLock();
    private final Lock w = lock.writeLock();

    public ReadWriteMap(Map<K, V> map) {
        this.map = map;
    }

    public V put(K key, V value) {
        w.lock();
        try {
            return map.put(key, value);
        } finally {
            w.unlock();
        }
    }

    public V remove(Object key) {
        w.lock();
        try {
            return map.remove(key);
        } finally {
            w.unlock();
        }
    }

    public void putAll(Map<? extends K, ? extends V> m) {
        w.lock();
        try {
            map.putAll(m);
        } finally {
            w.unlock();
        }
    }

    public void clear() {
        w.lock();
        try {
            map.clear();
        } finally {
            w.unlock();
        }
    }

    public V get(Object key) {
        r.lock();
        try {
            return map.get(key);
        } finally {
            r.unlock();
        }
    }

    public int size() {
        r.lock();
        try {
            return map.size();
        } finally {
            r.unlock();
        }
    }

    public boolean isEmpty() {
        r.lock();
        try {
            return map.isEmpty();
        } finally {
            r.unlock();
        }
    }

    public boolean containsKey(Object key) {
        r.lock();
        try {
            return map.containsKey(key);
        } finally {
            r.unlock();
        }
    }

    public boolean containsValue(Object value) {
        r.lock();
        try {
            return map.containsValue(value);
        } finally {
            r.unlock();
        }
    }
}