/**
  * 线程池使用一个int变量存储线程池状态和工作线程数
  * int4个字节，32位，用高三位存储线程池状态，低29位存储工作线程数
  * 为什么使用一个变量来同时表示线程状态和线程数？就是节省空间。咨询了一下写c的朋友，他们经常这么写
  **/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//COUNT_BITS=29
private static final int COUNT_BITS = Integer.SIZE - 3;
//理论上线程池最大线程数量CAPACITY=(2^29)-1，即 536,870,911
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

//获取线程池状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//获取工作线程数
private static int workerCountOf(int c)  { return c & CAPACITY; }
//初始化ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

/**
  * 线程池状态转换
  * RUNNING -> SHUTDOWN
  * RUNNING or SHUTDOWN -> STOP
  * SHUTDOWN or STOP -> TIDYING
  * TIDYING -> TERMINATED  terminated()执行完后变为该TERMINATED
  */
//接受新任务，可以处理阻塞队列里的任务
private static final int RUNNING    = -1 << COUNT_BITS;
//不接受新任务，可以处理阻塞队列里的任务。执行shutdown()会变为SHUTDOWN
private static final int SHUTDOWN   =  0 << COUNT_BITS;
//不接受新的任务，不处理阻塞队列里的任务，中断正在处理的任务。执行shutdownNow()会变为STOP
private static final int STOP       =  1 << COUNT_BITS;
//临时过渡状态，所有的任务都执行完了，当前线程池有效的线程数量为0，这个时候线程池的状态是TIDYING，执行terminated()变为TERMINATED
private static final int TIDYING    =  2 << COUNT_BITS;
//终止状态，terminated()调用完成后的状态
private static final int TERMINATED =  3 << COUNT_BITS;

//重入锁，更新线程池核心大小、线程池最大大小等都有用到
private final ReentrantLock mainLock = new ReentrantLock();
//用于存储woker
private final HashSet<Worker> workers = new HashSet<Worker>();
//用于终止线程池
private final Condition termination = mainLock.newCondition();
//记录线程池中曾经出现过的最大线程数
private int largestPoolSize;
//完成任务数量
private long completedTaskCount;   

/**
 * 核心线程数
 * 核心线程会一直存活，即使没有任务需要处理，当线程数小于核心线程数时。
 * 即使现有的线程空闲，线程池也会优先创建新线程来处理任务，而不是直接交给现有的线程处理。
 * 核心线程数在初始化时不会创建，只有提交任务的时候才会创建。核心线程在allowCoreThreadTimeout为true的时候超时会退出。
 */
private volatile int corePoolSize;
 /** 最大线程数
   * 当线程数大于或者等于核心线程，且任务队列已满时，线程池会创建新的线程，直到线程数量达到maxPoolSize。
   * 如果线程数已等于maxPoolSize，且任务队列已满，则已超出线程池的处理能力，线程池会采取拒绝操作。
   */
private volatile int maximumPoolSize;
/**
  * 线程空闲时间
  * 当线程空闲时间达到keepAliveTime，该线程会退出，直到线程数量等于corePoolSize。
  * 如果allowCoreThreadTimeout设置为true，则所有线程均会退出。
  */
private volatile long keepAliveTime;
//是否允许核心线程空闲超时退出，默认值为false。
private volatile boolean allowCoreThreadTimeOut;
//线程工厂
private volatile ThreadFactory threadFactory;
//用于保存等待执行的任务的阻塞队列。比如LinkedBlockQueue，SynchronousQueue等
private final BlockingQueue<Runnable> workQueue;
/**
 *  rejectedExecutionHandler：任务拒绝策略
 *  DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务
 *  AbortPolicy：抛出异常。这也是默认的策略
 *  CallerRunsPolicy：用调用者所在线程来运行任务
 *  DiscardPolicy：不处理，丢弃掉
 */
private volatile RejectedExecutionHandler handler;
//默认的拒绝策略：抛出异常
private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();
private static final RuntimePermission shutdownPerm =
    new RuntimePermission("modifyThread");

// 初始化ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

// 或运算符(|)规则：1|1=1
//                1|0=1
//                0|1=1
//                0|0=0
// 以初始化参数 ctlOf(RUNNING, 0) 为例：
  11100000000000000000000000000000
| 00000000000000000000000000000000
= 11100000000000000000000000000000

// 获取线程池状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }

// 取反运算符(~)规则：~1=0
//                   ~0=1
// 以 c = 111 11111111111111111111111111111（RUNNING full）为例：
     ~11111111111111111111111111111
