ConcurrentHashMap的使用场景

Posted 2023-06-22 哩个啷个波

一、并发容器ConcurrentHashMap

HashMap是我们用得非常频繁的一个集合，但是它是线程不安全的。并且在多线程环境下，put操作是有可能产生死循环，不过在JDK1.8的版本中更换了数据插入的顺序，已经解决了这个问题。

为了解决该问题，提供了Hashtable和Collections.synchronizedMap(hashMap)两种解决方案，但是这两种方案都是对读写加锁，独占式。一个线程在读时其他线程必须等待，吞吐量较低，性能较为低下。

而J.U.C给我们提供了高性能的线程安全HashMap：ConcurrentHashMap。

在1.8版本以前，ConcurrentHashMap采用分段锁的概念，使锁更加细化，但是1.8已经改变了这种思路，而是利用CAS+Synchronized来保证并发更新的安全，当然底层采用数组+链表+红黑树的存储结构。

　　ConcurrentHashMap通常只被看做并发效率更高的Map，用来替换其他线程安全的Map容器，比如 Hashtable和Collections.synchronizedMap。线程安全的容器，特别是Map，很多情况下一个业务中 涉及容器的操作有多个（读get写put,remove），即复合操作，而在并发执行时，线程安全的容器只能保证自身的数据不被破 坏，和数据在多个线程间是可见的，但无法保证业务的行为是否正确。

1、ConcurrentHashMap对比Hashtable

Hashtable和ConcurrentHashMap的不同点：

（1）、Hashtable 对get,put,remove都使用了同步操作，它的同步级别是正对Hashtable来进行同步的，也就是说如果有线程正在遍历集合，其他的线程就暂时不能使用该集合了，这样无疑就很容易对性能和吞吐量造成影响，从而形成单点。而ConcurrentHashMap则不同，它只对put,remove操作使用了同步操作，get操作并不影响。

（2）、Hashtable 在遍历的时候，如果其他线程，包括本线程对Hashtable进行了put，remove等更新操作的话，就会抛出ConcurrentModificationException异常，但如果使用ConcurrentHashMap的话，就不用考虑这方面的问题了

2、ConcurrentHashMap总结

（1）、HashMap 是线程不安全的，ConcurrentHashMap是线程安全的，但是线程安全仅仅指的是对容器操作的时候是线程安全的

（2）、ConcurrentHashMap 的public V get(Object key)不涉及到锁，也就是说获得对象时没有使用锁，它只对put,remove操作使用了同步操作

（3）、put 、remove方法，在jdk7使用锁，但多线程中并不一定有锁争用，原因在于ConcurrentHashMap将缓存的变量分到多个Segment，每个Segment上有一个锁，只要多个线程访问的不是一个Segment就没有锁争用，就没有堵塞，各线程用各自的锁，ConcurrentHashMap缺省情况下生成16个Segment，也就是允许16个线程并发的更新而尽量没有锁争用。而在jdk8中使用的CAS+Synchronized来保证线程安全，比加锁的性能更高

（4）、ConcurrentHashMap 线程安全的，允许一边更新、一边遍历，也就是说在对象遍历的时候，也可以进行remove,put操作，且遍历的数据会随着remove,put操作产出变化

以下例子分别使用HashMap、ConcurrentHashMap、HashTable在遍历的同时删除，

案例一：遍历的同时删除

说明：ConcurrentHashMap 线程安全的，允许一边更新、一边遍历，也就是说在对象遍历的时候，也可以进行remove,put操作，且遍历的数据会随着remove,put操作产出变化

情况一、使用HashMap进行遍历的同时删除

public class ConcurrentHashMapDemo 
    public static void main(String[] args) 
        Map<String, Integer> map = new HashMap<>();
        map.put("a",1);
        map.put("b",1);
        map.put("c",1);
        for (Map.Entry<String, Integer> entry : map.entrySet()) 
            map.remove(entry.getKey());
        
        System.out.println(map.size());
    

HashMap不能一边遍历一边更新，否则报异常ConcurrentModificationException

情况二、使用ConcurrentHashMap进行遍历的同时删除

