likes
comments
collection
share

ConcurrentHashMap 原理及使用详解

作者站长头像
站长
· 阅读数 42

ConcurrentHashMap是Java中的一种线程安全的哈希表实现，它提供了与Hashtable和HashMap类似的API，但通过使用分段锁技术（Segment），使得多个线程可以同时读取和写入不同的数据块，从而提高了并发性能。同时，ConcurrentHashMap也支持弱一致性，即在某些情况下，读取操作可能会返回稍早的值，但这对于很多应用场景来说是可以接受的。该类还提供了一些有用的方法，如putIfAbsent()、replace()、compute()等，方便开发者进行基于哈希表的数据处理。总之，ConcurrentHashMap是一个高效且可靠的多线程环境下的哈希表实现，非常适合在并发场景中使用。

一、ConcurrentHashMap 的数据结构

ConcurrentHashMap 是 Java 中的一种线程安全的哈希表实现，其数据结构是由多个 Node 节点组成的数组和链表或红黑树。

在 ConcurrentHashMap 中，Node 节点是一个键值对，其中键为 K 类型，值为 V 类型。每个节点包含了一个哈希值、一个键和一个值，以及指向下一个节点的引用。具体而言，ConcurrentHashMap 的内部数据结构如下：

ConcurrentHashMap 原理及使用详解

ConcurrentHashMap 中的每个 Segment 是一个独立的哈希表，而每个 Segment 又由多个 Bucket 组成，每个 Bucket 再维护一个链表或红黑树（红黑树出现的条件是 Bucket 中的元素数量大于等于 8）。

ConcurrentHashMap 的 put 操作可以分成两个步骤，首先根据 Key 的哈希值找到对应的 Segment 和 Bucket，然后在 Bucket 中插入新的 Node 节点。具体来说，它的处理流程如下：

  1. 对 Key 进行哈希操作，得到其哈希值。
  2. 根据哈希值和 Segment 数组的长度，计算出 Key 应该被放在哪个 Segment 中。
  3. 在 Segment 中获取 Key 应该放在哪个 Bucket 中。
  4. 如果该 Bucket 是空的，则直接在它的头部插入新的 Node 节点；否则，将新的 Node 节点插入到链表的尾部或红黑树上，并根据情况进行扩容或红黑树转换为链表。
  5. 如果插入新的节点后 Bucket 中的元素数量超过了一个阈值，则需要进行扩容操作。
  6. 如果插入新的节点后 Bucket 中的元素数量大于等于 8 个并且 Bucket 不是红黑树，则需要将链表转换为红黑树。
  7. 如果旧的红黑树中的节点数量少于 6 个，则需要将红黑树转换为链表。

ConcurrentHashMap 的 get 操作也很简单，直接根据 Key 的哈希值找到对应的 Segment 和 Bucket，然后在 Bucket 中查找对应的 Node 节点即可。

二、ConcurrentHashMap 的分段锁机制

ConcurrentHashMap 将整个哈希表分为多个 Segment，每个 Segment 又是一个独立的哈希表，可以单独进行加锁和扩容操作。在操作数据时，只需要获取对应 Segment 的锁，不需要锁住整个哈希表，这样可以避免多个线程之间的等待和竞争，同时提高吞吐量和并发性能。

简单实现示例：

class ConcurrentHashMap<K, V> {
    final Segment[] segments;

    class Segment extends ReentrantLock implements Serializable{
        // 每个 Segment 自己独立的哈希表
        private final Map<K,V> map = new HashMap<>();

        // Segment 内部加锁机制，确保线程安全
        public synchronized V put(K key, V value) {
            return map.put(key, value);
        }
    }

    // 获取 key 所属的 Segment 的索引
    private int getSegmentIndex(K key) {
        int hash = hash(key.hashCode()); // 对 hashcode() 进行哈希
        int segmentMask = segments.length - 1; // mask 值
        return hash & segmentMask; // 按位与，定位具体的 Segment
    }

    public V put(K key, V value) {
        Segment s = segments[getSegmentIndex(key)];
        s.lock(); // 获取 s 对应的 Segment 的锁
        try {
            return s.put(key, value); // 在 s 上进行 put 操作
        } finally {
            s.unlock(); // 释放 s 对应的 Segment 的锁
        }
    }
}

在实际的 ConcurrentHashMap 中，每个 Segment 会使用一个独立的哈希表来维护其内部的数据，同时也具备自己的锁机制，从而实现对其内部状态的并发安全访问。这样，不同线程访问不同的 Segment 时可以通过分段锁机制来实现并发访问，从而提高了 ConcurrentHashMap 的并发性能和吞吐量。

三、ConcurrentHashMap 的实现过程

在 JDK 1.8 以前，ConcurrentHashMap 的实现采用了与 Hashtable 类似的分段锁机制，每个 Segment 都对应一个 ReentrantLock 锁，用于并发访问。

而在 JDK 1.8 中，ConcurrentHashMap 引入了 CAS（Compare and Swap）技术，用于实现一个更加高效的并发控制机制。CAS 是一种无锁机制，可以避免线程争抢锁的情况。

我们以 put 操作为例，来看一下 ConcurrentHashMap 的实现过程：

  1. 首先计算 key 的哈希值；
  2. 根据哈希值找到对应的 Segment；
  3. 获取 Segment 对应的锁；
  4. 如果还没有元素，就直接插入到 Segment 中；
  5. 如果已经存在元素，就循环比较 key 是否相等；
  6. 如果 key 已经存在，就根据要求更新 value；
  7. 如果 key 不存在，就插入新的元素（链表或者红黑树）。

