当前位置：首页 > news >正文

JUC大揭秘：从ConcurrentHashMap到线程池，玩转Java并发编程！

news 来源：原创 2025/5/17 0:26:00

JUC实现类

ConcurrentHashMap

回顾HashMap

ConcurrentHashMap

CopyOnWriteArrayList

回顾ArrayList

CopyOnWriteArrayList:

CopyOnWriteArraySet

辅助类 CountDownLatch

线程池

线程池优点

ThreadPoolExecutor

构造器各个参数含义：

线程池的执行

线程池中的队列

线程池中的拒绝策略

execute和submit的区别、

关闭线程池

ThreadLocal

原理分析

编辑

对象四种引用

ThreadLocal内存泄漏问题

JUC实现类

Java 5.0 在 java.util.concurrent 包中提供了多种并发容器类来改进同步容器的性能。

ConcurrentHashMap

回顾HashMap

双列集合实现Map接口

键值对

键不能重复,值可以重复

只能存储一个为null的键

键是无序的

是线程不安全的.

HashMap不能有多个线程同时操作 ,如果有,则会抛出java.util.ConcurrentModificationException(并发修改异常)

键是如何判断是否重复

hashCode() 和 equals()

用到的一些结构

1.哈希表默认长度是16 哈希每次扩容为原来的2倍哈希表的负载因子为0.75

2.链表链表长度>= 8 且哈希表长度大于等于64 才会把链表转为红黑树否则会先扩容哈希表 3.红黑树

讲讲添加一个元素的过程

HashMap不能再多线程场景下使用,否则会报异常

线程安全 : Hashtable 给操作的方法都添加了synchronized 但是效并发率低了

package com.ffyc.javaPro.thread.juc;

import java.util.HashMap;
import java.util.Hashtable;
import java.util.concurrent.ConcurrentHashMap;

public class HashMapDemo {

    public static void main(String[] args) {
        HashMap<String,String> map = new HashMap<>();

        //模拟多线程场景
        for (int i = 0; i <100000 ; i++) {
            new Thread(){
                @Override
                public void run() {
                    map.put(Thread.currentThread().getName(),"aaaa");
                    System.out.println(map); 
                }
            }.start();
        }
    }
}

package com.ffyc.javaPro.thread.juc;

import java.util.HashMap;
import java.util.Hashtable;
import java.util.concurrent.ConcurrentHashMap;

public class HashMapDemo {

    public static void main(String[] args) {
        Hashtable<String,String> map = new Hashtable<>();
        //模拟多线程场景
        for (int i = 0; i <100000 ; i++) {
            new Thread(){
                @Override
                public void run() {
                    map.put(Thread.currentThread().getName(),"aaaa");
                    System.out.println(map);
                }
            }.start();
        }
    }
}

ConcurrentHashMap

ConcurrentHashMap也是线程安全的,但是与Hashtable实现线程安全的方式不同,他没有直接给方法加锁,
给哈希表的每一个位置加锁,将锁的粒度细化了,提高了并发效率.
如何细化锁: 不使用专门的分段锁了,而是采用每一个位置上的第一个节点Node对象,作为锁对象
使用CAS+synchronized实现线程安全
当哈希表的某个位置上还没有Node对象时,如果此时有多个线程操作,采用cas机制进行比较判断
如果某个位置上已经有了Node对象,那么直接使用Node对象作为锁即可

ConcurrentHashMap 和Hashtable 都不能存储为null的键和为null值
为了消除歧义因为他们都是在多线程场景下使用的,返回null时,不能分辨出时key的值为null,还是没有这个key,返回的null

package com.ffyc.javaPro.thread.juc;

import java.util.HashMap;
import java.util.Hashtable;
import java.util.concurrent.ConcurrentHashMap;

public class HashMapDemo {

    public static void main(String[] args) {
        ConcurrentHashMap<String,String> map = new ConcurrentHashMap<>();

        //模拟多线程场景
        for (int i = 0; i <100000 ; i++) {
            new Thread(){
                @Override
                public void run() {
                    System.out.println(map.get("a")); //值为null  还是没有这个键
                }
            }.start();
        }
    }
}

CopyOnWriteArrayList

回顾ArrayList

ArrayList 数组列表线程不安全的
Vector 数组列表线程安全的

