Java并发

public class DeadLockDemo {
    private static Object resource1 = new Object();//资源 1
    private static Object resource2 = new Object();//资源 2

    public static void main(String[] args) {
        new Thread(() -> {
            synchronized (resource1) {
                System.out.println(Thread.currentThread() + "get resource1");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + "waiting get resource2");
                synchronized (resource2) {
                    System.out.println(Thread.currentThread() + "get resource2");
                }
            }
        }, "线程 1").start();

        new Thread(() -> {
            synchronized (resource2) {
                System.out.println(Thread.currentThread() + "get resource2");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + "waiting get resource1");
                synchronized (resource1) {
                    System.out.println(Thread.currentThread() + "get resource1");
                }
            }
        }, "线程 2").start();
    }
}

Thread[线程 1,5,main]get resource1
Thread[线程 2,5,main]get resource2
Thread[线程 1,5,main]waiting get resource2
Thread[线程 2,5,main]waiting get resource1

new Thread(() -> {
            synchronized (resource1) {
                System.out.println(Thread.currentThread() + "get resource1");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread() + "waiting get resource2");
                synchronized (resource2) {
                    System.out.println(Thread.currentThread() + "get resource2");
                }
            }
        }, "线程 2").start();

Thread[线程 1,5,main]get resource1
Thread[线程 1,5,main]waiting get resource2
Thread[线程 1,5,main]get resource2
Thread[线程 2,5,main]get resource1
Thread[线程 2,5,main]waiting get resource2
Thread[线程 2,5,main]get resource2

Process finished with exit code 0

public class VisibilityProblem {
    private static boolean flag = false; // 没有volatile

    public static void main(String[] args) throws InterruptedException {
        Thread writerThread = new Thread(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            flag = true; // 写线程修改flag
            System.out.println("Writer: flag设置为true");
        });

        Thread readerThread = new Thread(() -> {
            while (!flag) {
                // 可能永远循环,因为看不到flag的修改
            }
            System.out.println("Reader: 检测到flag为true");
        });

        readerThread.start();
        writerThread.start();
    }
}

private static volatile boolean flag = false; // 添加volatile

@State(Scope.Benchmark)
public class VolatileBenchmark {
    private boolean normalFlag = false;
    private volatile boolean volatileFlag = false;

    @Benchmark
    @BenchmarkMode(Mode.AverageTime)
    public void testNormalFlag() {
        normalFlag = true;
        while (!normalFlag) {
            // 可能被优化
        }
    }

    @Benchmark
    @BenchmarkMode(Mode.AverageTime)
    public void testVolatileFlag() {
        volatileFlag = true;
        while (!volatileFlag) {
            // 保证可见性
        }
    }
}

public class VolatileAtomicityDemo {
    public volatile static int inc = 0;

    public void increase() {
        inc++;
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        VolatileAtomicityDemo volatileAtomicityDemo = new VolatileAtomicityDemo();
        for (int i = 0; i < 5; i++) {
            threadPool.execute(() -> {
                for (int j = 0; j < 500; j++) {
                    volatileAtomicityDemo.increase();
                }
            });
        }
        // 等待1.5秒，保证上面程序执行完成
        Thread.sleep(1500);
        System.out.println(inc);
        threadPool.shutdown();
    }
}

/**
  *  CAS
  * @param o         包含要修改field的对象
  * @param offset    对象中某field的偏移量
  * @param expected  期望值
  * @param update    更新值
  * @return          true | false
  */
public final native boolean compareAndSwapObject(Object o, long offset,  Object expected, Object update);

public final native boolean compareAndSwapInt(Object o, long offset, int expected,int update);

public final native boolean compareAndSwapLong(Object o, long offset, long expected, long update);

// 获取 Unsafe 实例
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long valueOffset;

static {
    try {
        // 获取“value”字段在AtomicInteger类中的内存偏移量
        valueOffset = unsafe.objectFieldOffset
            (AtomicInteger.class.getDeclaredField("value"));
    } catch (Exception ex) { throw new Error(ex); }
}
// 确保“value”字段的可见性
private volatile int value;

// 如果当前值等于预期值，则原子地将值设置为newValue
// 使用 Unsafe#compareAndSwapInt 方法进行CAS操作
public final boolean compareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

// **原子地**将当前值加 delta 并返回旧值
public final int getAndAdd(int delta) {
    return unsafe.getAndAddInt(this, valueOffset, delta);
}