上述操作中，步骤 2 到 3 相当于加了一个悲观锁，在整个哈希表上加锁，如果只有一个 Segment，效果与 Hashtable 类似；如果存在多个 Segment，效果就相当于使用了分段锁机制，提高了并发访问性能。

四、使用场景案例

1. 高并发的计数器

ConcurrentHashMap 可以用来实现高并发的计数器，例如记录网站访问量、接口调用次数等。具体地，我们可以使用 ConcurrentHashMap 的 compute 方法来实现计数操作，如下所示：

import java.util.concurrent.ConcurrentHashMap;

public class Counter {
    private final ConcurrentHashMap<String, Integer> map;

    public Counter() {
        this.map = new ConcurrentHashMap<>();
    }

    public void increase(String key) {
        map.compute(key, (k, v) -> (v == null) ? 1 : v + 1);
    }

    public int get(String key) {
        return map.getOrDefault(key, 0);
    }
}

在上述代码中，我们创建了一个 Counter 类，使用 ConcurrentHashMap 存储计数器数据。具体地，我们使用 compute 方法实现对计数器的增加操作，如果 key 不存在则新建一个值为 1 的计数器；否则将其递增 1。通过 get 方法可以获取指定 key 对应的计数器值。

2. 线程池任务管理

ConcurrentHashMap 还可以用来实现线程池任务的管理，例如记录每个任务的执行状态、结果等信息。具体地，我们可以将一个 ConcurrentHashMap 实例作为任务管理器，在任务执行前将任务信息添加到该管理器中，然后再在任务完成后更新对应的信息，如下所示：

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TaskManager {
    private final ConcurrentHashMap<String, TaskInfo> tasks;

    public TaskManager() {
        this.tasks = new ConcurrentHashMap<>();
    }

    public void addTask(String taskId, Runnable task) {
        // 添加任务信息
        tasks.put(taskId, new TaskInfo());

        // 提交任务到线程池
        ExecutorService executor = Executors.newCachedThreadPool();
        executor.submit(() -> {
            try {
                // 执行任务
                task.run();

                // 更新任务状态和结果
                TaskInfo info = tasks.get(taskId);
                info.setStatus(TaskStatus.COMPLETED);
                info.setResult("Task completed successfully!");

            } catch (Exception ex) {
                // 更新任务状态和结果
                TaskInfo info = tasks.get(taskId);
                info.setStatus(TaskStatus.FAILED);
                info.setResult(ex.getMessage());
            }
        });

        // 关闭线程池
        executor.shutdown();
    }

    public TaskInfo getTaskInfo(String taskId) {
        return tasks.getOrDefault(taskId, new TaskInfo());
    }
}

enum TaskStatus {
    NEW,
    RUNNING,
    COMPLETED,
    FAILED;
}

class TaskInfo {
    private TaskStatus status;
    private String result;

    public TaskInfo() {
        this.status = TaskStatus.NEW;
        this.result = "";
    }

    public TaskStatus getStatus() {
        return status;
    }

    public void setStatus(TaskStatus status) {
        this.status = status;
    }

    public String getResult() {
        return result;
    }

    public void setResult(String result) {
        this.result = result;
    }
}

在上述代码中，我们创建了一个 TaskManager 类，使用 ConcurrentHashMap 存储任务信息。具体地，我们定义了一个 TaskInfo 类来表示任务信息，其中包括任务状态和结果两个属性。在添加任务时，我们新建一个 TaskInfo 实例并添加到 ConcurrentHashMap 中，然后在异步执行任务的线程中更新其状态和结果；同时我们使用 ExecutorService 来管理并发执行的任务。通过 getTaskInfo 方法可以获取指定 taskId 对应的任务信息。

3. 缓存管理器

ConcurrentHashMap 还可以用来实现缓存管理器，例如存储经常使用的业务数据、系统配置等信息，从而避免频繁的数据库查询或网络请求。具体地，我们可以使用 ConcurrentHashMap 存储缓存数据，并设定缓存过期时间，如下所示：

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class CacheManager<K, V> {
    private final Map<K, CacheEntry<V>> cache;

    public CacheManager() {
        this.cache = new ConcurrentHashMap<>();
    }

    public void put(K key, V value, long ttl) {
        // 添加缓存项，同时记录当前时间戳和缓存生存时间
        CacheEntry<V> entry = new CacheEntry<>(value, System.currentTimeMillis(), ttl);
        cache.put(key, entry);
    }

    public V get(K key) {
        // 获取缓存项和其缓存生存时间
        CacheEntry<V> entry = cache.get(key);

        // 检查缓存项是否过期，如果过期则删除缓存项并返回 null
        if (entry != null && !entry.isExpired()) {
            return entry.getValue();
        } else {
            cache.remove(key);
            return null;
        }
    }

    static class CacheEntry<V> {
        private final V value;
        private final long timestamp;
        private final long ttl;

        public CacheEntry(V value, long timestamp, long ttl) {
            this.value = value;
            this.timestamp = timestamp;
            this.ttl = ttl;
        }

        public V getValue() {
            return value;
        }

        public boolean isExpired() {
            return System.currentTimeMillis() - timestamp > ttl;
        }
    }
}

在上述代码中，我们创建了一个 CacheManager 类，使用 ConcurrentHashMap 存储缓存数据。具体地，我们定义了一个 CacheEntry 类来表示缓存项，其中包括值、时间戳和缓存生存时间三个属性。在添加缓存项时，我们新建一个 CacheEntry 实例并添加到 ConcurrentHashMap 中，然后在获取缓存项时检查其是否过期；如果未过期则返回其值，否则删除缓存项并返回 null。

转载自:https://juejin.cn/post/7241749918451367997
评论
请登录