帶你快速搞定java并發(fā)庫(kù)

更新時(shí)間：2021年07月15日 10:11:06 作者：香菜聊游戲

本文主要介紹了java高并發(fā)寫(xiě)入用戶信息到數(shù)據(jù)庫(kù)的幾種方法，具有很好的參考價(jià)值。下面跟著小編一起來(lái)看下吧，希望能給你帶來(lái)幫助

一、總覽

計(jì)算機(jī)程序 = 數(shù)據(jù) + 算法。

并發(fā)編程的一切根本原因是為了保證數(shù)據(jù)的正確性，線程的效率性。

Java并發(fā)庫(kù)共分為四個(gè)大的部分，如下圖

Executor 和 future 是為了保證線程的效率性

Lock 和數(shù)據(jù)結(jié)構(gòu) 是為了維持?jǐn)?shù)據(jù)的一致性。

Java并發(fā)編程的時(shí)候，思考順序?yàn)椋?/p>

對(duì)自己的數(shù)據(jù)要么加鎖。要么使用提供的數(shù)據(jù)結(jié)構(gòu)，保證數(shù)據(jù)的安全性

調(diào)度線程的時(shí)候使用Executor更好的調(diào)度。

二、Executor總覽

Executor 提供一種將任務(wù)提交與每個(gè)任務(wù)將如何運(yùn)行的機(jī)制（包括線程使用的細(xì)節(jié)、調(diào)度等）分離開(kāi)來(lái)的方法。

相當(dāng)于manager,老板讓manager去執(zhí)行一件任務(wù)，具體的是誰(shuí)執(zhí)行，什么時(shí)候執(zhí)行，就不管了。

看上圖的繼承關(guān)系，介紹幾個(gè)

內(nèi)置的線程池基本上都在這里

newScheduledThreadPool 定時(shí)執(zhí)行的線程池

newCachedThreadPool 緩存使用過(guò)的線程

newFixedThreadPool 固定數(shù)量的線程池

newWorkStealingPool 將大任務(wù)分解為小任務(wù)的線程池

三、繼承結(jié)構(gòu)

構(gòu)造函數(shù)

包含一個(gè)定時(shí)的service

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1));
}
static class DelegatedScheduledExecutorService
        extends DelegatedExecutorService
        implements ScheduledExecutorService {
    private final ScheduledExecutorService e;
    DelegatedScheduledExecutorService(ScheduledExecutorService executor) {
        super(executor);
        e = executor;
    }

四、怎么保證只有一個(gè)線程

定時(shí)執(zhí)行的時(shí)候調(diào)用這個(gè)方法，調(diào)用過(guò)程如下，注意看其中的注釋，由上往下的調(diào)用順序

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                 long initialDelay,
                                                 long delay,
                                                 TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (delay <= 0)
        throw new IllegalArgumentException();
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(-delay));
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    //  延遲執(zhí)行
    delayedExecute(t);
    return t;
}
private void delayedExecute(RunnableScheduledFuture<?> task) {
    if (isShutdown())
        reject(task);
    else {
        // 加入任務(wù)隊(duì)列
        super.getQueue().add(task);
        if (isShutdown() &&
            !canRunInCurrentRunState(task.isPeriodic()) &&
            remove(task))
            task.cancel(false);
        else
            // 確保執(zhí)行
            ensurePrestart();
    }
}
// 如果worker數(shù)量小于corePoolSize,創(chuàng)建新的線程，其他情況不處理
void ensurePrestart() {
    int wc = workerCountOf(ctl.get());
    if (wc < corePoolSize)
        addWorker(null, true);
    else if (wc == 0)
        addWorker(null, false);
}

五、怎么保證時(shí)間可以定時(shí)執(zhí)行

public ScheduledFuture<?> schedule(Runnable command,
                                   long delay,
                                   TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    RunnableScheduledFuture<?> t = decorateTask(command,
        new ScheduledFutureTask<Void>(command, null,
                                      triggerTime(delay, unit)));
    delayedExecute(t);
    return t;
}

在每次執(zhí)行的時(shí)候會(huì)把下一次執(zhí)行的時(shí)間放進(jìn)任務(wù)中

private long triggerTime(long delay, TimeUnit unit) {
    return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}
/**
 * Returns the trigger time of a delayed action.
 */
long triggerTime(long delay) {
    return now() +
        ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}

FutureTask 定時(shí)是通過(guò)LockSupport.parkNanos(this, nanos);LockSupport.park(this);

private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        int s = state;
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
        else if (q == null)
            q = new WaitNode();
        else if (!queued)
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
            }
            //注意這里
            LockSupport.parkNanos(this, nanos);
        }
        else //注意這里
            LockSupport.park(this);
    }
}

總結(jié)：Executor是通過(guò)將任務(wù)放在隊(duì)列中，生成的futureTask。然后將生成的任務(wù)在隊(duì)列中排序，將時(shí)間最近的需要出發(fā)的任務(wù)做檢查。如果時(shí)間不到，就阻塞線程到下次出發(fā)時(shí)間。

注意：newSingleThreadScheduledExecutor只會(huì)有一個(gè)線程，不管你提交多少任務(wù)，這些任務(wù)會(huì)順序執(zhí)行，如果發(fā)生異常會(huì)取消下面的任務(wù)，線程池也不會(huì)關(guān)閉，注意捕捉異常

六、使用

ScheduledExecutorService single = Executors.newSingleThreadScheduledExecutor();
Runnable runnable1 = () -> {
    try {
        Thread.sleep(4000);
        System.out.println("11111111111111");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
};
Runnable runnable2 = () -> {
    try {
        Thread.sleep(4000);
        System.out.println("222");
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
};
single.scheduleWithFixedDelay(runnable1,0,1, TimeUnit.SECONDS);
single.scheduleWithFixedDelay(runnable2,0,2, TimeUnit.SECONDS);

11111111111111 222 11111111111111 222 11111111111111

在項(xiàng)目中要注意關(guān)閉線程池

actionService = Executors.newSingleThreadScheduledExecutor();
        actionService.scheduleWithFixedDelay(() -> {
            try {
                Thread.currentThread().setName("robotActionService");
                Integer robotId = robotQueue.poll();
                if (robotId == null) {
                    //    關(guān)閉線程池
                    actionService.shutdown();
                } else {
                    int aiLv = robots.get(robotId);
                    if (actionQueueMap.containsKey(aiLv)) {
                        ActionQueue actionQueue = actionQueueMap.get(aiLv);
                        actionQueue.doAction(robotId);
                    }
                }
            } catch (Exception e) {
                //    捕捉異常
                LOG.error("",e);
            }
        }, 1, 1, TimeUnit.SECONDS);