// **原子地**将当前值加 1 并返回加之前的值（旧值）
// 使用 Unsafe#getAndAddInt 方法进行CAS操作。
public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}

// **原子地**将当前值减 1 并返回减之前的值（旧值）
public final int getAndDecrement() {
    return unsafe.getAndAddInt(this, valueOffset, -1);
}

// 原子地获取并增加整数值
public final int getAndAddInt(Object o, long offset, int delta) {
    int v;
    do {
        // 以 volatile 方式获取对象 o 在内存偏移量 offset 处的整数值
        v = getIntVolatile(o, offset);
    } while (!compareAndSwapInt(o, offset, v, v + delta));
    // 返回旧值
    return v;
}

synchronized void method() {
    //业务代码
}

synchronized static void method() {
    //业务代码
}

synchronized(this) {
    //业务代码
}

public class SynchronizedDemo {
    public void method() {
        synchronized (this) {
            System.out.println("synchronized 代码块");
        }
    }
}

public class SynchronizedDemo2 {
    public synchronized void method() {
        System.out.println("synchronized 方法");
    }
}

public class ReentrantLock implements Lock, java.io.Serializable {}

// 传入一个 boolean 值，true 时为公平锁，false 时为非公平锁
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

class Singleton {
    private static volatile Singleton instance; // 必须 volatile

    public static Singleton getInstance() {
        if (instance == null) {
            synchronized (Singleton.class) {
                if (instance == null) {
                    instance = new Singleton();
                }
            }
        }
        return instance;
    }
}

class ImmutableObject {
    private final int x;
    public ImmutableObject(int x) { this.x = x; }
}

AtomicInteger counter = new AtomicInteger(0);
counter.incrementAndGet();                       // 原子自增
counter.updateAndGet(x -> x * 2);                // 原子函数式更新
boolean ok = counter.compareAndSet(2, 5);        // CAS

private void replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot) {
    Entry[] tab = table;
    int len = tab.length, slotToExpunge = staleSlot;
    for (int i = prevIndex(staleSlot, len); (e = tab[i]) \!= null; i = prevIndex(i, len))
        if (e.get() == null) slotToExpunge = i;
    for (int i = nextIndex(staleSlot, len); (e = tab[i]) \!= null; i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();
        if (k == key) { e.value = value; tab[i] = tab[staleSlot]; tab[staleSlot] = e;
            if (slotToExpunge == staleSlot) slotToExpunge = i;
            cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); return; }
        if (k == null && slotToExpunge == staleSlot) slotToExpunge = i;
    }
    tab[staleSlot].value = null; tab[staleSlot] = new Entry(key, value);
    if (slotToExpunge \!= staleSlot) cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
}

private int expungeStaleEntry(int staleSlot) {
    Entry[] tab = table; int len = tab.length;
    tab[staleSlot].value = null; tab[staleSlot] = null; size--;
    Entry e; int i;
    for (i = nextIndex(staleSlot, len); (e = tab[i]) \!= null; i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();
        if (k == null) { e.value = null; tab[i] = null; size--; }
        else { int h = k.threadLocalHashCode & (len - 1);
            if (h \!= i) { tab[i] = null; while (tab[h] \!= null) h = nextIndex(h, len); tab[h] = e; } }
    }
    return i;
}

public class InheritableThreadLocalExample {
    private static InheritableThreadLocal<String> inheritableThreadLocal = new InheritableThreadLocal<>();

