深入理解java多线程技术

1.为什么使用多线程

Java 多线程技术是 Java 并发编程的核心，从基础的线程创建到高级的并发工具包（JUC），再到 JVM 底层的内存模型，是一个庞大而精密的体系。

简单来说，Java 多线程的核心目的只有两个：1. 充分利用硬件资源（特别是 CPU）；2. 提升系统的响应速度和吞吐量。

如果没有多线程，现代互联网高并发系统（如淘宝、微信、抖音）将完全无法运行。

以下从核心痛点、具体场景和直观对比三个维度详细拆解：

一、核心痛点：单线程的致命缺陷

在单线程模型中，程序是顺序执行的。如果前一个任务没做完，后一个任务必须等着。这会带来两个致命问题：

1. CPU 资源的极大浪费（针对 CPU 密集型）

• 现状：现在的服务器通常是多核 CPU（比如 8 核、16 核甚至更多）。

• 单线程问题：单线程程序同一时刻只能利用1 个 CPU 核心。剩下的 7/15 个核心都在“围观”，处于空闲状态。

• 后果：花了几万块买的服务器，实际只发挥了 1/8 的性能。

• 多线程解决：开启多个线程，让每个核心都跑起来，算力最大化。

2. 阻塞导致的系统“假死”（针对 IO 密集型）

• 现状：现代应用大部分时间花在IO 操作上（读数据库、调远程接口、读写文件、网络请求）。

• 速度差异：

  • CPU 计算速度：纳秒级（极快）。

  • 磁盘/网络 IO 速度：毫秒级（极慢，相差百万倍）。

• 单线程问题：

  • 线程 A 发起一个数据库查询（耗时 100ms）。

  • 在这 100ms 内，CPU 无事可做，只能傻等（阻塞）。

  • 此时如果有用户 B 访问系统，必须排队等待用户 A 的 100ms 结束。

  • 如果有 1000 个用户，每人等 100ms，最后一个人要等 100 秒！系统直接卡死。

• 多线程解决：

  • 线程 A 发起查询后，挂起等待（不占用 CPU）。

  • CPU 立即切换去执行线程 B、线程 C 的任务。

  • 等线程 A 的数据库结果回来了，CPU 再切回来处理。

  • 效果：CPU 永远在忙，没有空转，系统吞吐量成倍提升。

什么是线程

线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。Java 天生支持多线程。

线程的状态

Java 线程状态定义在Thread.State枚举中：

1. NEW: 新建，未启动 (start())
2. RUNNABLE: 运行中（包含就绪态和运行态，由 OS 调度）
3. BLOCKED: 阻塞，等待获取 synchronized 锁
4. WAITING: 无限期等待，等待其他线程通知 (wait(),join(),LockSupport.park())
5. TIMED_WAITING: 限期等待 (sleep(ms),wait(ms))
6. TERMINATED: 终止，执行结束

2.怎么使用多线程

2.1创建多线程的四种方式

2.1.1 继承Thread类

我们以经典的火车站卖票为例子：