public class ConcurrentHashMapDemo 
    public static void main(String[] args) 
//        Map<String, Integer> map = new HashMap<>();
        Map<String, Integer> map = new ConcurrentHashMap<>();
        map.put("a",1);
        map.put("b",1);
        map.put("c",1);
        for (Map.Entry<String, Integer> entry : map.entrySet()) 
            map.remove(entry.getKey());
        
        System.out.println(map.size());
    

而ConcurrentHashMap不存在该问题，输出结果为0.

情况三、使用HashTable进行遍历的同时删除

public class ConcurrentHashMapDemo 
    public static void main(String[] args) 
//        Map<String, Integer> map = new HashMap<>();
//        Map<String, Integer> map = new ConcurrentHashMap<>();
        Map<String, Integer> map = new Hashtable<>();
        map.put("a",1);
        map.put("b",1);
        map.put("c",1);
        for (Map.Entry<String, Integer> entry : map.entrySet()) 
            map.remove(entry.getKey());
        
        System.out.println(map.size());
    

如果用性能较低的安全容器HashTable，也报异常ConcurrentModificationException。

案例2：业务操作的线程安全不能保证

说明：线程安全的容器，特别是Map，很多情况下一个业务中 涉及容器的操作有多个（读get写put,remove），即复合操作，而在并发执行时，线程安全的容器只能保证自身的数据不被破 坏，和数据在多个线程间是可见的，但无法保证业务的行为是否正确，即ConcurrentHashMap多线程操作不能保证数据同步。

以下分别使用HashMap、ConcurrentHashMap、HashTable边遍历时边更新，运行了3个线程，理论上最后得到6000，

public class ConcurrentHashMapDemo2 
    public static void main(String[] args) 
        final Map<String, Integer> count = new ConcurrentHashMap<>();
//        final Map<String, Integer> count = new HashMap<>();
//        final Map<String, Integer> count = new Hashtable<>();
        count.put("count",0);
        Runnable task = new Runnable()
            @Override
            public void run() 
                int value;
                for (int i = 0; i < 2000; i++) 
                    value = count.get("count");
                    count.put("count",value + 1);
                
            
        ;
        new Thread(task).start();
        new Thread(task).start();
        new Thread(task).start();
        try 
            Thread.sleep(1000l);
            System.out.println(count);
         catch (InterruptedException e) 
            e.printStackTrace();
        
    

运行结果如下：

1、如果使用HashMap，结果为3426

count=3426

2、如果使用ConcurrentHashMap，结果为2525，因为只能保证对容器的操作是没问题的，但是不能保证业务是没有问题的，因为是复和操作且并发执行。

count=2525

3、HashTable也不能保证业务没有问题。

count=3814

如果非要在这种情况下保证线程安全问题，同步就可以了，加同步代码块，保证读写是同步的

public class ConcurrentHashMapDemo2 
    public static void main(String[] args) 
        final Map<String, Integer> count = new ConcurrentHashMap<>();
//        final Map<String, Integer> count = new HashMap<>();
//        final Map<String, Integer> count = new Hashtable<>();
        count.put("count",0);
        Runnable task = new Runnable()
            @Override
            public void run() 
                synchronized (count)
                    int value;
                    for (int i = 0; i < 2000; i++) 
                        value = count.get("count");
                        count.put("count",value + 1);
                    
                

            
        ;
        new Thread(task).start();
        new Thread(task).start();
        new Thread(task).start();
        try 
            Thread.sleep(1000l);
            System.out.println(count);
         catch (InterruptedException e) 
            e.printStackTrace();
        
    

结果如下：

count=6000

案例3：多线程删除

public class ConcurrentHashMapDemo3 
    public static void main(String[] args) 
//        final Map<String, Integer> count = new HashMap<>();
        final Map<String, Integer> count = new ConcurrentHashMap<>();
        //final Hashtable<String, Integer> count = new Hashtable<>();
        for (int i = 0; i < 2000; i++) 
            count.put("count" + i, 1);
        
        Runnable task1 = new Runnable() 
            @Override
            public void run() 
                for (int i = 0; i < 500; i++) 
                    count.remove("count" + i);
                
            
        ;
        Runnable task2 = new Runnable() 
            @Override
            public void run() 
                for (int i = 1000; i < 1500; i++) 
                    count.remove("count" + i);
                
            
        ;
        new Thread(task1).start();
        new Thread(task2).start();
        try 
            Thread.sleep(1000l);
            System.out.println(count.size());
         catch (Exception e) 
            e.printStackTrace();
        
    

