手写线程池与锁

public class SimpleThreadPool {
    private final BlockingQueue<Runnable> queue;
    private final List<Worker> workers = new ArrayList<>();
    private volatile boolean running = true;
    public SimpleThreadPool(int coreThreads, int cap){
        queue = new ArrayBlockingQueue<>(cap);
        for (int i=0;i<coreThreads;i++){ Worker w=new Worker(); workers.add(w); w.start(); }
    }
    public void execute(Runnable task) {
        if (!running) throw new RejectedExecutionException("已关闭");
        queue.offer(task); // 满则拒绝
    }
    public void shutdown() {
        running = false;
        for (Worker w : workers) w.interrupt(); // 中断等待
    }
    class Worker extends Thread {
        public void run() {
            while (running || !queue.isEmpty()) {
                try {
                    Runnable task = queue.poll(100, TimeUnit.MILLISECONDS);
                    if (task != null) task.run();
                } catch (InterruptedException e) { /* 退出 */ }
            }
        }
    }
}

public class ReentrantLockSimple {
    private Thread owner;
    private int count;
    public synchronized void lock() {
        Thread cur = Thread.currentThread();
        if (owner == null) { owner = cur; count = 1; }
        else if (owner == cur) count++; // 重入
        else { while (owner != null) { try { wait(); } catch (InterruptedException e){} } owner=cur; count=1; }
    }
    public synchronized void unlock() {
        if (--count == 0) { owner = null; notifyAll(); }
    }
}

ThreadLocal<Integer> tl = new ThreadLocal<>();
try {
    tl.set(1);
    int v = tl.get();
} finally {
    tl.remove(); // 显式清理，防止内存泄漏
}

// CountDownLatch 简化：AQS 共享锁思想
public class CountDownLatchSimple {
    private int count;
    public CountDownLatchSimple(int n){ count = n; }
    public synchronized void countDown(){ if (--count == 0) notifyAll(); }
    public synchronized void await() throws InterruptedException {
        while (count > 0) wait();
    }
}
// CyclicBarrier 简化：可重置
public class CyclicBarrierSimple {
    private final int parties;
    private int count;
    private int generation;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition cond = lock.newCondition();
    public CyclicBarrierSimple(int n){ parties = n; count = n; }
    public void await() throws InterruptedException {
        lock.lock();
        try {
            int myGen = generation;
            if (--count == 0) { generation++; count = parties; cond.signalAll(); } // 新一代
            else while (myGen == generation) cond.await();
        } finally { lock.unlock(); }
    }
}

public abstract class AbstractQueuedSynchronizerSimple {
    volatile int state;
    final class Node { Thread thread; Node prev, next; }
    volatile Node head = new Node(), tail = head;
    protected abstract boolean tryAcquire(int arg);
    protected abstract boolean tryRelease(int arg);
    public void acquire(int arg) {
        if (!tryAcquire(arg)) {
            Node n = new Node(); n.thread = Thread.currentThread();
            // 简化：入队 tail
            synchronized (this) { tail.next = n; n.prev = tail; tail = n; }
            for (;;) {
                if (n.prev == head && tryAcquire(arg)) { head = n; return; }
                LockSupport.park();
            }
        }
    }
    public void release(int arg) {
        if (tryRelease(arg)) {
            Node next = head.next;
            if (next != null) LockSupport.unpark(next.thread);
        }
    }
}

public class ReadWriteLockPriority {
    private int readers = 0, writers = 0, waitingWriters = 0;
    public synchronized void readLock() throws InterruptedException {
        while (writers > 0 || waitingWriters > 0) wait(); // 写者优先：有写者等待则读者等待
        readers++;
    }
    public synchronized void readUnlock(){ readers--; if (readers==0) notifyAll(); }
    public synchronized void writeLock() throws InterruptedException {
        waitingWriters++;
        try { while (readers>0 || writers>0) wait(); writers++; }
        finally { waitingWriters--; }
    }
    public synchronized void writeUnlock(){ writers--; notifyAll(); }
}

public class TokenBucket {
    private final int capacity;
    private final double rate; // 每秒补充令牌数
    private double tokens;
    private long lastRefill;
    public TokenBucket(int cap, double rate){ capacity=cap; this.rate=rate; tokens=cap; lastRefill=System.currentTimeMillis(); }
    public synchronized boolean tryAcquire() {
        long now = System.currentTimeMillis();
        tokens = Math.min(capacity, tokens + (now-lastRefill)/1000.0 * rate);
        lastRefill = now;
        if (tokens >= 1) { tokens -= 1; return true; }
        return false;
    }
}

public class SemaphoreSimple {
    private int permits;
    public SemaphoreSimple(int n){ permits = n; }
    public synchronized void acquire() throws InterruptedException {
        while (permits == 0) wait();
        permits--;
    }
    public synchronized void release(){ permits++; notify(); }
}

// 工作窃取思路：每个线程一个 Deque
public class WorkStealing {
    final Deque<Runnable>[] deques;
    final int nThreads;
    public WorkStealing(int n){ nThreads=n; deques=new Deque[n]; for(int i=0;i<n;i++) deques[i]=new ArrayDeque<>(); }
    public void submit(int tid, Runnable task){ deques[tid].push(task); } // 推入队尾
    public Runnable steal(int tid){
        for (int i=0;i<nThreads;i++){
            int other = (tid+i)%nThreads;
            Runnable t = deques[other].pollLast(); // 从队尾窃取
            if (t!=null) return t;
        }
        return null;
    }
}

