Hikari连接池源码解读
public class ConcurrentBagimplements AutoCloseable {private static final Logger LOGGER = LoggerFactory.getLogger(ConcurrentBag.class);// 所有连接：通过CopyOnWriteArrayList + State + cas 来避免了上锁
private final CopyOnWriteArrayListsharedList;
// threadList是否使用弱引用
private final boolean weakThreadLocals;
// 归还的时候缓存空闲连接到 ThreadLocal：requite()、borrow()
private final ThreadLocal<1list> threadList;
private final IBagStateListener listener;
// 等待获取连接的线程数：调 borrow() 方法+1，调完-1
private final AtomicInteger waiters;
// 连接池关闭标识
private volatile boolean closed;
// 队列大小为0的阻塞队列：生产者消费者模式
private final SynchronousQueuehandoffQueue;
public interface IConcurrentBagEntry {
int STATE_NOT_IN_USE = 0; // 空闲
int STATE_IN_USE = 1; // 活跃
int STATE_REMOVED = -1; // 移除
int STATE_RESERVED = -2; // 不可用
boolean compareAndSet(int expectState, int newState);
void setState(int newState);
int getState();
}
public interface IBagStateListener {
void addBagItem(int waiting);
}
public ConcurrentBag(final IBagStateListener listener) {
this.listener = listener;
this.weakThreadLocals = useWeakThreadLocals();
this.handoffQueue = new SynchronousQueue<>(true);
this.waiters = new AtomicInteger();
this.sharedList = new CopyOnWriteArrayList<>();
if (weakThreadLocals) {
this.threadList = ThreadLocal.withInitial(() -> new ArrayList<>(16));
} else {
this.threadList = ThreadLocal.withInitial(() -> new FastList<>(IConcurrentBagEntry.class, 16));
}
}
public T borrow(long timeout, final TimeUnit timeUnit) throws InterruptedException {
// Try the thread-local list first
// 先从 threadLocal 缓存中获取
final List list = threadList.get();
for (int i = list.size() - 1; i >= 0; i--) {
// 从尾部读取：后缓存的优先用，细节！
final Object entry = list.remove(i);
@SuppressWarnings("unchecked") final T bagEntry = weakThreadLocals ? ((WeakReference) entry).get() : (T) entry;
if (bagEntry != null && bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
return bagEntry;
}
}
// Otherwise, scan the shared list ... then poll the handoff queue
// 如果本地缓存获取不到，从 shardList 连接池中获取，等待连接数+1
final int waiting = waiters.incrementAndGet();
try {
for (T bagEntry : sharedList) {
if (bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
// If we may have stolen another waiter's connection, request another bag add.
// 并发情况下，保证能够及时补充连接
if (waiting > 1) {
listener.addBagItem(waiting - 1);
}
return bagEntry;
}
}
// 如果 shardList 连接池中也没获得连接，提交添加连接的异步任务，然后再从 handoffQueue 阻塞获取。
listener.addBagItem(waiting);
timeout = timeUnit.toNanos(timeout);
do {
final long start = currentTime();
final T bagEntry = handoffQueue.poll(timeout, NANOSECONDS);
if (bagEntry == null || bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
return bagEntry;
}
timeout -= elapsedNanos(start);
} while (timeout > 10_000);
return null;
} finally {
// 等待连接数减 1
waiters.decrementAndGet();
}
}
public void requite(final T bagEntry) {
bagEntry.setState(STATE_NOT_IN_USE);
// 如果有线程正在获取链接，则优先通过 handoffQueue 阻塞队列归还给其他线程使用
for (int i = 0; waiters.get() > 0; i++) {
if (bagEntry.getState() != STATE_NOT_IN_USE || handoffQueue.offer(bagEntry)) {
return;
} else if ((i & 0xff) == 0xff) {
// 每遍历 255 个休眠 10 微妙
parkNanos(MICROSECONDS.toNanos(10));
} else {
// 线程让步
yield();
}
}
// 没有其它线程用，就放入本地缓存
final List threadLocalList = threadList.get();
threadLocalList.add(weakThreadLocals ? new WeakReference<>(bagEntry) : bagEntry);
}
public void add(final T bagEntry) {
if (closed) {
LOGGER.info("ConcurrentBag has been closed, ignoring add()");
throw new IllegalStateException("ConcurrentBag has been closed, ignoring add()");
}
sharedList.add(bagEntry);
// spin until a thread takes it or none are waiting
// 如果有线程等待获取连接，循环通过 handoffQueue 提交连接
while (waiters.get() > 0 && !handoffQueue.offer(bagEntry)) {
yield();
}
}
public boolean remove(final T bagEntry) {
// 使用 CAS 将连接置为 STATE_REMOVED 状态
if (!bagEntry.compareAndSet(STATE_IN_USE, STATE_REMOVED) && !bagEntry.compareAndSet(STATE_RESERVED, STATE_REMOVED) && !closed) {
LOGGER.warn("Attempt to remove an object from the bag that was not borrowed or reserved: {}", bagEntry);
return false;
}
// CAS 成功后再删除连接
final boolean removed = sharedList.remove(bagEntry);
if (!removed && !closed) {
LOGGER.warn("Attempt to remove an object from the bag that does not exist: {}", bagEntry);
}
return removed;
}
@Override
public void close() {
closed = true;
}
public boolean reserve(final T bagEntry) {
return bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_RESERVED);
}
}
http://blog.itpub.net/69955379/viewspace-2908264/
页:
[1]