=     00000000000000000000000000000
& 111 11111111111111111111111111111
= 111

// 获取工作线程数
private static int workerCountOf(int c)  { return c & CAPACITY; }

// 与运算符(&)规则：1&1=1
//                 1&0=0
//                 0&1=0
//                 0&0=0
// 以 c = 111 11111111111111111111111111111（RUNNING full）为例：
  111 11111111111111111111111111111
&     11111111111111111111111111111
=     11111111111111111111111111111

ThreadFactory threadFactory = runnable -> {
    Thread t = new Thread(runnable);
    t.setName("wechat-notify-1");
    t.setDaemon(false);
    t.setPriority(1);
    t.setUncaughtExceptionHandler((thread, exception) -> {});
    return t;
};

ThreadFactory threadFactory = Executors.defaultThreadFactory();

ThreadFactory threadFactory = new ThreadFactoryBuilder()
  .setNameFormat("wechat-notify-%d")
  .setDaemon(false)
  .setPriority(1)
  .setUncaughtExceptionHandler((thread, exception) -> {})
  .build();

private static ThreadFactory build(ThreadFactoryBuilder builder) {
  final String nameFormat = builder.nameFormat;
  final Boolean daemon = builder.daemon;
  final Integer priority = builder.priority;
  final UncaughtExceptionHandler uncaughtExceptionHandler = builder.uncaughtExceptionHandler;
  final ThreadFactory backingThreadFactory = (builder.backingThreadFactory != null)
          ? builder.backingThreadFactory
          : Executors.defaultThreadFactory();
  final AtomicLong count = (nameFormat != null) ? new AtomicLong(0) : null;

  return new ThreadFactory() {
    @Override
    public Thread newThread(Runnable runnable) {
      Thread thread = backingThreadFactory.newThread(runnable);
      if (nameFormat != null) {
        thread.setName(format(nameFormat, count.getAndIncrement()));
      }
      if (daemon != null) {
        thread.setDaemon(daemon);
      }
      if (priority != null) {
        thread.setPriority(priority);
      }
      if (uncaughtExceptionHandler != null) {
        thread.setUncaughtExceptionHandler(uncaughtExceptionHandler);
      }
      return thread;
    }
  };
}

public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    // 参数校验
    if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();

    this.acc = System.getSecurityManager() == null ? null : AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

final void runWorker(Worker w) {

  ...

  beforeExecute(wt, task);
  Throwable thrown = null;
  try {
    task.run();
  } catch (RuntimeException x) {
    thrown = x; throw x;
  } catch (Error x) {
    thrown = x; throw x;
  } catch (Throwable x) {
    thrown = x; throw new Error(x);
  } finally {
    afterExecute(task, thrown);
  }

  ...

}

public class MdcAwareThreadPoolExecutor extends ThreadPoolExecutor {

    public MdcAwareThreadPoolExecutor(int corePoolSize, 
      int maximumPoolSize, 
      long keepAliveTime, 
      TimeUnit unit, 
      BlockingQueue<Runnable> workQueue, 
      ThreadFactory threadFactory, 
      RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    /**
     * Use ThreadPoolExecutor hooks and perform necessary cleanups after each execution.
     */
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        System.out.println("Cleaning the MDC context");
        MDC.clear();
        org.apache.log4j.MDC.clear();
        ThreadContext.clearAll();
    }
}

logger.info("{}, taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]",
                threadPoolExecutor.getThreadNamePrefix(),
                threadPoolExecutor.getTaskCount(),
                threadPoolExecutor.getCompletedTaskCount(),
                threadPoolExecutor.getActiveCount(),
                threadPoolExecutor.getQueue().size()
);

// 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
public static ExecutorService newFixedThreadPool(...) {...}
// 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序（FIFO, LIFO, 优先级）执行。
public static ExecutorService newSingleThreadExecutor(...) {...}
// 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
public static ExecutorService newCachedThreadPool(...) {...}
// 创建一个定长线程池，支持定时及周期性任务执行。
public static ScheduledExecutorService newScheduledThreadPool(...) {...}

// Fixed 限定 corePoolSize 和 maximumPoolSize 为相同大小，即线程池大小固定（意味着无法扩展）
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

// Single 其实就是 Fixed 为 1 的变种
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

// ⚠️ 允许的请求队列长度（capacity）为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。
public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

// ⚠️ 允许的创建线程数量（maximumPoolSize）为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