public class TimedLock {
    private final AtomicBoolean locked = new AtomicBoolean();
    public boolean tryLock(long timeoutNanos) throws InterruptedException {
        long deadline = System.nanoTime() + timeoutNanos;
        while (!locked.compareAndSet(false, true)) {
            long remaining = deadline - System.nanoTime();
            if (remaining <= 0) return false;
            LockSupport.parkNanos(remaining); // 超时自动返回
            if (Thread.interrupted()) throw new InterruptedException(); // 响应中断
        }
        return true;
    }
    public void unlock(){ locked.set(false); } // 无等待者队列，等待者靠 parkNanos 超时后重试 CAS；若需真正唤醒应维护等待者队列
}

public class MonitoredThreadPool {
    private final LongAdder rejected = new LongAdder();
    private final BlockingQueue<Runnable> queue;
    public void execute(Runnable task) {
        if (!queue.offer(task)) { rejected.increment(); alert(); } // 队列满拒绝并告警
    }
    public long queueDepth() { return queue.size(); }
    public long rejectedCount() { return rejected.sum(); }
    private void alert(){ /* 超过阈值告警 */ }
}

public class KeepAliveWorker extends Thread {
    // allowCoreThreadTimeOut 与 alive：
    // 取任务：若允许核心超时，用 poll(keepAlive)；否则额外线程用 poll，核心线程用 take
    Runnable task = allowCoreTimeout || isExtra
        ? queue.poll(keepAliveNanos, NANOSECONDS)
        : queue.take();
    if (task == null) { /* 超时，回收线程 */ return; }
}

// 简化：任务就绪计数
public class TaskNode {
    final Runnable action;
    final List<TaskNode> deps = new ArrayList<>();
    final List<TaskNode> dependents = new ArrayList<>();
    int readyCount;
    public void runAndNotify() {
        try { action.run(); } catch (Exception e) { /* 失败传播 */ }
        for (TaskNode d : dependents) if (--d.readyCount == 0) schedule(d);
    }
}
// 对应 CompletableFuture
CompletableFuture.allOf(f1, f2).thenApply(...); // 都完成后再合并

public class RetryPolicy implements RejectedExecutionHandler {
    private final BlockingQueue<Runnable> backup;
    public RetryPolicy(BlockingQueue<Runnable> b){ backup = b; }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        try { backup.offer(r, 100, TimeUnit.MILLISECONDS); } // 重试入队到备用队列
        catch (InterruptedException ex) { Thread.currentThread().interrupt(); }
    }
}

public class ProducerConsumer {
    private final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(100);
    public void producer() throws InterruptedException {
        for (int i=0;i<1000;i++) queue.put(i); // 满则阻塞，不忙等
    }
    public void consumer() throws InterruptedException {
        while (true) {
            int v = queue.take(); // 空则阻塞
            process(v);
        }
    }
    void process(int v){}
}

public class FutureTaskSimple<V> {
    private volatile int state; // 0 NEW,1 RUNNING,2 DONE,3 CANCELLED
    private final Callable<V> callable;
    private V result;
    private Exception error;
    private final Object lock = new Object();
    public FutureTaskSimple(Callable<V> c){ callable=c; }
    public void run() {
        if (state != 0) return;
        state = 1;
        try { result = callable.call(); }
        catch (Exception e) { error = e; }
        finally { state = 2; synchronized(lock){ lock.notifyAll(); } }
    }
    public V get() throws Exception {
        synchronized(lock) {
            while (state != 2 && state != 3) lock.wait();
        }
        if (error != null) throw error;
        return result;
    }
    public boolean cancel(){ if (state==0){ state=3; synchronized(lock){lock.notifyAll();} return true;} return false; }
}

ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
// 写锁降级为读锁（安全）
rw.writeLock().lock();
try {
    // 写操作
    rw.readLock().lock(); // 先获取读锁
    rw.writeLock().unlock(); // 再释放写锁，降级为读锁
    // 继续以读锁读
} finally { rw.readLock().unlock(); }
// 读锁升级为写锁：ReentrantReadWriteLock 不支持，会死锁

public class TicketLock {
    private final AtomicInteger ticket = new AtomicInteger();
    private final AtomicInteger serving = new AtomicInteger();
    public int lock() {
        int my = ticket.getAndIncrement();
        while (serving.get() != my) { /* 自旋等待 */ }
        return my;
    }
    public void unlock(int my){ serving.compareAndSet(my, my+1); }
}

public boolean offer(T v, long timeout, TimeUnit unit) throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    lock.lock();
    try {
        while (count == items.length) {
            if (nanos <= 0) return false;
            nanos = notFull.awaitNanos(nanos); // 等待并在剩余时间后返回
        }
        enqueue(v); return true;
    } finally { lock.unlock(); }
}
public T poll(long timeout, TimeUnit unit) throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    lock.lock();
    try {
        while (count == 0) {
            if (nanos <= 0) return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        return dequeue();
    } finally { lock.unlock(); }
}