public synchronized boolean add(E e) 直接给方法加锁,效率低
public synchronized E get(int index) { get方法也加了锁,如果只有多个线程读操作,也只能一个一个读,效率低了

package com.ffyc.javaPro.thread.juc;

import java.util.ArrayList;

public class CopyOnWriteArrayListDemo {

    public static void main(String[] args) {

        ArrayList arrayList = new ArrayList();
        for (int i = 0; i <100000 ; i++) {
            new Thread(){
                @Override
                public void run() {
                    arrayList.add("aaa");
                    System.out.println(arrayList);
                }
            }.start();
        }
    }
}

import java.util.Vector;

public class CopyOnWriteArrayListDemo {

    public static void main(String[] args) {
        Vector arrayList = new Vector();
        for (int i = 0; i <100000 ; i++) {
            new Thread(){
                @Override
                public void run() {
                    arrayList.add("aaa");
                    arrayList.get(i);
                    System.out.println(arrayList);
                }
            }.start();
        }
    }
}

不报错，但是效率低

CopyOnWriteArrayList:

将读写并发效率进一步提升了.
读操作(get())是完全不加锁的,
只给能改变数据的方法(add,set,remove)进行了加锁,而且为了操作时,不影响读操作,
操作前现将数组进行拷贝,在副本上修改,修改之后,将副本重新赋值到底层数组.

做到了只有写写是互斥的, 读写,读读都不互斥

适用于,读操作多,写操作少场景

import java.util.concurrent.CopyOnWriteArrayList;

public class CopyOnWriteArrayListDemo {

    public static void main(String[] args) {

        CopyOnWriteArrayList arrayList = new CopyOnWriteArrayList();
        for (int i = 0; i <100000 ; i++) {
            new Thread(){
                @Override
                public void run() {
                    arrayList.add("aaa");
                    arrayList.get(i);
                    System.out.println(arrayList);
                }
            }.start();
        }
    }
}

CopyOnWriteArraySet

CopyOnWriteArraySet 的实现基于 CopyOnWriteArrayList,不能存储重复数据。

import java.util.concurrent.CopyOnWriteArraySet;

public class CopyOnWriteArrayListDemo {

    public static void main(String[] args) {

        CopyOnWriteArraySet arrayList = new CopyOnWriteArraySet();
        for (int i = 0; i <100000 ; i++) {
            new Thread(){
                @Override
                public void run() {
                    arrayList.add("aaa");
                    arrayList.set()
                    arrayList.get(i);
                    System.out.println(arrayList);
                }
            }.start();
        }
    }
}

辅助类 CountDownLatch

CountDownLatch 辅助类递减计数器
使一个线程等待其他线程执行结束后再执行
相当于一个线程计数器,是一个递减的计数器
先指定一个数量,当有一个线程执行结束后就减一直到为0 关闭计数器
这样线程就可以执行了

package com.ffyc.javaPro.thread.juc;

import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {

    public static void main(String[] args) throws InterruptedException {

       CountDownLatch downLatch = new CountDownLatch(6);//计数
        for (int i = 0; i <6 ; i++) {
            new Thread(
                ()->{
                    System.out.println(Thread.currentThread().getName());
                    downLatch.countDown();//计数器减一操作
                }
            ).start();
        }
        downLatch.await();//关闭计数

        System.out.println("main线程执行");
    }
}

线程池

字符串常量池

string a = "abc";

string b = "abc";

a==b;//true

IntegerCache.cache -128---+127进行缓存自动装箱

数据库连接池未来避免重复创建连接对象和销毁连接对象，实现创建若干个连接对象

使用jdbc

package com.ffyc.javaPro.thread.dbconnection.jdbcdemo;

import org.junit.Test;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Date;

public class TestJDBC {
    @Test
    public void test() throws SQLException {
        Date date1 = new Date();
        for (int i = 0; i < 5000; i++) {
            Connection connection = JdbcUtil.getConnection();
            System.out.println(connection);
            JdbcUtil.close(connection);
        }
        Date date2 = new Date();
        System.out.println(date2.getTime()-date1.getTime());//23476
    }
}

使用阿里巴巴数据源

import org.junit.Test;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Date;

public class TestDruid {

    @Test
    public void test() throws SQLException {
        Date date1 = new Date();
        for (int i = 0; i < 5000; i++) {
            Connection connection = DruidUtil.getConnection();
            System.out.println(connection);
            DruidUtil.close(connection);
        }
        Date date2 = new Date();
        System.out.println(date2.getTime()-date1.getTime());//851
    }
}

数据库连接池

public class DataConfig {
    static final  String URL = "jdbc:mysql://127.0.0.1:3306/dormdb?serverTimezone=Asia/Shanghai";
    static final  String JDBC_DRIVER = "com.mysql.cj.jdbc.Driver";
    static final  String JDBC_USER_NAME = "root";
    static final  String JDBC_PASSWORD = "root";
    static final  int POOL_SIZE = 10;

}

封装Connection
数据库连接管道，就是对JDBC Connection进行封装而已，但是需要注意useState的这个标示。
连接池中的关闭连接实际上是将连接放回到连接池中以便其他使用者复用，实际上只是标示的改变而已

package com.ffyc.javaPro.thread.dbconnection.myconnectionpool;

import java.sql.Connection;

public class MyConnection{

    private  Connection connection;//接收一个真正的连接对象
    private boolean state = false; //false-未使用, true-使用


    public MyConnection(Connection connection,boolean state) {
        this.connection = connection;
        this.state = state;
    }

    /**
     * 关闭连接,本质是修改标识
     */
    public void close() {
        this.state = false;
    }


    public boolean getState() {
        return state;
    }

    public void setState(boolean state) {
        this.state = state;
    }
}

package com.ffyc.javaPro.thread.dbconnection.myconnectionpool;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Vector;

/**
 * 自定义连接池对象
 */
public class MyConnectionPool {

    /*
       连接池容器
     */
    private Vector<MyConnection> myConnections = new Vector();

    // 数据库驱动
    private String jdbcDriver;
    // 数据库访问地址
    private String jdbcURL;
    // 数据库连接用户名
    private String jdbcUsername;
    // 数据库连接密码
    private String jdbcPassword;
    // 数据库连接池大小
    private int poolSize;

    /*
      构造方法,初始化整个数据库连接池
     */
    public MyConnectionPool() {
            init();
            createMyPooledConnection();
    }
     /*
        初始化数据库连接信息
      */
    private void init() {
        //默认初始化数据库信息,正常情况从配置文件读取过来, 初始化数量和最大数量由构造方法指定
        this.jdbcDriver =  DataConfig.JDBC_DRIVER;
        this.jdbcURL = DataConfig.URL;
        this.jdbcUsername = DataConfig.JDBC_USER_NAME;
        this.jdbcPassword = DataConfig.JDBC_PASSWORD;
        this.poolSize = DataConfig.POOL_SIZE;
        // 加载数据库驱动程序
        try {
            Class.forName(this.jdbcDriver);
            System.out.println("驱动加载成功");
        } catch (ClassNotFoundException e) {
            System.out.print("驱动加载失败");
            e.printStackTrace();
        }
    }

    /*
       创建数据库连接池
     */
    private void createMyPooledConnection() {
        //创建指定数量的数据库连接
        for (int i = 0; i < poolSize; ++i) {
            try {
                //创建连接对象
                Connection connection = DriverManager.getConnection(jdbcURL, jdbcUsername, jdbcPassword);
                //将连接对象封装到自定义连接对象中, 设置状态为未使用
                MyConnection myPooledConnection = new MyConnection(connection, false);
                //将连接对象添加到连接池中
                myConnections.add(myPooledConnection);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    //如果得不到操作管道，需要去创建管道！
    public synchronized MyConnection getMyPooledConnection() {
        MyConnection myPooledConnection = null;
        try {
            //从连接池中获取一个连接对象
            myPooledConnection = getRealConnectionFromPool();
            //如果获取为空,说明连接池没有空闲连接,则循环继续获得,直到获得一个连接对象
            while (myPooledConnection == null) {
                myPooledConnection = getRealConnectionFromPool();
                if(myPooledConnection!=null){
                    return myPooledConnection;//获得到连接对象,直接返回,结束循环
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return myPooledConnection;
    }

    /**
     * 真正执行从连接池中获取连接对象
     * @return
     * @throws SQLException
     */
    private synchronized MyConnection getRealConnectionFromPool() throws SQLException {
        MyConnection myConnection = null;
        //循环连接池集合
        for (MyConnection connection : myConnections) {
            //如果状态为未使用
            if (!connection.getState()) {
                    try {
                        //将连接对象状态改为使用中
                        connection.setState(true);
                        myConnection = connection;//获取到连接对象
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                return myConnection;
            }
        }
        return null;
    }


}

package com.ffyc.javaPro.thread.dbconnection.myconnectionpool;

import java.util.Date;

public class Test {


    public static void main(String[] args) {
         MyConnectionPool myConnectionPool = new MyConnectionPool();//创建了自己的数据库连接池
        Date date1 = new Date();
        for (int i = 0; i < 5000; i++) {
            new Thread(()->{
                // 从连接池中申请获取一个连接
                MyConnection myConnection = myConnectionPool.getMyPooledConnection();
                System.out.println(myConnection);
                myConnection.close();
            }).start();
        }
        Date date2 = new Date();
        System.out.println(date2.getTime()-date1.getTime());//430
    }
}

线程池

有时，有许多任务需要执行，而且每个任务都比较短，这种场景下，需要大量创建线程，

这样依赖创建的开销就变大了

可以事先创建一部分线程，不销毁，有任务时提交给线程去执行，执行完后不结束线程，避免了频繁的创建线程

池就是一个缓冲，可以事先准备好一些数据，用的时候直接使用即可，提高效率

package com.ffyc.javaPro.thread.pool;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class PoolDemo {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        executorService.submit(new Runnable() {//提交任务
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName());
            }
        });
    }
}

线程池优点

重复利用线程，降低线程创建和销毁带来的资源消耗

统一管理线程，线程的创建和销毁都由线程池进行管理

提高响应速度，线程创建已经完成，任务来到可直接处理，省去了创建时间

ThreadPoolExecutor

在jdk5之后，Java中就提供了线程池的实现类

Executors.newFixedThreadPool();

//阿里巴巴开发规约建议使用的
public ThreadPoolExecutor(int corePoolSize,
                         int maximumPoolSize,
                         long keepAliveTime,
                         TimeUnit unit,
                         BlockingQueue<Runnable> workQueue,
                         ThreadFactory threadFactory,
                         RejectedExecutionHandler handler) {

构造器各个参数含义：

corePoolSize：核心线程池数量 5 创建ThreadPoolExecutor对象后，其实线程数量为0，有任务到来时，才会创建新的线程放到线程，直到核心线程池数量达到设定的值

可以直接调用prestartAllCoreThreads()或者 prestartCoreThread()，创建线程对象后，就可以立即创建线程。
maximumPoolSize：线程池最大线程数 10
keepAliveTime：非核心线程池中的线程在没有任务执行时，保持空闲多久后销毁，时间到期后，可以销毁空闲的线程
unit：keepAliveTime时间单位
workQueue：一个阻塞队列，用来存放等待执行的任务
threadFactory：线程工厂，主要用来创建线程
handler：表示拒绝执行任务时的策略(拒绝策略)

线程池的执行

当任务到达时，首先在核心线程池创建线程任务，如果核心线程池未满，那么直接让核心线程池执行，如果核心线程池已经满了，那么就将任务存放到队列中，等待执行

当任务继续提交过来时，如果队列已经放满了，就看非核心线程池中的线程数量有没有达到最大线程数量

如果已经达到并且没有空闲的线程，那么就采取某种拒绝的策略

线程池中的队列

ArrayBlockingQueue 给定队列的数量

LinkedBlockingQueue

package com.ffyc.javaPro.thread.threadpool;

public class MyTask implements Runnable {

    private int taskNum;

    public MyTask(int num) {
        this.taskNum = num;
    }

    @Override
    public void run() {
        try {
            Thread.currentThread().sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()+"task "+taskNum+"执行完毕");
    }
}

package com.ffyc.javaPro.thread.threadpool;

import java.util.ArrayList;
import java.util.concurrent.*;

public class Test {
    public static void main(String[] args) {
        /*MyTask myTask1 = new MyTask(1);创建任务对象
        Thread thread = new Thread(myTask1);/./创建线程对象  提交任务
        thread.start(); 启动线程*/

        /*
           通过线程池执行任务
         */
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 5, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy());
        for(int i=1;i<8;i++){
            MyTask myTask = new MyTask(i);
            executor.execute(myTask);//添加任务到线程池
        }
    }
}

线程池中的拒绝策略

当线程池中线程和队列都已经装满时，继续到来的任务无法处理时，可以采取以下四种策略进行拒绝。

package com.ffyc.javaPro.thread.threadpool;

import java.util.ArrayList;
import java.util.concurrent.*;

public class Test {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 5, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy());
        for(int i=1;i<=8;i++){
            MyTask myTask = new MyTask(i);
            executor.execute(myTask);//添加任务到线程池
        }
    }
}

AbortPolicy 策略：直接抛出异常

CallerRunsPolicy 策略：让提交任务的线程去执行，例如main线程

DiscardOleddestPolicy 策略：丢弃等待时间最长的任务，将新来的任务添加进去

DiscardPolicy 策略：直接丢弃无法执行的任务

execute和submit的区别、

execute();提交任务但是不能接收返回值

submit();提交任务可以接收返回值

package com.ffyc.javaPro.thread.threadpool;

import java.util.ArrayList;
import java.util.concurrent.*;

public class Test {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 5, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy());
        for(int i=1;i<=8;i++){
            MyTask myTask = new MyTask(i);
            //executor.execute(myTask);//添加任务到线程池
            Future<?> submit = executor.submit(myTask);
        }
    }
}

关闭线程池

shutdow();关闭时，会把以及提交到线程池中的线程执行完

package com.ffyc.javaPro.thread.threadpool;

import java.util.ArrayList;
import java.util.concurrent.*;

public class Test {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 5, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy());
        for(int i=1;i<=8;i++){
            MyTask myTask = new MyTask(i);
            Future<?> submit = executor.submit(myTask);
        }
        executor.shutdown();
    }
}