public class UnsafeThreadTicket extends Thread{ //静台变量，所有线程共享 private static int totalTickets = 100; public UnsafeThreadTicket(String name) { super(name); } @Override public void run(){ while(true){ //判断是否有票 危险区域：没有 synchronized，没有 lock if (totalTickets > 0){ try { Thread.sleep(10); // 这里最容易发生线程切换 }catch (InterruptedException e){ e.printStackTrace(); } //2.买票 System.out.println(Thread.currentThread().getName()+"卖出了第"+totalTickets+"张票"); totalTickets--; } else { break; } } System.out.println(Thread.currentThread().getName()+"下班了，剩余票数"+totalTickets); } }

调用过程如下：

public class Demo { public static void main(String[] args) { UnsafeThreadTicket test1 = new UnsafeThreadTicket("窗口A"); UnsafeThreadTicket test2 = new UnsafeThreadTicket("窗口B"); UnsafeThreadTicket test3 = new UnsafeThreadTicket("窗口C"); test1.start(); test2.start(); test3.start(); } }

运行结果如下：

我们可以看到，对于共享资源，不加锁简直是灾难的

接下来我们来看看加锁版本的

public class UnsafeThreadTicket extends Thread{ //静台变量，所有线程共享 private static int totalTickets = 100; public UnsafeThreadTicket(String name) { super(name); } // 锁对象：也可以使用 static Object lock = new Object(); // 这里直接用类对象锁或者实例锁均可，因为 totalTickets 是 static 的，建议锁 class 对象或静态变量 private static final Object lock = new Object(); @Override public void run(){ while(true) { //判断是否有票 加锁 // 锁住 lock 对象，保证同一时刻只有一个线程能进入这块代码 synchronized (lock) { if (totalTickets <= 0) { break; // 票卖完了，退出循环 } try { Thread.sleep(10); // 这里最容易发生线程切换 } catch (InterruptedException e) { e.printStackTrace(); } //2.买票 System.out.println(Thread.currentThread().getName() + "卖出了第" + totalTickets + "张票"); totalTickets--; } } System.out.println(Thread.currentThread().getName()+"下班了，剩余票数"+totalTickets); } }

执行如下：

业务正常，多线程访问共享资源一定要加锁

2.1.2实现Runnable类

先不加锁为例子：

public class UnsafeRunnableTask implements Runnable { private int totalTickets = 100; @Override public void run() { while (true) { if (totalTickets > 0) { try { Thread.sleep(5);//制造竞争条件 } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + "卖出了第" + totalTickets + "张票"); totalTickets--; } else { break; } } System.out.println(Thread.currentThread().getName() + "下班了,剩余" + totalTickets); } }

main函数代码如下：

public class Demo { public static void main(String[] args) { UnsafeRunnableTask task = new UnsafeRunnableTask(); Thread t1 = new Thread(task,"窗口A"); Thread t2 = new Thread(task,"窗口B"); Thread t3 = new Thread(task,"窗口C"); t1.start(); t2.start(); t3.start(); } }

执行结果如下：

同样的，会有严重的超卖问题

接下来我们改善一下，加锁版本如下：

public class UnsafeRunnableTask implements Runnable { private int totalTickets = 100; // 显式锁 private final ReentrantLock lock = new ReentrantLock(); @Override public void run() { while (true) { lock.lock(); try { if (totalTickets <= 0) { break; } try { Thread.sleep(5);//制造竞争条件 } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + "卖出了第" + totalTickets + "张票"); totalTickets--; }finally { // 【关键】必须在 finally 中释放锁，否则一旦发生异常，锁永远不释放，系统死锁 lock.unlock(); } } System.out.println(Thread.currentThread().getName() + "下班了,剩余" + totalTickets); } }

测试如下：

2.1.3实现 Callable 接口

public class TicketCallable implements Callable<String> { //Callable通常用于独立任务，若要共享状态，需要外部传入共享对象 //为了演示，我们让外部传入一个共享的计数器对象 private final TicketCounter counter; private final String windowName; private int soldCount = 0; //记录本窗口卖出的数量 public TicketCallable(TicketCounter counter, String windowName) { this.counter = counter; this.windowName = windowName; } @Override public String call() throws Exception { while(true){ //调用共享对象中的加锁方法 int ticketNo = counter.sellTicket(); if (ticketNo == -1){ break; //无票 } soldCount++; System.out.println(windowName+"卖出了第"+ticketNo+"张票"); Thread.sleep(10); } return windowName +"共卖出："+soldCount+"张票"; } } class TicketCounter { private int totalTickets = 100; public int sellTicket() { synchronized (this) { if (totalTickets <=0) { return -1; } return totalTickets--; } } }

main函数调用

public class Demo { public static void main(String[] args) throws Exception{ TicketCounter counter = new TicketCounter(); TicketCallable c1 = new TicketCallable(counter, "窗口A"); TicketCallable c2 = new TicketCallable(counter, "窗口B"); TicketCallable c3 = new TicketCallable(counter, "窗口C"); //配合FutureTask FutureTask<String> f1 = new FutureTask<>(c1); FutureTask<String> f2 = new FutureTask<>(c2); FutureTask<String> f3 = new FutureTask<>(c3); new Thread(f1,"窗口A现成").start(); new Thread(f2,"窗口A现成").start(); new Thread(f3,"窗口A现成").start(); System.out.println("结果A"+f1.get()); System.out.println("结果B"+f2.get()); System.out.println("结果C"+f3.get()); } }

测试结果如下：

2.4线程池

public class DemoThreadPoolFixed { // 共享计数器 private static final AtomicInteger count = new AtomicInteger(100); public static void main(String[] args) { // 1. 创建线程池 ThreadPoolExecutor pool = new ThreadPoolExecutor( 3, 3, 50L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), new ThreadPoolExecutor.CallerRunsPolicy() ); try { // 2. 提交任务 for (int i = 1; i <= 3; i++) { final int windowId = i; pool.submit(() -> { String name = "窗口-" + windowId; while (true) { // decrementAndGet: 先减 1，再返回新值 int currentVal = count.decrementAndGet(); // 如果减完之后 < 0，说明减之前 <= 0，即没票了 // 注意：如果有多个线程同时抢到最后一张，只有一个能拿到 0，其他拿到 -1, -2... if (currentVal < 0) { break; // 没票了，退出循环 } // 能走到这里，说明 currentVal >= 0，代表卖出了第 (currentVal + 1) 张 System.out.println(name + " 卖出了第 " + (currentVal + 1) + " 张票"); // 模拟业务耗时 try { Thread.sleep(10); } catch (InterruptedException e) { Thread.currentThread().interrupt(); // 恢复中断状态 break; } } System.out.println(name + " 下班了"); }); } } finally { // 3. 【关键修复】关闭线程池，否则主线程结束后程序不会退出 pool.shutdown(); // 可选：等待所有任务执行完毕后再彻底关闭（防止任务没跑完程序就退了） try { if (!pool.awaitTermination(60, TimeUnit.SECONDS)) { pool.shutdownNow(); } } catch (InterruptedException e) { pool.shutdownNow(); Thread.currentThread().interrupt(); } } } }

现在有这样一个需求，我有几千万的商品api,有六千多个商品池（三级分类id）,包含商品id,商品信息，价格都需要调用单独的接口，需要快速导入到我们数据库里，这时候肯定要使用多线程，示例代码如下：

@Component @Slf4j public class SpuSyncTask { /** * 注入持久层对象 */ @Resource private JdSkuDao jdSkuDao; @Resource private JdSpuDao jdSpuDao; @Resource @Qualifier("jdSpuCategoryDao") private JdSpuCategoryDao jdSpuCategoryDao; @Resource private SkuDao skuDao; @Resource private SpuDao spuDao; @Resource private SpuCategoryDao spuCategoryDao; @Resource private CategoryDao categoryDao; @Resource private StringRedisTemplate template; /** * syncSkuData 专用线程池：4 核 8G 服务器，I/O 为主，按 CPU*2 官方推荐设为 8 线程 */ private static final int SKU_SYNC_THREADS = 8; private static final int SKU_SYNC_QUEUE_CAPACITY = 200; private final ExecutorService skuSyncExecutor = new ThreadPoolExecutor( SKU_SYNC_THREADS, SKU_SYNC_THREADS, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(SKU_SYNC_QUEUE_CAPACITY), new ThreadPoolExecutor.CallerRunsPolicy() ); // @Scheduled(cron = "30 50 16 07 12 ?") public R syncSkuData() throws Exception { List<GetSkuPoolInfoItemGoodsResp> skuPoolList = ReadExcel("/data/project/backend/employeebenefits/jdvop/商品池分类.xlsx"); log.info("syncSkuData 定时任务开始，商品池数量: {}", skuPoolList.size()); String token = template.opsForValue().get(Constants.JdVopRedisKey); JdClient jdClient = JDClientSingleton.getInstance(Constants.SERVER_URL, token, Constants.APP_KEY, Constants.APP_SECRET); processSkuPoolBatch(skuPoolList, jdClient); log.info("syncSkuData 定时任务完成"); return R.ok(); } /** 获取当前 JdClient（每次从 Redis 取 token） */ private JdClient getJdClient() { String token = template.opsForValue().get(Constants.JdVopRedisKey); return JDClientSingleton.getInstance(Constants.SERVER_URL, token, Constants.APP_KEY, Constants.APP_SECRET); } /** Token 过期时同步刷新，避免多线程重复刷新 */ private JdClient refreshTokenAndGetClient() { synchronized (JdAccess.class) { try { JdAccess.getToken(); } catch (Exception e) { log.error("syncSkuData 刷新 token 失败", e); } } return getJdClient(); } /** * 处理单个 SKU（可在线程中执行）：查详情、查价格、落库 */ private void processSingleSku(Long skuId, JdClient client) { try { // 3. 查询商品详情 VopGoodsGetSkuDetailInfoRequest getGoodsRequest = new VopGoodsGetSkuDetailInfoRequest(); getGoodsRequest.setSkuId(skuId); VopGoodsGetSkuDetailInfoResponse getGoodsResponse = client.execute(getGoodsRequest); String resultCode = getGoodsResponse.getCode(); if ("18".equals(resultCode) || "19".equals(resultCode)) { client = refreshTokenAndGetClient(); getGoodsResponse = client.execute(getGoodsRequest); } GetSkuPoolInfoGoodsResp result = getGoodsResponse.getOpenRpcResult().getResult(); if (result == null) { return; } // 4. 查询商品价格 VopGoodsGetSellPriceRequest goodsPriceReq = new VopGoodsGetSellPriceRequest(); goodsPriceReq.setSkuId(String.valueOf(skuId)); VopGoodsGetSellPriceResponse goodsPriceResp = client.execute(goodsPriceReq); resultCode = goodsPriceResp.getCode(); if ("18".equals(resultCode) || "19".equals(resultCode)) { client = refreshTokenAndGetClient(); goodsPriceResp = client.execute(goodsPriceReq); } if (goodsPriceResp.getOpenRpcResult().getResult() == null) { return; } JdSku sku = new JdSku(); sku.setSkuId(result.getSkuId()); sku.setSkuName(result.getSkuName()); sku.setSkuStatus(result.getSkuState()); sku.setSpuId(result.getSpuId()); sku.setImagePath(result.getImagePath()); sku.setSelfSellType(result.getSelfSellType()); sku.setSalePrice(goodsPriceResp.getOpenRpcResult().getResult().get(0).getSalePrice()); sku.setJdPrice(goodsPriceResp.getOpenRpcResult().getResult().get(0).getJdPrice()); sku.setLowestBuy(result.getLowestBuy() == null ? 1 : result.getLowestBuy()); sku.setSaleUnit(result.getSaleUnit()); JdSpu spu = new JdSpu(); spu.setSpuId(result.getSpuId()); spu.setName(result.getSpuName()); spu.setBrandName(result.getBrandName()); spu.setImagePath(result.getImagePath()); spu.setLogisticsType(result.getLogisticsType()); spu.setSalePrice(goodsPriceResp.getOpenRpcResult().getResult().get(0).getSalePrice()); jdSpuDao.upsertSpuData(spu); jdSkuDao.upsertSkuData(sku); log.debug("添加商品 id: {}", result.getSpuId()); // 同步商品分类 List<JdSpuCategory> categories = new ArrayList<>(); String categoryIdsStr = result.getCategory(); if (categoryIdsStr != null && !categoryIdsStr.isEmpty()) { for (String categoryId : categoryIdsStr.split(";")) { int categoryIdInt = Integer.parseInt(categoryId.trim()); JdSpuCategory category = new JdSpuCategory(); category.setCategoryId(categoryIdInt); category.setSpuId(result.getSpuId()); categories.add(category); } if (!categories.isEmpty()) { jdSpuCategoryDao.batchInsert(categories); } } } catch (Exception e) { log.warn("processSingleSku 失败 skuId={}", skuId, e); } } //插入全部商品 private void processSkuPoolBatch(List<GetSkuPoolInfoItemGoodsResp> skuPoolBatch, JdClient jdClient) { for (GetSkuPoolInfoItemGoodsResp item : skuPoolBatch) { long offset = 0; while (true) { VopGoodsQuerySkuByPageRequest vopGoodsRequest = new VopGoodsQuerySkuByPageRequest(); vopGoodsRequest.setBizPoolId(item.getBizPoolId()); vopGoodsRequest.setOffset(offset); vopGoodsRequest.setPageSize(10); try { VopGoodsQuerySkuByPageResponse vopGoodsResponse = jdClient.execute(vopGoodsRequest); String resultCode = vopGoodsResponse.getCode(); log.debug("分页查询状态: {}", resultCode); if ("18".equals(resultCode) || "19".equals(resultCode)) { jdClient = refreshTokenAndGetClient(); vopGoodsResponse = jdClient.execute(vopGoodsRequest); } if (vopGoodsResponse.getOpenRpcResult().getResult() == null) { break; } List<Long> skus = vopGoodsResponse.getOpenRpcResult().getResult().getSkus(); if (skus.isEmpty()) { break; } // 多线程并行处理当前页的 SKU List<CompletableFuture<Void>> futures = skus.stream() .map(skuId -> CompletableFuture.runAsync( () -> processSingleSku(skuId, getJdClient()), skuSyncExecutor)) .collect(Collectors.toList()); CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); offset = skus.get(skus.size() - 1); if (vopGoodsResponse.getOpenRpcResult().getResult().getRemainPage() <= 0) { break; } } catch (Exception e) { log.error("processSkuPoolBatch 分页异常 bizPoolId={}, offset={}", item.getBizPoolId(), offset, e); break; } } } } }

关于线程池的七个参数：

public ThreadPoolExecutor( int corePoolSize, // 1. 核心线程数 int maximumPoolSize, // 2. 最大线程数 long keepAliveTime, // 3. 非核心线程存活时间 TimeUnit unit, // 4. 时间单位 BlockingQueue<Runnable> workQueue, // 5. 任务队列 ThreadFactory threadFactory, // 6. 线程工厂 (可选) RejectedExecutionHandler handler // 7. 拒绝策略 )

想象一个银行大厅：

1. corePoolSize(核心线程数)：正式柜员数量。

   • 即使没有客户，这些柜员也一直在岗（除非设置了allowCoreThreadTimeOut）。

   • 任务来了，优先交给他们。

2. workQueue(任务队列)：等候区的椅子数量。

   • 当正式柜员都忙不过来时，新来的客户先在椅子上排队等待。

   • 如果椅子坐满了，就要叫临时工了。

3. maximumPoolSize(最大线程数)：正式柜员 + 临时柜员的总上限。

   • 当等候区（队列）满了，银行会开启临时窗口（非核心线程）。

   • 一旦临时窗口也开到了上限，再来的客户就会被拒绝。

   • 注意：只有当队列满了之后，才会创建超过 corePoolSize 的线程。

4. keepAliveTime+unit(存活时间)：临时工裁员时限。

   • 如果临时柜员（非核心线程）在这么长时间内没活干，就会被辞退（线程销毁），只保留正式柜员。

5. threadFactory(线程工厂)：招聘专员。

   • 用来创建新线程的。通常用来给线程起有意义的名字（如 "Order-Thread-1"），方便出问题时排查。

6. handler(拒绝策略)：满员后的处理方式。当正式工忙、临时工满、椅子也坐满了，新来的客户怎么办？

当队列满且线程达到最大值时，触发拒绝策略：

1. AbortPolicy(默认)：
   • 行为：直接抛出RejectedExecutionException异常。
   • 后果：如果不捕获，程序会崩溃。
   • 适用：不允许丢失任务，且希望快速感知系统过载的场景。
2. CallerRunsPolicy(推荐用于重要业务)：
   • 行为：谁提交的任務，谁自己去执行（调用者线程执行）。
   • 后果：提交任务的主线程（或线程池中的线程）会被阻塞，不再提交新任务，从而降低提交速度，给后端处理争取时间。
   • 适用：不允许丢失任务，且希望系统自动降速保护的场景（如你的卖票系统）。
3. DiscardPolicy：
   • 行为：直接丢弃任务，不抛异常。
   • 适用：任务丢失也没关系的场景（如日志记录、非关键监控数据）。
4. DiscardOldestPolicy：
   • 行为：丢弃队列中最老的一个任务，然后尝试重新提交当前任务。
   • 适用：希望尽量处理最新数据的场景。

《阿里巴巴 Java 开发手册》中强制规定：线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式。

// 1. FixedThreadPool: 允许请求队列长度为 Integer.MAX_VALUE // 风险：堆积大量请求，导致 OOM (内存溢出) ExecutorService executor = Executors.newFixedThreadPool(10); // 2. CachedThreadPool: 允许创建线程数量为 Integer.MAX_VALUE // 风险：创建大量线程，导致 CPU 100% 或 OOM ExecutorService executor = Executors.newCachedThreadPool(); // 3. SingleThreadExecutor: 允许请求队列长度为 Integer.MAX_VALUE // 风险：同上，OOM ExecutorService executor = Executors.newSingleThreadExecutor();

原因：Executors创建的线程池，其队列长度或最大线程数默认为Integer.MAX_VALUE。在高并发场景下，如果任务处理速度慢于提交速度，内存会瞬间爆满，导致服务器崩溃。

2.2多线程使用问题

2.2.1线程安全

多线程最大的问题是线程安全，主要源于：原子性、可见性、有序性。

1. 三大问题详解

• 原子性：操作要么全部成功，要么全部失败，不可中断。

  • 反例：i++不是原子操作（读->改->写）。

• 可见性：一个线程修改了变量，其他线程能立即看到。

  • 原因：CPU 缓存不一致。

• 有序性：指令重排序导致执行顺序与代码顺序不一致。

  • 原因：编译器和处理器优化。

解决办法：加锁