当使用ConcurrentHashMap时，结果为1000；当使用HashMap时，结果为1024；当使用HashTable时，结果为1000。

案例4：多线程遍历删除不同的的数据

public class ConcurrentHashMapDemo4 
    public static void main(String[] args) 
        final Map<Integer, Integer> count = new ConcurrentHashMap<>();
        for (int i = 0; i < 2000; i++) 
            count.put(i, 1);
        
        Runnable task1 = new Runnable() 
            @Override
            public void run() 
                Iterator<Map.Entry<Integer, Integer>> it = count.entrySet().iterator();
                while (it.hasNext()) 
                    Map.Entry<Integer, Integer> entry = it.next();
                    if (entry.getKey() < 5) 
                        count.remove(entry.getKey()); // 多线程遍历的时候删除
                        System.out.println(entry.getKey());
                    
                
            
        ;
        Runnable task2 = new Runnable() 
            @Override
            public void run() 
                Iterator<Map.Entry<Integer, Integer>> it = count.entrySet().iterator();
                while (it.hasNext()) 
                    Map.Entry<Integer, Integer> entry = it.next();
                    if (entry.getKey() >= 1995) 
                        count.remove(entry.getKey()); // 多线程遍历的时候删除
                        System.out.println(entry.getKey());
                    
                
            
        ;
        new Thread(task1).start();
        new Thread(task2).start();
        try 
            Thread.sleep(1000l);
            System.out.println("map中键值对的数量"+count.size());
         catch (Exception e) 
            e.printStackTrace();
        
    

结果如下：

0
1
2
3
4
1995
1996
1997
1998
1999
map中键值对的数量1990

推荐HashMap应用场景：

　　多线程操作下HashMap无法保证数据同步，多线程修改HashMap并且有遍历的操作时，可能会产生ConcurrentModificationException异常。所以，推荐的HashMap应用场景是单线程运行环境，并且不需要遍历操作的场景。
这个推荐场景不是硬性条件。比如多线操作HashMap，我们通过加锁或者加入同步控制依然能正常应用HashMap，只是需要加上同步操作的代价。（单线程且不需要遍历时使用HashMap）

ConcurrentHashMap推荐应用场景：

多线程对HashMap数据添加删除操作时，可以采用ConcurrentHashMap。

下面三个场景都使用ConcurrentHashMap:

1、多线程添加或删除。2、遍历的时候删除。3、多线程遍历的时候删除数据

注意ConcurrentHashMap多线程操作不能保证数据同步，此时可以加同步代码块进行同步操作。

二、项目中ConcurrentHashMap的使用案例

在实际的生产环境中，单看代码很多操作并没有使用到我们常说的实现多线程的方式，但是结合具体的使用场景，某个接口或者方法会多次由不同请求发起时候，一个请求就会打开一个新的线程，其场景和直接使用多线程的效果差不多。

应用1：webSocket用来存放客户端的信息

（1）、建立连接后,把登录用户的uid和session通过put操作写入ConcurrentHashMap中。同时广播时要遍历ConcurrentHashMap来给每一个存活的session发消息该用户上线了。

session值如下：

sessionID为4，session中携带的值uid为2

（2）、登录用户下线后，需要遍历ConcurrentHashMap，如果ConcurrentHashMap中的sessionID等于当前下线的用户的sessionID，则从ConcurrentHashMap中移出。然后广播时再遍历ConcurrentHashMap，然后给每一个Session发送该用户下线的消息。

/**
 * Socket处理器
 */
