Java并发编程之锁、并发容器、阻塞队列与异步编程实战代码

// ReentrantLock.NonfairSync.tryAcquire()核心逻辑
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 锁空闲，CAS尝试获取
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        // 同一线程重入，state累加
        int nextc = c + acquires;
        if (nextc < 0) throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

public void lockInterruptibly() throws InterruptedException {
    sync.acquireInterruptibly(1);
}

// AQS.acquireInterruptibly()
public final void acquireInterruptibly(int arg) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (!tryAcquire(arg))
        doAcquireInterruptibly(arg); // 可中断的等待逻辑
}

// 超时获取锁使用示例
public boolean tryLockWithTimeout(ReentrantLock lock, long timeoutMs) {
    try {
        return lock.tryLock(timeoutMs, TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        return false;
    }
}

ReentrantLock fairLock = new ReentrantLock(true);    // 公平锁
ReentrantLock unfairLock = new ReentrantLock(false); // 非公平锁（默认）

/**
 * ReentrantLock完整使用示例
 */
public class ReentrantLockDemo {
    private final ReentrantLock lock = new ReentrantLock(true);
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    private final Queue<Integer> queue = new LinkedList<>();
    private static final int CAPACITY = 10;

    public void produce(int data) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == CAPACITY) {
                notFull.await(); // 队列满，生产者等待
            }
            queue.offer(data);
            notEmpty.signal(); // 唤醒消费者
        } finally {
            lock.unlock(); // 必须在finally中释放锁
        }
    }

    public int consume() throws InterruptedException {
        lock.lockInterruptibly(); // 可中断方式获取锁
        try {
            while (queue.isEmpty()) {
                notEmpty.await();
            }
            int data = queue.poll();
            notFull.signal();
            return data;
        } finally {
            lock.unlock();
        }
    }
}

ConcurrentHashMap
└── Segment[] (默认16个)
    └── HashEntry[] (每个Segment独立的哈希表)
        └── HashEntry链表

// JDK8 ConcurrentHashMap.putVal()核心逻辑
final V putVal(K key, V value, boolean onlyIfAbsent) {
    int hash = spread(key.hashCode());
    for (Node<K,V>[] tab = table;;) {
        int i = (n - 1) & hash;
        Node<K,V> f = tabAt(tab, i);
        
        if (f == null) {
            // 桶为空，CAS直接插入
            if (casTabAt(tab, i, null, new Node(hash, key, value)))
                break;
        } else {
            synchronized (f) { // 仅锁定桶的首节点
                if (tabAt(tab, i) == f) { // 双重检查
                    // 链表或红黑树插入逻辑
                }
            }
        }
    }
}

// 扩容时的ForwardingNode标记
static final class ForwardingNode<K,V> extends Node<K,V> {
    final Node<K,V>[] nextTable;
    ForwardingNode(Node<K,V>[] tab) {
        super(MOVED, null, null, null);
        this.nextTable = tab;
    }
}

// ArrayBlockingQueue核心结构
public class ArrayBlockingQueue<E> {
    final Object[] items;      // 存储数组
    int takeIndex;             // 取元素位置
    int putIndex;              // 放元素位置
    int count;                 // 元素数量
    final ReentrantLock lock;  // 单锁
    private final Condition notEmpty;
    private final Condition notFull;
}

// LinkedBlockingQueue双锁设计
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();

// SynchronousQueue典型用法
SynchronousQueue<Integer> queue = new SynchronousQueue<>();
// 生产者线程
new Thread(() -> {
    queue.put(1); // 会阻塞直到有消费者take
}).start();
// 消费者线程
new Thread(() -> {
    queue.take(); // 会阻塞直到有生产者put
}).start();

// 延时任务元素定义
public class DelayedTask implements Delayed {
    private final long executeTime;
    private final Runnable task;

    public DelayedTask(long delayMs, Runnable task) {
        this.executeTime = System.currentTimeMillis() + delayMs;
        this.task = task;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed other) {
        return Long.compare(this.executeTime, ((DelayedTask)other).executeTime);
    }
}

// 延时队列使用示例
DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();
delayQueue.put(new DelayedTask(30000, () -> System.out.println("30秒后执行")));
delayQueue.put(new DelayedTask(60000, () -> System.out.println("60秒后执行")));

// 消费者线程
while (true) {
    DelayedTask task = delayQueue.take(); // 阻塞直到任务到期
    task.run();
}

// 使用默认线程池
CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> System.out.println("异步任务"));
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "返回结果");

// 使用自定义线程池（推荐）
ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "自定义线程池", executor);

// 手动完成
CompletableFuture<String> future4 = new CompletableFuture<>();
future4.complete("手动设置结果");
future4.completeExceptionally(new RuntimeException("手动异常"));

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
    .thenApply(s -> s + " World")           // 同步转换
    .thenApplyAsync(s -> s.toUpperCase())   // 异步转换
    .thenAccept(s -> System.out.println(s)) // 消费结果
    .thenRun(() -> System.out.println("执行完成")); // 仅执行，不消费结果

// AND组合：两个都完成才执行
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "World");

f1.thenCombine(f2, (s1, s2) -> s1 + " " + s2)
  .thenAccept(System.out::println); // 输出 Hello World

// OR组合：任意一个完成就执行
CompletableFuture<String> fast = f1.applyToEither(f2, s -> s + " faster");

CompletableFuture.supplyAsync(() -> {
    if (true) throw new RuntimeException("出错了");
    return "正常";
})
.exceptionally(ex -> {
    System.out.println("捕获异常: " + ex.getMessage());
    return "默认值"; // 异常时返回默认值
})
.handle((result, ex) -> {
    if (ex != null) {
        return "处理异常";
    }
    return result;
})
.whenComplete((result, ex) -> {
    // 无论成功失败都会执行，不改变结果
    System.out.println("执行完成");
});

// 所有任务都完成
CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3);

// 任意一个任务完成
CompletableFuture<Object> any = CompletableFuture.anyOf(f1, f2, f3);

/**
 * 并行调用多个服务，聚合结果
 */
public class CompletableFutureDemo {
    public UserInfo getUserInfo(Long userId) {
        // 并行调用三个接口
        CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> userService.getUser(userId));
        CompletableFuture<List<Order>> orderFuture = CompletableFuture.supplyAsync(() -> orderService.getOrders(userId));
        CompletableFuture<List<Address>> addrFuture = CompletableFuture.supplyAsync(() -> addressService.getAddresses(userId));

        // 等待所有完成，聚合结果
        return CompletableFuture.allOf(userFuture, orderFuture, addrFuture)
            .thenApply(v -> {
                UserInfo info = new UserInfo();
                info.setUser(userFuture.join());
                info.setOrders(orderFuture.join());
                info.setAddresses(addrFuture.join());
                return info;
            })
            .exceptionally(ex -> {
                log.error("获取用户信息失败", ex);
                return null;
            })
            .join();
    }
}