// 问题在于 ScheduledThreadPoolExecutor 构造方法的默认参数
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

// ⚠️ 允许的创建线程数量（maximumPoolSize）为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

try (ChannelContext ctx = new ChannelContext(channel)) {
    ...
}

@Slf4j
public class ChannelContext implements AutoCloseable {

    private static final ThreadLocal<Channel> CTX = new ThreadLocal<>();

    public ChannelContext(FundChannelDTO dto) {
        Channel channel = Channel.builder()
                .appId(dto.getAppId().toString())
                .build();
        CTX.set(channel);
    }

    public ChannelContext(Channel channel) {
        CTX.set(channel);
    }

    public static Channel get() {
        return CTX.get();
    }

    @Override
    public void close() {
        try {
            CTX.remove();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    @Getter
    @Builder
    static class Channel {
        private final String appId;
    }

}

@Override
protected ExecutorService initializeExecutor(
        ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

    BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

    ThreadPoolExecutor executor;
    if (this.taskDecorator != null) {
        executor = new ThreadPoolExecutor(
                this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
                queue, threadFactory, rejectedExecutionHandler) {
            @Override
            public void execute(Runnable command) {
                super.execute(taskDecorator.decorate(command));
            }
        };
    }
    else {
        executor = new ThreadPoolExecutor(
                this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
                queue, threadFactory, rejectedExecutionHandler);

    }

    if (this.allowCoreThreadTimeOut) {
        executor.allowCoreThreadTimeOut(true);
    }

    this.threadPoolExecutor = executor;
    return executor;
}

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.HashSet;
import java.util.Set;

import static org.junit.Assert.assertEquals;

@Slf4j
public class IdGeneratorTest {

    @Test
    public void test() {
        int initialCapacity = 1000;
        Set<Long> ids = new HashSet<>(initialCapacity);
        IdGenerator idGenerator = new IdGenerator(3, 31);
        
        for (int i = 0; i < initialCapacity; i++) {
            long id = idGenerator.nextId();
            ids.add(id);

            IdGenerator.log(id);
        }
        assertEquals(ids.size(), initialCapacity);
    }

}

@Slf4j
class IdGenerator {

    // 时间戳
    private long timestamp;
    // 数据中心
    private int dataCenterNo;
    // 机器
    private int workerNo;
    // 序列号
    private int seqNo;
    // 扩展位
    private int ext;

    // 数据中心 取值范围（十进制）：0~3
    private static final int DATA_CENTER_NO_BITS = 2;
    // 机器 取值范围（十进制）：0~31
    private static final int WORKER_NO_BITS = 5;
    // 序列号 取值范围（十进制）：0~4095
    private static final int SEQ_NO_BITS = 12;
    // 扩展位 取值范围（十进制）：0~7
    private static final int EXT_BITS = 3;

    // 数据中心 最大值（十进制）：3
    private static final int MAX_DATA_CENTER_NO = (1 << DATA_CENTER_NO_BITS) - 0B1;
    // 机器 最大值（十进制）：31
    private static final int MAX_WORKER_NO = (1 << WORKER_NO_BITS) - 0B1;
    // 序列号 最大值（十进制）：4095
    private static final int MAX_SEQ_NO = (1 << SEQ_NO_BITS) - 0B1;
    // 扩展位 最大值（十进制）：7
    private static final int MAX_EXT = (1 << EXT_BITS) - 0B1;
  
    // 时间戳 左移 22 位
    private static final int TIMESTAMP_SHIFT = DATA_CENTER_NO_BITS + WORKER_NO_BITS + SEQ_NO_BITS + EXT_BITS;
    // 数据中心 左移 20 位
    private static final int DATA_CENTER_NO_SHIFT = WORKER_NO_BITS + SEQ_NO_BITS + EXT_BITS;
    // 机器 左移 15 位
    private static final int WORKER_NO_SHIFT = SEQ_NO_BITS + EXT_BITS;
    // 序列号 左移 3 位
    private static final int SEQ_NO_SHIFT = EXT_BITS;

    // 最大回拨毫秒数
    private static final int MAX_BACKWARD_MILLIS = 5;

    /**
     *
     * @param dataCenterNo 数据中心编号
     * @param workerNo 机器编号
     */
    public IdGenerator(int dataCenterNo, int workerNo) {
        if (Integer.toBinaryString(dataCenterNo).length() > DATA_CENTER_NO_BITS) {
            throw new IllegalArgumentException(String.format("当前数据中心编号 %s 超限，最大支持 %s", dataCenterNo, MAX_DATA_CENTER_NO));
        } else if (Integer.toBinaryString(workerNo).length() > WORKER_NO_BITS) {
            throw new IllegalArgumentException(String.format("当前机器编号 %s 超限，最大支持 %s", workerNo, MAX_WORKER_NO));
        }
        this.dataCenterNo = dataCenterNo;
        this.workerNo = workerNo;
    }

    /**
     * 生成分布式 ID
     * @return 分布式 ID
     */
    public synchronized long nextId() {
        long now = System.currentTimeMillis();
        // init or reset
        if (timestamp == 0 || timestamp < now) {
            timestamp = now;
            seqNo = 0;
        }
        // seqNo increment
        else if (timestamp == now) {
            if (seqNo < MAX_SEQ_NO) {
                seqNo++;
            } else {
                log.warn("序列号已耗尽，等待重新生成。seqNo = {}, MAX_SEQ_NO = {}", seqNo, MAX_SEQ_NO);
                return sleepAndNextId(1);
            }
        }
        // clock backward
        else {
            return nextIdForBackward(now);
        }

        return timestamp << TIMESTAMP_SHIFT
                | dataCenterNo << DATA_CENTER_NO_SHIFT
                | workerNo << WORKER_NO_SHIFT
                | seqNo << SEQ_NO_SHIFT
                | ext;
    }

    private long nextIdForBackward(long now) {
        log.warn("发生时间回拨，timestamp = {}, now = {}", timestamp, now);

        long duration = timestamp - now;
        // 回拨不多，直接等待并重试
        if (duration <= MAX_BACKWARD_MILLIS) {
            return sleepAndNextId(duration);
        }
        // 回拨过多，则使用扩展位
        else {
            if (ext < MAX_EXT) {
                // 将时间戳修正为回拨时间，为防止重复生成 ID，扩展位加一，并重试
                ext++;
                timestamp = now;
                return nextId();
            } else {
                throw new IllegalStateException(String.format("扩展位已耗尽。ext = %s, MAX_EXT = %s", ext, MAX_EXT));
            }
        }
    }

    @SneakyThrows
    private long sleepAndNextId(long millis) {
        Thread.sleep(millis);
        return nextId();
    }

    public static void log(long id) {
        long timestamp = id >> TIMESTAMP_SHIFT;
        long dataCenterNo = id >> DATA_CENTER_NO_SHIFT & MAX_DATA_CENTER_NO;
        long workerNo = id >> WORKER_NO_SHIFT & MAX_WORKER_NO;
        long seqNo = id >> SEQ_NO_SHIFT & MAX_SEQ_NO;
        long ext = id & MAX_EXT;

        log.info("Binary is {}, id is {}", Long.toBinaryString(id), id);
        log.info("Binary is {}, time is {}", Long.toBinaryString(timestamp), getLocalDateTime(timestamp));
        log.info("Binary is {}, dataCenterNo is {}", Long.toBinaryString(dataCenterNo), dataCenterNo);
        log.info("Binary is {}, workerNo is {}", Long.toBinaryString(workerNo), workerNo);
        log.info("Binary is {}, seqNo is {}", Long.toBinaryString(seqNo), seqNo);
        log.info("Binary is {}, ext is {}", Long.toBinaryString(ext), ext);
    }

    private static LocalDateTime getLocalDateTime(long timestamp) {
        return Instant.ofEpochMilli(timestamp).atZone(ZoneId.systemDefault()).toLocalDateTime();
    }

}

static final int getOffset(Object key, int length) 
    int hashcode;
    int hash = (hashcode = key.hashCode()) ^ (hashcode >>> 16);
    int offset = hash & (length - 1);
    log.info("key: {}, hashcode: {}, hash: {}, offset: {}", key, Integer.toBinaryString(hashcode), Integer.toBinaryString(hash), offset);
    return offset;
}

// 0
long a = 0;
// 2^31-1, 2147483647
long b = Integer.MAX_VALUE;

// 1970-01-01T00:00:00.000Z
Instant.ofEpochSecond(a).atZone(ZoneOffset.of("-00:00")).toLocalDateTime()
// 2038-01-19T03:14:07.000Z
Instant.ofEpochSecond(b).atZone(ZoneOffset.of("-00:00")).toLocalDateTime()

// 秒时间戳
Instant.now().getEpochSecond()
// 毫秒时间戳
Instant.now().toEpochMilli()

// 毫秒时间戳
System.currentTimeMillis()