@Component
public class MyWebSocketHandler implements WebSocketHandler 
    //用于保存HttpSession与WebSocketSession的映射关系
    public static final Map<Long, WebSocketSession> userSocketSessionMap;

    @Autowired
    LoginService loginservice;

    static 
        userSocketSessionMap = new ConcurrentHashMap<Long, WebSocketSession>();
    

    /**
     * 建立连接后,把登录用户的id写入WebSocketSession
     */
    public void afterConnectionEstablished(WebSocketSession session)
            throws Exception 
        Long uid = (Long) session.getAttributes().get("uid");
        String username=loginservice.getnamebyid(uid);
        if (userSocketSessionMap.get(uid) == null) 
            userSocketSessionMap.put(uid, session); // 多线程添加操作
            Message msg = new Message();
            msg.setFrom(0L);//0表示上线消息
            msg.setText(username);
            this.broadcast(new TextMessage(new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create().toJson(msg)));
        
    

    /**
     * 消息处理，在客户端通过Websocket API发送的消息会经过这里，然后进行相应的处理
     */
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception 
            if(message.getPayloadLength()==0)
                return;
            Message msg=new Gson().fromJson(message.getPayload().toString(),Message.class);
            msg.setDate(new Date());
            sendMessageToUser(msg.getTo(), new TextMessage(new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create().toJson(msg)));
    

    /**
     * 消息传输错误处理
     */
    public void handleTransportError(WebSocketSession session,
            Throwable exception) throws Exception 
        if (session.isOpen()) 
            session.close();
        
        Iterator<Entry<Long, WebSocketSession>> it = userSocketSessionMap.entrySet().iterator();
        // 移除当前抛出异常用户的Socket会话
        while (it.hasNext()) 
            Entry<Long, WebSocketSession> entry = it.next();
            if (entry.getValue().getId().equals(session.getId())) 
                userSocketSessionMap.remove(entry.getKey());
                System.out.println("Socket会话已经移除:用户ID" + entry.getKey());
                String username=loginservice.getnamebyid(entry.getKey());
                Message msg = new Message();
                msg.setFrom(-2L);
                msg.setText(username);
                this.broadcast(new TextMessage(new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create().toJson(msg)));
                break;
            
        
    

    /**
     * 关闭连接后
     */
    public void afterConnectionClosed(WebSocketSession session,CloseStatus closeStatus) throws Exception 
        System.out.println("Websocket:" + session.getId() + "已经关闭");
        Iterator<Entry<Long, WebSocketSession>> it = userSocketSessionMap.entrySet().iterator();
        // 移除当前用户的Socket会话
        while (it.hasNext()) 
            Entry<Long, WebSocketSession> entry = it.next();
            if (entry.getValue().getId().equals(session.getId())) 
                userSocketSessionMap.remove(entry.getKey()); // 多线程遍历的时候删除
                System.out.println("Socket会话已经移除:用户ID" + entry.getKey());
                String username=loginservice.getnamebyid(entry.getKey());
                Message msg = new Message();
                msg.setFrom(-2L);//下线消息，用-2表示
                msg.setText(username);
                this.broadcast(new TextMessage(new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create().toJson(msg)));
                break;
            
        
    

    public boolean supportsPartialMessages() 
        return false;
    

    /**
     * 给所有在线用户发送消息
     * @param message
     * @throws IOException
     */
    public void broadcast(final TextMessage message) throws IOException 
        Iterator<Entry<Long, WebSocketSession>> it = userSocketSessionMap.entrySet().iterator();

        //多线程群发
        while (it.hasNext()) 

            final Entry<Long, WebSocketSession> entry = it.next();

            if (entry.getValue().isOpen()) 
                // entry.getValue().sendMessage(message);
                new Thread(new Runnable() 

                    public void run() 
                        try 
                            if (entry.getValue().isOpen()) 
                                entry.getValue().sendMessage(message);
                            
                         catch (IOException e) 
                            e.printStackTrace();
                        
                    

                ).start();
            

        
    

    /**
     * 给某个用户发送消息
     *
     * @param uid
     * @param message
     * @throws IOException
     */
    public void sendMessageToUser(Long uid, TextMessage message) throws IOException 
        WebSocketSession session = userSocketSessionMap.get(uid);
        if (session != null && session.isOpen()) 
            session.sendMessage(message);