快捷導(dǎo)航

Redisson分布式閉鎖RCountDownLatch的使用詳細(xì)講解

更新時(shí)間：2023年02月11日 11:04:12 作者：每天都要進(jìn)步一點(diǎn)點(diǎn)

分布式鎖和我們java基礎(chǔ)中學(xué)習(xí)到的synchronized略有不同，synchronized中我們的鎖是個(gè)對(duì)象，當(dāng)前系統(tǒng)部署在不同的服務(wù)實(shí)例上，單純使用synchronized或者lock已經(jīng)無(wú)法滿足對(duì)庫(kù)存一致性的判斷。本次主要講解基于rediss實(shí)現(xiàn)的分布式鎖

一、RCountDownLatch的使用

RCountDownLatch的功能跟CountDownLatch，用于實(shí)現(xiàn)某個(gè)線程需要等待其他線程都完成之后，我再去執(zhí)行，這種場(chǎng)景就可以使用CountDownLatch。

@Test
public void testRCountDownLatch() {
    Config config = new Config();
    config.useSingleServer().setAddress("redis://127.0.0.1:6379");
    RedissonClient redissonClient = Redisson.create(config);
    RCountDownLatch rCountDownLatch = redissonClient.getCountDownLatch("anyCountDownLatch");
    rCountDownLatch.trySetCount(5);
    for (int i = 1; i <= 5; i++) {
        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "離開(kāi)教師...");
            rCountDownLatch.countDown();
        }, "A" + i).start();
    }
    try {
        rCountDownLatch.await();
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
    System.out.println("班長(zhǎng)鎖門...");
}

A1離開(kāi)教師...
A2離開(kāi)教師...
A4離開(kāi)教師...
A3離開(kāi)教師...
A5離開(kāi)教師...
班長(zhǎng)鎖門...

二、trySetCount()設(shè)置計(jì)數(shù)器

/**
 * 僅當(dāng)先前的計(jì)數(shù)已達(dá)到零或根本未設(shè)置時(shí)才設(shè)置新的計(jì)數(shù)值。
 */
boolean trySetCount(long count);

public RFuture<Boolean> trySetCountAsync(long count) {
    return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            // 往redis中寫(xiě)入一個(gè)String類型的數(shù)據(jù)    anyCountDownLatch:5
            "if redis.call('exists', KEYS[1]) == 0 then "
                + "redis.call('set', KEYS[1], ARGV[2]); "
                + "redis.call('publish', KEYS[2], ARGV[1]); "
                + "return 1 "
            + "else "
                + "return 0 "
            + "end",
            Arrays.asList(getRawName(), getChannelName()), CountDownLatchPubSub.NEW_COUNT_MESSAGE, count);
}

同樣，在redis中寫(xiě)入了一個(gè){key}:{計(jì)數(shù)器總數(shù)}的String類型的數(shù)據(jù)。

三、countDown()源碼

減少鎖存器的計(jì)數(shù)器。當(dāng)計(jì)數(shù)達(dá)到零時(shí)通知所有等待線程。

public RFuture<Void> countDownAsync() {
    return commandExecutor.evalWriteNoRetryAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                    // 減少redis中計(jì)數(shù)器的值
                    "local v = redis.call('decr', KEYS[1]);" +
                    // 計(jì)數(shù)器減為0后，刪除對(duì)應(yīng)的key        
                    "if v <= 0 then redis.call('del', KEYS[1]) end;" +
                    "if v == 0 then redis.call('publish', KEYS[2], ARGV[1]) end;",
                Arrays.<Object>asList(getRawName(), getChannelName()), CountDownLatchPubSub.ZERO_COUNT_MESSAGE);
}

四、await()源碼

等到計(jì)數(shù)器達(dá)到零。

public void await() throws InterruptedException {
    // 如果計(jì)數(shù)器為0，直接返回
    if (getCount() == 0) {
        return;
    }
    // 訂閱redisson_countdownlatch__channel__{anyCountDownLatch}的消息
    CompletableFuture<RedissonCountDownLatchEntry> future = subscribe();
    RedissonCountDownLatchEntry entry = commandExecutor.getInterrupted(future);
    try {
        // 不斷循環(huán)判斷計(jì)數(shù)器的值是否大于0，大于0說(shuō)明還有線程沒(méi)執(zhí)行完成，在這里阻塞：LockSupport.park(this)
        while (getCount() > 0) {
            // waiting for open state
            entry.getLatch().await();
        }
    } finally {
        unsubscribe(entry);
    }
}

到此這篇關(guān)于Redisson分布式閉鎖RCountDownLatch的使用詳細(xì)講解的文章就介紹到這了,更多相關(guān)Redisson RCountDownLatch內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: