1.6w字图解Java并发：多线程挑战、线程状态和通信、死锁；AQS、ReentrantLock、Condition 使用和原理

sudo -u admin /opt/magebyte/java/bin/jstack 31177 > /home/magebyte/dump17

grep java.lang.Thread.State dump17 | awk '{print $2$3$4$5}'
| sort | uniq -c
39 RUNNABLE
21 TIMED_WAITING(onobjectmonitor)
6 TIMED_WAITING(parking)
51 TIMED_WAITING(sleeping)
305 WAITING(onobjectmonitor)
3 WAITING(parking)

"http-0.0.0.0-7001-97" daemon prio=10 tid=0x000000004f6a8000 nid=0x555e in
Object.wait() [0x0000000052423000]
java.lang.Thread.State: WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
- waiting on <0x00000007969b2280> (a org.apache.tomcat.util.net.AprEndpoint$Worker)
at java.lang.Object.wait(Object.java:485)
at org.apache.tomcat.util.net.AprEndpoint$Worker.await(AprEndpoint.java:1464)
- locked <0x00000007969b2280> (a org.apache.tomcat.util.net.AprEndpoint$Worker)
at org.apache.tomcat.util.net.AprEndpoint$Worker.run(AprEndpoint.java:1489)
at java.lang.Thread.run(Thread.java:662)

# 最大工作线程数，默认200。
server.tomcat.max-threads=200

server:
  port:9000
tomcat:
    uri-encoding:UTF-8
    max-threads:800#最大工作线程数量
    min-spare-threads:20#最小工作线程数量
    max-connections:10000#一瞬间最大支持的并发的连接数
    accept-count:200#等待队列长度

public class DeadLockExample {
    public Object resourceA = new Object();
    public Object resourceB = new Object();
    public static void main(String[] args) {
        DeadLockExample deadLockExample = new DeadLockExample();
        Runnable runnableA = new Runnable() {
            @Override
            public void run() {
                synchronized(deadLockExample.resourceA) {
                    System.out.printf(
                        "[INFO]: %s get resourceA" + System.lineSeparator(),
                        Thread.currentThread().getName()
                    );
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.printf(
                        "[INFO]: %s trying to get resourceB" + System.lineSeparator(),
                        Thread.currentThread().getName()
                    );
                    synchronized(deadLockExample.resourceB) {
                        System.out.printf(
                            "[INFO]: %s get resourceB" + System.lineSeparator(),
                            Thread.currentThread().getName()
                        );
                    }
                    System.out.printf(
                        "[INFO]: %s has done" + System.lineSeparator(),
                        Thread.currentThread().getName()
                    );
                }
            }
        };
        Runnable runnableB = new Runnable() {
            @Override
            public void run() {
                synchronized(deadLockExample.resourceB) {
                    System.out.printf(
                        "[INFO]: %s get resourceB" + System.lineSeparator(),
                        Thread.currentThread().getName()
                    );
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.printf(
                        "[INFO]: %s trying to get resourceA" + System.lineSeparator(),
                        Thread.currentThread().getName()
                    );
                    synchronized(deadLockExample.resourceA) {
                        System.out.printf(
                            "[INFO]: %s get resourceA" + System.lineSeparator(),
                            Thread.currentThread().getName()
                        );
                    }
                    System.out.printf(
                        "[INFO]: %s has done" + System.lineSeparator(),
                        Thread.currentThread().getName()
                    );
                }
            }
        };
        new Thread(runnableA).start();
        new Thread(runnableB).start();
    }
}

[INFO]: Thread-0 get resourceA
[INFO]: Thread-1 get resourceB
[INFO]: Thread-0 trying to get resourceB
[INFO]: Thread-1 trying to get resourceA

$ jstack $(jps -l | grep 'DeadLockExample' | cut -f1 -d ' ')

...
Java stack information for the threads listed above:
===================================================
"Thread-1":
    at DeadLockExample$2.run(DeadLockExample.java:58)
    - waiting to lock <0x000000076ab660a0> (a java.lang.Object)
    - locked <0x000000076ab660b0> (a java.lang.Object)
    at java.lang.Thread.run(Thread.java:748)
"Thread-0":
    at DeadLockExample$1.run(DeadLockExample.java:28)
    - waiting to lock <0x000000076ab660b0> (a java.lang.Object)
    - locked <0x000000076ab660a0> (a java.lang.Object)
    at java.lang.Thread.run(Thread.java:748)

Found 1 deadlock.

private staticclass Runner implements Runnable{
        privatelong i;
        privatevolatileboolean running =true;
        @Override
        public void run() {
            System.out.println("current Thread Name:"+Thread.currentThread().getName());
            while (running ){
                i++;
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) {
                }
            }
            System.out.println("Count i= "+i);
            System.out.println("current Thread Name:"+Thread.currentThread().getName());

        }
        public void cancel(){
            running =false;
            System.out.println("running=false");
        }
    }

public class TestSync {
    public static void main(String[] args) {
        // 定义一个锁对象
        Object lock = new Object();
        List<String>  list = new ArrayList<>();
        // 实现线程A
        Thread threadA = new Thread(() -> {
            synchronized (lock) {
                for (int i = 1; i <= 10; i++) {
                    list.add("abc");
                    System.out.println("线程A向list中添加一个元素，此时list中的元素个数为：" + list.size());
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    if (list.size() == 5)
                        lock.notify();// 唤醒B线程
                }
            }
        });
        // 实现线程B
        Thread threadB = new Thread(() -> {
            while (true) {
                synchronized (lock) {
                    if (list.size() != 5) {
                        try {
                            lock.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    System.out.println("线程B收到通知，开始执行自己的业务...");
                }
            }
        });
        // 需要先启动线程B
        threadB.start();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 再启动线程A
        threadA.start();
    }
}

synchronized(对象) {
  while(条件不满足) {
   对象.wait();
  }

   满足条件对应的处理逻辑
}

synchronized(对象) {
  改变条件
  对象.notifyAll();
}

public class UserContext {
    // 使用静态内部类实现懒加载，保证线程安全
    privatestaticfinal ThreadLocal<UserInfo> currentUser = new ThreadLocal<>();

    public static void set(UserInfo user) {
        currentUser.set(user);
    }

    public static UserInfo get() {
        UserInfo user = currentUser.get();
        if (user == null) {
            thrownew IllegalStateException("User not found in current thread context");
        }
        return user;
    }

    public static void clear() {
        currentUser.remove(); // 必须显式清理防止内存泄漏
    }
}

@Component
publicclass AuthInterceptor implements HandlerInterceptor {
    @Autowired
    private JwtTokenService jwtTokenService; // 自定义的JWT解析服务

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        // 从请求头获取Token
        String token = request.getHeader("Authorization");
        if (token != null && token.startsWith("Bearer ")) {
            token = token.substring(7);
            UserInfo user = jwtTokenService.parseToken(token); // 解析用户信息
            UserContext.set(user); // 存储到ThreadLocal
        }
        returntrue;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response,
                               Object handler, Exception ex) {
        UserContext.clear(); // 请求结束时必须清理ThreadLocal
    }
}

@Configuration
public class WebConfig implements WebMvcConfigurer {
    @Autowired
    private AuthInterceptor authInterceptor;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(authInterceptor)
                .addPathPatterns("/api/**") // 拦截API请求
                .excludePathPatterns("/api/public/**"); // 排除公共接口
    }
}

@RestController
@RequestMapping("/api")
public class OrderController {
    @GetMapping("/orders")
    public ResponseEntity<List<Order>> getOrders() {
        UserInfo currentUser = UserContext.get(); // 无需参数传递
        return orderService.findByUserId(currentUser.getUserId());
    }
}

// 关键字在实例方法上，锁为当前实例
public synchronized void instanceLock() {
    // code
}

// 关键字在静态方法上，锁为当前Class对象
public static synchronized void classLock() {
    // code
}

// 关键字在代码块上，锁为括号里面的对象
public void blockLock() {
    Object o = new Object();
    synchronized (o) {
        // code
    }
}

Lock lock = new ReentrantLock();
lock.lock();
try {

} finally {
 lock.unlock();
}

class Mutex {

    privatestaticclass Sync extends AbstractQueuedSynchronizer {
        @Override
        protected boolean tryAcquire(int arg) {
            return compareAndSetState(0, 1);
        }

        @Override
        protected boolean tryRelease(int arg) {
            setState(0);
            returntrue;
        }

        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
    }

    privatefinal Sync sync = new Sync();

    public void lock() {
        sync.tryAcquire(1);
    }

    public void unlock() {
        sync.tryRelease(1);
    }

    public boolean isLocked() {
        return sync.isHeldExclusively();
    }

}

public final void acquire(int arg) {
 if (!tryAcquire(arg) &&
   acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 这里 Node.EXCLUSIVE 的值是 null
   selfInterrupt();
}

public final boolean release(int arg) {
if (tryRelease(arg)) { // 尝试释放锁，如果成功则唤醒后继节点的线程
  Node h = head;
if (h != null && h.waitStatus != 0)
      //唤醒CHL队列第二个线程节点
   unparkSuccessor(h);
returntrue;
 }
returnfalse;
}

private final Lock lock = new ReentrantLock();
public void foo() {
    // 获取锁
    lock.lock();
    try{
        // 程序执行逻辑
    } finally{
        // finally语句块可以确保lock被正确释放
        lock.unlock();
    }
}

abstract staticclass Sync extends AbstractQueuedSynchronizer {
        privatestaticfinallong serialVersionUID = -5179523762034025860L;

        /**
         * 非公平锁获取资源
         */
        @ReservedStackAccess
        final boolean nonfairTryAcquire(int acquires) {
            // 获取当前线程
            final Thread current = Thread.currentThread();
            //获取当前状态
            int c = getState();
            // state==0 代表资源可获取
            if (c == 0) {
              //cas设置state为acquires，acquires传入的是1
                if (compareAndSetState(0, acquires)) {
                   // cas成功，将当前线程设置持有锁的线程
                    setExclusiveOwnerThread(current);
                  // 返回成功
                    returntrue;
                }
            }
          //如果state!=0,但是当前线程是持有锁线程，直接重入
            elseif (current == getExclusiveOwnerThread()) {
               //state状态+1
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    thrownew Error("Maximum lock count exceeded");
              //设置state状态，此处不需要cas，因为持有锁的线程只有一个
                setState(nextc);
                returntrue;
            }
          // 获取锁失败
            returnfalse;
        }

     /**
     * 释放资源
     */
        @ReservedStackAccess
        protected final boolean tryRelease(int releases) {
          //state状态-releases，releases传入的是1
            int c = getState() - releases;
           //如果当前线程不是持有锁线程，抛出异常
            if (Thread.currentThread() != getExclusiveOwnerThread())
                thrownew IllegalMonitorStateException();
            boolean free = false;
          //state-1后，如果c==0代表释放资源成功
            if (c == 0) {
              //返回状态设置为true
                free = true;
              //清空持有锁线程
                setExclusiveOwnerThread(null);
            }
          //如果state-1后，state还是>0，
          //代表当前线程有锁重入操作，需要做相应的释放次数，设置state值
            setState(c);
            return free;
        }

        ....省略其他代码
    }

 static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;
        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
 }

static finalclass FairSync extends Sync {
    privatestaticfinallong serialVersionUID = -3000897897090466540L;
    /**
     * 公平策略获取锁
     */
    @ReservedStackAccess
    protected final boolean tryAcquire(int acquires) {
        //获取当前线程
        final Thread current = Thread.currentThread();
        //获取state状态
        int c = getState();
      // state==0 代表资源可获取
        if (c == 0) {
          //1.hasQueuedPredecessors判断当前线程是不是CLH队列被唤醒的线程，如果是执行下一个步骤
          //2.cas设置state为acquires，acquires传入的是1
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
               //cas成功，设置当前持有锁的线程
                setExclusiveOwnerThread(current);
              //返回成功
                returntrue;
            }
        }
      //如果state!=0,但是当前线程是持有锁线程，直接重入
        elseif (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                thrownew Error("Maximum lock count exceeded");
            setState(nextc);
            returntrue;
        }
        returnfalse;
    }
}