shutdowNow();立即关闭线程池，为执行的线程也会被中断

package com.ffyc.javaPro.thread.threadpool;

import java.util.ArrayList;
import java.util.concurrent.*;

public class Test {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 5, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy());
        for(int i=1;i<=8;i++){
            MyTask myTask = new MyTask(i);
            Future<?> submit = executor.submit(myTask);
        }
        executor.shutdownNow();
    }
}

ThreadLocal

需求: 多个线程中都有一个属于自己的num, 而不是多个线程共用同一个num

package com.ffyc.javaPro.thread.threadlocal;


public class Demo {

    static  int num = 0;

    public static void main(String[] args) {
         new Thread(){
             @Override
             public void run() {
                 num++;
                 System.out.println(num);
             }
         }.start();
         
         new Thread(){
             @Override
             public void run() {
                 num++;
                 System.out.println(num);
              }
        }.start();
    }
}

创建一个ThreadLocal对象,用来为每个线程会复制保存一份变量,实现线程封闭

package com.ffyc.javaPro.thread.threadlocal;


public class ThreadLocalDemo {

    private  static ThreadLocal<Integer> localNum = new ThreadLocal<Integer>(){
        @Override
        protected Integer initialValue() {
            return 0;
        }
    };


    public static void main(String[] args) {
          new Thread(){
              @Override
              public void run() {
                  localNum.set(1);
                  try {
                      Thread.sleep(2000);
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  }
                  localNum.set(localNum.get()+10);
                  System.out.println(Thread.currentThread().getName()+":"+localNum.get());//11
              }
          }.start();

         new Thread(){
            @Override
            public void run() {
                 localNum.set(3);
                 try {
                     Thread.sleep(2000);
                 } catch (InterruptedException e) {
                     e.printStackTrace();
                 }
                 localNum.set(localNum.get()+20);
                 System.out.println(Thread.currentThread().getName()+":"+localNum.get());//23
             }
        }.start();

        System.out.println(Thread.currentThread().getName()+":"+localNum.get());//0
    }
}

