🔗 线程通信实战

public class VolatileDemo { // ⚠️ volatile:保证可见性 // 没有 volatile 的话,子线程可能永远看不到主线程把 running 改成 false // (JIT 会把 running 缓存到寄存器/工作内存) private static volatile boolean running = true; public static void main(String[] args) throws InterruptedException { // 子线程:监控任务,running=true 时一直跑 new Thread(() -> { while (running) { // 监控逻辑 System.out.println("监控中..."); try { Thread.sleep(500); } catch (Exception e) {} } System.out.println("监控线程退出"); }).start(); // 主线程:等 2 秒后发停机信号 Thread.sleep(2000); running = false; // ⚠️ 子线程立刻看到,退出循环 System.out.println("主线程已发停机信号"); } }

public class WaitNotifyDemo { private static final List<String> buffer = new ArrayList<>(); private static final int MAX = 5; // 缓冲区上限 public static void main(String[] args) { // 生产者 new Thread(() -> { while (true) { synchronized (buffer) { // ⚠️ 必须 synchronized 同一把锁 while (buffer.size() == MAX) { // ⚠️ 用 while 不用 if(防虚假唤醒) try { buffer.wait(); } // ⚠️ wait 会释放锁 catch (InterruptedException e) { return; } } buffer.add("data-" + System.currentTimeMillis()); System.out.println("生产,size=" + buffer.size()); buffer.notifyAll(); // ⚠️ 通知消费者(用 notifyAll 不用 notify) } } }, "Producer").start(); // 消费者 new Thread(() -> { while (true) { synchronized (buffer) { while (buffer.isEmpty()) { try { buffer.wait(); } catch (InterruptedException e) { return; } } System.out.println("消费: " + buffer.remove(0)); buffer.notifyAll(); // 通知生产者 } } }, "Consumer").start(); } }

import java.util.concurrent.locks.*; public class ConditionDemo { private final ReentrantLock lock = new ReentrantLock(); // ⚠️ 两个条件:一个等"非满",一个等"非空" private final Condition notFull = lock.newCondition(); private final Condition notEmpty = lock.newCondition(); private final Object[] items = new Object[5]; private int count, putIdx, takeIdx; public void put(Object item) throws InterruptedException { lock.lock(); try { while (count == items.length) { notFull.await(); // ⚠️ 满了 → 等"非满"条件 } items[putIdx] = item; if (++putIdx == items.length) putIdx = 0; count++; notEmpty.signal(); // ⚠️ 精确通知"等非空"的线程 } finally { lock.unlock(); // ⚠️ 必须在 finally 释放 } } public Object take() throws InterruptedException { lock.lock(); try { while (count == 0) { notEmpty.await(); // ⚠️ 空了 → 等"非空"条件 } Object x = items[takeIdx]; items[takeIdx] = null; if (++takeIdx == items.length) takeIdx = 0; --count; notFull.signal(); // ⚠️ 精确通知"等非满"的线程 return x; } finally { lock.unlock(); } } }

import java.util.concurrent.*; public class AsyncLogger { // ⚠️ 有界队列:防止日志积压 OOM private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(10000); private volatile boolean running = true; public AsyncLogger() { // 消费线程:从队列取日志,写文件 new Thread(() -> { while (running || !queue.isEmpty()) { try { // ⚠️ take 队列空时阻塞,put 队列满时阻塞 String log = queue.take(); // 阻塞获取 writeToFile(log); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } }, "log-writer").start(); } // 生产者:业务线程调用 public void log(String msg) { // ⚠️ offer 不阻塞,队列满就丢弃(防业务线程被拖垮) // 如果用 put,队列满了业务线程会阻塞 — 别这么干 boolean ok = queue.offer(msg); if (!ok) { System.err.println("日志队列满,丢弃: " + msg); } } public void shutdown() { running = false; } private void writeToFile(String log) { /* 写文件 */ } }

import java.util.concurrent.*; public class CountDownLatchDemo { public static void main(String[] args) throws InterruptedException { // 场景:接口聚合 — 并行查 3 个服务,等全部返回后合并 int n = 3; // ⚠️ 初始计数 = 3,每完成一个 countDown 减 1 CountDownLatch latch = new CountDownLatch(n); ExecutorService pool = Executors.newFixedThreadPool(n); Map<String, Object> result = new ConcurrentHashMap<>(); for (int i = 0; i < n; i++) { final int idx = i; pool.submit(() -> { try { // 模拟调用下游服务 Thread.sleep(1000 + idx * 500); result.put("svc" + idx, "data-" + idx); System.out.println(Thread.currentThread().getName() + " 完成"); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { // ⚠️ 必须在 finally 里 countDown,防止异常导致主线程死等 latch.countDown(); } }); } // ⚠️ 主线程阻塞等待,最多等 5 秒(防止子任务卡死) boolean done = latch.await(5, TimeUnit.SECONDS); if (!done) { System.err.println("部分子任务超时"); } System.out.println("所有子任务完成,结果: " + result); pool.shutdown(); } }

import java.util.concurrent.*; public class CyclicBarrierDemo { public static void main(String[] args) { // 场景:3 个线程分阶段处理数据,每阶段都等齐后再进下一阶段 // ⚠️ 第二个参数:所有线程到齐后执行的回调 CyclicBarrier barrier = new CyclicBarrier(3, () -> { System.out.println("=== 所有线程到齐,进入下一阶段 ==="); }); ExecutorService pool = Executors.newFixedThreadPool(3); for (int i = 0; i < 3; i++) { final int id = i; pool.submit(() -> { try { System.out.println("线程" + id + " 阶段1开始"); Thread.sleep(1000 * (id + 1)); barrier.await(); // ⚠️ 等其他线程 System.out.println("线程" + id + " 阶段2开始"); Thread.sleep(500); barrier.await(); // ⚠️ CyclicBarrier 可复用(区别于 CountDownLatch) System.out.println("线程" + id + " 完成"); } catch (Exception e) { Thread.currentThread().interrupt(); } }); } pool.shutdown(); } }

import java.util.concurrent.*; @Service public class UserDetailService { @Autowired private UserClient userClient; @Autowired private OrderClient orderClient; @Autowired private PointClient pointClient; // ⚠️ 用独立的线程池,不要用 ForkJoinPool.commonPool(会被其他任务挤占) private final ExecutorService pool = Executors.newFixedThreadPool(10); public UserDetailVO getUserDetail(Long userId) throws Exception { // ① 三个任务并行发起(CompletableFuture.supplyAsync 异步执行) CompletableFuture<User> userFuture = CompletableFuture.supplyAsync( () -> userClient.getUser(userId), pool); CompletableFuture<List<Order>> orderFuture = CompletableFuture.supplyAsync( () -> orderClient.getOrders(userId), pool); CompletableFuture<Integer> pointFuture = CompletableFuture.supplyAsync( () -> pointClient.getPoints(userId), pool); // ② 等三个任务都完成,合并结果 // ⚠️ allOf 返回 CompletableFuture,join 阻塞等待全部完成 CompletableFuture.allOf(userFuture, orderFuture, pointFuture).join(); // ③ 取结果(此时一定已完成,join 不会阻塞) UserDetailVO vo = new UserDetailVO(); vo.setUser(userFuture.get()); // 已完成,立刻返回 vo.setOrders(orderFuture.get()); vo.setPoints(pointFuture.get()); return vo; } // 链式调用示例:查用户 → 查订单(依赖用户结果) public UserDetailVO getUserThenOrders(Long userId) { return CompletableFuture .supplyAsync(() -> userClient.getUser(userId), pool) .thenApplyAsync(user -> { // ⚠️ thenApply:用上一步结果继续算 List<Order> orders = orderClient.getOrders(user.getId()); UserDetailVO vo = new UserDetailVO(); vo.setUser(user); vo.setOrders(orders); return vo; }, pool) .exceptionally(ex -> { // ⚠️ 异常处理:任何一步失败都走这里 log.error("获取用户详情失败", ex); return new UserDetailVO(); // 降级返回空对象 }) .join(); // 阻塞获取最终结果 } }

public class ThreadLocalTrap { private static final ThreadLocal<String> userId = ThreadLocal.withInitial(() -> null); public static void main(String[] args) throws Exception { userId.set("user-001"); // 主线程 set System.out.println("主线程: " + userId.get()); // user-001 new Thread(() -> { // ⚠️ 子线程取到的是 null!ThreadLocal 线程隔离,父子也不通 System.out.println("子线程: " + userId.get()); // null }).start(); Thread.sleep(100); } }

public class InheritableDemo { // ⚠️ InheritableThreadLocal:子线程创建时会继承父线程的值 private static final InheritableThreadLocal<String> userId = new InheritableThreadLocal<>(); public static void main(String[] args) throws Exception { userId.set("user-001"); // ✅ 场景 1:new Thread() — 子线程能拿到 user-001 new Thread(() -> { System.out.println("new Thread 子线程: " + userId.get()); // user-001 ✅ }).start(); // ❌ 场景 2:线程池 — 子线程拿不到! ExecutorService pool = Executors.newFixedThreadPool(2); // ⚠️ 第一次提交:线程池新建 worker,此时继承父线程值 user-001 pool.submit(() -> System.out.println("pool 第1次: " + userId.get())); Thread.sleep(100); // ⚠️ 主线程改了值 userId.set("user-002"); // ⚠️ 第二次提交:worker 已存在(复用),不再触发继承 // 拿到的是 user-001(创建时的值),不是 user-002!❌ pool.submit(() -> System.out.println("pool 第2次: " + userId.get())); pool.shutdown(); } }

// 依赖:com.alibaba:transmittable-thread-local import com.alibaba.ttl.TransmittableThreadLocal; import com.alibaba.ttl.TtlRunnable; import com.alibaba.ttl.threadpool.TtlExecutors; public class TtlDemo { // ⚠️ 1. 用 TransmittableThreadLocal 替换 ThreadLocal private static final TransmittableThreadLocal<String> userId = new TransmittableThreadLocal<>(); public static void main(String[] args) throws Exception { userId.set("user-001"); // ⚠️ 2. 线程池要用 TtlExecutors 包装 ExecutorService pool = TtlExecutors.getTtlExecutorService(Executors.newFixedThreadPool(2)); pool.submit(() -> { System.out.println("第1次: " + userId.get()); // user-001 ✅ }); Thread.sleep(100); userId.set("user-002"); // 主线程改值 pool.submit(() -> { // ⚠️ TTL 在任务提交时"快照"父线程值,在任务执行时"回放" System.out.println("第2次: " + userId.get()); // user-002 ✅ }); pool.shutdown(); } } /* TTL 原理: 1. submit 时,TTL 把当前线程(提交者)的所有 TTL 变量快照一份 2. 任务执行前,TTL 把快照"回放"到执行线程 3. 任务执行后,TTL 把执行线程之前的值恢复(避免污染) ⚠️ 不用 TtlExecutors 包装的话,也可以用 TtlRunnable.get(runnable) 手动包装: pool.submit(TtlRunnable.get(() -> { ... })); */

import java.io.*; public class PipedDemo { public static void main(String[] args) throws Exception { PipedOutputStream out = new PipedOutputStream(); PipedInputStream in = new PipedInputStream(out); // 连接 // 线程 1:写管道 new Thread(() -> { try { out.write("hello from thread-1".getBytes()); out.close(); } catch (Exception e) {} }).start(); // 线程 2:读管道 new Thread(() -> { try { byte[] buf = new byte[1024]; int len = in.read(buf); System.out.println("收到: " + new String(buf, 0, len)); } catch (Exception e) {} }).start(); } } // ⚠️ 实际工程用 BlockingQueue 替代,API 更友好,功能更强