    public static void main(String[] args) {
        // 在主线程中设置 InheritableThreadLocal 的值
        inheritableThreadLocal.set("Hello from parent thread");

        // 创建子线程
        Thread childThread = new Thread(() -> {
            // 在子线程中获取 InheritableThreadLocal 的值
            String value = inheritableThreadLocal.get();
            System.out.println("Child thread value: " + value);
        });

        // 启动子线程
        childThread.start();

        // 等待子线程执行完毕
        try {
            childThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

import com.alibaba.ttl.TransmittableThreadLocal;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TransmittableThreadLocalExample {
    private static TransmittableThreadLocal<String> transmittableThreadLocal = new TransmittableThreadLocal<>();

    public static void main(String[] args) {
        // 在主线程中设置 TransmittableThreadLocal 的值
        transmittableThreadLocal.set("Hello from parent thread");

        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(2);

        // 提交任务到线程池
        executorService.submit(() -> {
            // 在子线程中获取 TransmittableThreadLocal 的值
            String value = transmittableThreadLocal.get();
            System.out.println("Child thread value: " + value);
        });

        // 关闭线程池
        executorService.shutdown();
    }
}

public ThreadPoolExecutor(int corePoolSize,//线程池的核心线程数量
                              int maximumPoolSize,//线程池的最大线程数
                              long keepAliveTime,//当线程数大于核心线程数时，多余的空闲线程存活的最长时间
                              TimeUnit unit,//时间单位
                              BlockingQueue<Runnable> workQueue,//任务队列，用来储存等待执行任务的队列
                              ThreadFactory threadFactory,//线程工厂，用来创建线程，一般默认即可
                              RejectedExecutionHandler handler//拒绝策略，当提交的任务过多而不能及时处理时，我们可以定制策略来处理任务

ThreadFactory threadFactory = new ThreadFactoryBuilder()
                        .setNameFormat(threadNamePrefix + "-%d")
                        .setDaemon(true).build();
ExecutorService threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程工厂，它设置线程名称，有利于我们定位问题。
 */
public final class NamingThreadFactory implements ThreadFactory {

    private final AtomicInteger threadNum = new AtomicInteger();
    private final String name;

    /**
     * 创建一个带名字的线程池生产工厂
     */
    public NamingThreadFactory(String name) {
        this.name = name;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r);
        t.setName(name + " [#" + threadNum.incrementAndGet() + "]");
        return t;
    }
}

import java.util.concurrent.*;

public class DynamicThreadPoolExample {
    public static void main(String[] args) throws InterruptedException {
        // 初始参数
        int corePoolSize = 2;
        int maximumPoolSize = 4;
        long keepAliveTime = 10;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(2);
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();

        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                workQueue,
                threadFactory,
                handler
        );

        // 提交初始任务
        for (int i = 1; i <= 6; i++) {
            final int taskId = i;
            try {
                executor.submit(() -> {
                    System.out.println("Task " + taskId + " is running on thread " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(2000); // 模拟任务执行时间
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            } catch (RejectedExecutionException e) {
                System.out.println("Task " + taskId + " was rejected");
            }
        }

        // 动态调整线程池参数
        Thread.sleep(3000); // 等待一段时间
        System.out.println("Adjusting thread pool parameters...");

        // 增加核心线程数和最大线程数
        executor.setCorePoolSize(4);
        executor.setMaximumPoolSize(6);

        // 提交更多任务
        for (int i = 7; i <= 12; i++) {
            final int taskId = i;
            try {
                executor.submit(() -> {
                    System.out.println("Task " + taskId + " is running on thread " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(2000); // 模拟任务执行时间
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            } catch (RejectedExecutionException e) {
                System.out.println("Task " + taskId + " was rejected");
            }
        }

        // 关闭线程池
        executor.shutdown();
    }
}

ThreadPoolExecutor executor = new ThreadPoolExecutor(...);

// 定期监控线程池状态
ScheduledExecutorService monitor = Executors.newScheduledThreadPool(1);
monitor.scheduleAtFixedRate(() -> {
    System.out.println("=== 线程池监控 ===");
    System.out.println("核心线程数: " + executor.getCorePoolSize());
    System.out.println("最大线程数: " + executor.getMaximumPoolSize());
    System.out.println("当前线程数: " + executor.getPoolSize());
    System.out.println("活跃线程数: " + executor.getActiveCount());
    System.out.println("队列大小: " + executor.getQueue().size());
    System.out.println("已完成任务数: " + executor.getCompletedTaskCount());
    System.out.println("总任务数: " + executor.getTaskCount());
    System.out.println("==================");
}, 0, 5, TimeUnit.SECONDS);

import java.lang.management.ManagementFactory;
import javax.management.*;

public class ThreadPoolMonitor implements ThreadPoolMonitorMBean {
    private final ThreadPoolExecutor executor;

    public ThreadPoolMonitor(ThreadPoolExecutor executor) {
        this.executor = executor;
        // 注册MBean
        try {
            MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
            ObjectName name = new ObjectName("com.example:type=ThreadPool");
            mbs.registerMBean(this, name);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public int getActiveCount() {
        return executor.getActiveCount();
    }

    @Override
    public int getQueueSize() {
        return executor.getQueue().size();
    }

    @Override
    public long getCompletedTaskCount() {
        return executor.getCompletedTaskCount();
    }
}

// 使用JConsole或VisualVM连接查看

import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;

ThreadPoolExecutor executor = new ThreadPoolExecutor(...);

// 绑定到Micrometer
ExecutorServiceMetrics.monitor(meterRegistry, executor, "my-thread-pool");

// 可以导出到Prometheus、Grafana等监控系统

Exception in thread "main" java.util.concurrent.RejectedExecutionException
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution

// 1. 检查线程池配置
System.out.println("核心线程数: " + executor.getCorePoolSize());
System.out.println("最大线程数: " + executor.getMaximumPoolSize());
System.out.println("队列容量: " + executor.getQueue().remainingCapacity());

// 2. 检查当前状态
System.out.println("当前线程数: " + executor.getPoolSize());
System.out.println("活跃线程数: " + executor.getActiveCount());
System.out.println("队列大小: " + executor.getQueue().size());

// 方案1: 增加队列容量
BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(1000);

// 方案2: 增加最大线程数
executor.setMaximumPoolSize(20);

// 方案3: 使用CallerRunsPolicy(让调用者执行)
RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();

// 方案4: 自定义拒绝策略(记录日志+告警)
RejectedExecutionHandler customHandler = (r, executor) -> {
    logger.error("Task rejected: {}", r.toString());
    // 发送告警
    alertService.sendAlert("线程池任务被拒绝");
    // 可以选择持久化任务
    taskRepository.save(r);
};

# 1. 查看进程CPU使用率
top -p <pid>

# 2. 查看线程CPU使用率
top -H -p <pid>

# 3. 使用jstack查看线程栈
jstack <pid> > thread_dump.txt

# 4. 分析线程栈,找出CPU占用高的线程
# 将线程ID转换为16进制,在thread_dump.txt中搜索
printf "%x\n" <thread_id>

// 1. 合理设置核心线程数
// CPU密集型: corePoolSize = CPU核心数 + 1
int corePoolSize = Runtime.getRuntime().availableProcessors() + 1;

// IO密集型: corePoolSize = CPU核心数 * 2
int corePoolSize = Runtime.getRuntime().availableProcessors() * 2;

// 2. 设置线程空闲回收时间
executor.setKeepAliveTime(60, TimeUnit.SECONDS);
executor.allowCoreThreadTimeOut(true);

// 1. 记录任务执行时间
ThreadPoolExecutor executor = new ThreadPoolExecutor(...) {
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        startTime.set(System.currentTimeMillis());
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        try {
            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime.get();
            if (duration > 1000) {
                logger.warn("Task execution time: {}ms", duration);
            }
        } finally {
            startTime.remove();
        }
        super.afterExecute(r, t);
    }
};

// 2. 使用VisualVM或JProfiler分析
// 3. 检查是否有死锁
ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
long[] deadlockedThreads = threadMXBean.findDeadlockedThreads();
if (deadlockedThreads != null) {
    logger.error("发现死锁线程: {}", Arrays.toString(deadlockedThreads));
}

class ResizableCapacityLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> {
    private final AtomicInteger capacity; // 使用 AtomicInteger 来支持动态调整容量

    public ResizableCapacityLinkedBlockingQueue(int capacity) {
        super(capacity); // 初始化队列容量
        this.capacity = new AtomicInteger(capacity);
    }

    public synchronized void setCapacity(int newCapacity) {
        if (newCapacity < size()) {
            throw new IllegalArgumentException("New capacity cannot be less than current size");
        }
        this.capacity.set(newCapacity);
    }

    public int getCapacity() {
        return capacity.get();
    }

    @Override
    public boolean offer(E e) {
        // 检查当前队列大小是否超过容量
        if (size() >= capacity.get()) {
            return false; // 队列已满，拒绝添加
        }
        return super.offer(e);
    }

    @Override
    public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
        // 检查当前队列大小是否超过容量
        if (size() >= capacity.get()) {
            return false; // 队列已满，拒绝添加
        }
        return super.offer(e, timeout, unit);
    }

    @Override
    public void put(E e) throws InterruptedException {
        // 检查当前队列大小是否超过容量
        while (size() >= capacity.get()) {
            Thread.yield(); // 等待队列有空闲空间
        }
        super.put(e);
    }
}