原理分析

ThreadLocal为每一个线程提供变量副本

为每一个线程创建ThreadLocalMap对象，在ThreadLocalMap对象中存储线程自己的变量副本

对象四种引用

强引用

String s = new String();

String s1 = s; 有引用指向对象

s1=null; s=null;

软引用

被SoftReference对象管理的对象,在内存不够时,先不回收被SoftReference管理的对象,

先进行一次垃圾回收,当垃圾回收后,如果内存够用了,那就不会被SoftReference管理的对象,

如果回收后,内存还不够,那么就会回收被SoftReference管理的对象

SoftReference<byte[]> m = new SoftReference<>(new byte[10]);

弱引用

被WeakReference管理的对象, 只要遇到一次GC,就会被回收掉

WeakReference<String> m = new WeakReference<>(new String("我是弱引用"));

虚引用

ThreadLocal内存泄漏问题

ThreadLocal与弱引用WeakReference有关系，那么在垃圾回收时，会把键回收了，但是值还存在强引用，不能回收，造成内存泄漏问题

每次使用完 ThreadLocal 都调用它的 remove()方法清除数据。

          new Thread(){
              @Override
              public void run() {
                  localNum.set(1);
                  try {
                      Thread.sleep(2000);
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  }
                  localNum.set(localNum.get()+10);
                  localNum.remove();
                  System.out.println(Thread.currentThread().getName()+":"+localNum.get());//11
              }
          }.start();