public interface ReadWriteLock {
    /**
     * 获取读锁
     */
    Lock readLock();

    /**
     * 获取写锁
     */
    Lock writeLock();
}

public class RWDictionary {
    // 共享数据
    privatefinal Map<String, Data> m = new TreeMap<>();
    // 定义读写锁
    privatefinal ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    // 定义读锁
    privatefinal Lock r = rwl.readLock();
    // 定义写锁
    privatefinal Lock w = rwl.writeLock();

    public Data get(String key) {
        r.lock();
        try {
            return m.get(key);
        } finally {
            r.unlock();
        }
    }

    public List<String> allKeys() {
        r.lock();
        try {
            returnnew ArrayList<>(m.keySet());
        } finally {
            r.unlock();
        }
    }

    public Data put(String key, Data value) {
        w.lock();
        try {
            return m.put(key, value);
        } finally {
            w.unlock();
        }
    }

    public void clear() {
        w.lock();
        try {
            m.clear();
        } finally {
            w.unlock();
        }
    }
}

public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    privatestaticfinallong serialVersionUID = -6992448646407690164L;
    /** 内部类，读锁 */
    privatefinal ReentrantReadWriteLock.ReadLock readerLock;
    /** 内部类，写锁 */
    privatefinal ReentrantReadWriteLock.WriteLock writerLock;
    /** 内部类，继承 AQS */
    final Sync sync;

    /**
     * 默认非公平策略获取读写锁
     */
    public ReentrantReadWriteLock() {
        this(false);
    }

    /**
     * 公平策略获取读写锁
     */
    public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }
  ...省略部分代码
}

 protected final boolean tryAcquire(int acquires) {
      // 当前线程
      Thread current = Thread.currentThread();
      int c = getState();
   // 计算写锁数量
      int w = exclusiveCount(c);
      if (c != 0) {
          /// 存在读锁或者当前获取线程不是已经获取写锁的线程
          if (w == 0 || current != getExclusiveOwnerThread())
              returnfalse;
        // 超过最大范围
          if (w + exclusiveCount(acquires) > MAX_COUNT)
              thrownew Error("Maximum lock count exceeded");
          // Reentrant acquire
          setState(c + acquires);
          returntrue;
      }
   // 写锁是否阻塞或者状态设置失败，返回 false
      if (writerShouldBlock() ||
          !compareAndSetState(c, c + acquires))
          returnfalse;
      setExclusiveOwnerThread(current);
      returntrue;
  }

@ReservedStackAccess
protected final boolean tryRelease(int releases) {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    int nextc = getState() - releases;
    boolean free = exclusiveCount(nextc) == 0;
    if (free)
        setExclusiveOwnerThread(null);
    setState(nextc);
    return free;
}

class BoundedBuffer {

    final Lock lock = new ReentrantLock();
    // condition 实例依赖于 lock 实例
    final Condition notFull = lock.newCondition();
    final Condition notEmpty = lock.newCondition();

    final Object[] items = new Object[100];

    int putPtr, takePtr, count;

    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            //  put 时判断是否已经满了
            // 则线程在 notFull 条件上排队阻塞
            while (count == items.length) {
                notFull.await();
            }
            items[putPtr] = x;
            if (++putPtr == items.length) {
                putPtr = 0;
            }
            ++count;
            // put 成功之后，队列中有元素
            // 唤醒在 notEmpty 条件上排队阻塞的线程
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            // take 时，发现为空
            // 则线程在 notEmpty 的条件上排队阻塞
            while (count == 0) {
                notEmpty.await();
            }
            Object x = items[takePtr];
            if (++takePtr == items.length) {
                takePtr = 0;
            }
            --count;
            // take 成功，队列不可能是满的
            // 唤醒在 notFull 条件上排队阻塞的线程
            notFull.signal();
            return x;
        } finally {
            lock.unlock();
        }
    }
}