前言

上一篇文章寫了Redis分布式鎖的原理和缺陷，覺(jué)得有些不過(guò)癮，只是簡(jiǎn)單的介紹了下Redisson這個(gè)框架，具體的原理什么的還沒(méi)說(shuō)過(guò)呢。趁年前項(xiàng)目忙的差不多了，反正閑著也是閑著，不如把Redisson的源碼也學(xué)習(xí)一遍好了。

雖說(shuō)是一時(shí)興起，但仔細(xì)研究之后發(fā)現(xiàn)Redisson的源碼解讀工作量還是挺大的，其中用到了大量的Java并發(fā)類，并且引用了Netty作為通信工具，實(shí)現(xiàn)與Redis組件的遠(yuǎn)程調(diào)用，這些知識(shí)點(diǎn)如果要全部講解的話不太現(xiàn)實(shí)，本文的重點(diǎn)主要是關(guān)于Redisson分布式鎖的實(shí)現(xiàn)原理，所以網(wǎng)絡(luò)通信和并發(fā)原理這塊的代碼解讀不會(huì)太仔細(xì)，有不足之處還望見(jiàn)諒！

Redis 發(fā)布訂閱

之前說(shuō)過(guò)，分布式鎖的核心功能其實(shí)就三個(gè)：加鎖、解鎖、設(shè)置鎖超時(shí)。這三個(gè)功能也是我們研究Redisson分布式鎖原理的方向。

在學(xué)習(xí)之前，我們有必要先了解一個(gè)知識(shí)點(diǎn)，就是有關(guān)Redis的發(fā)布訂閱功能。

Redis 發(fā)布訂閱 (pub/sub) 是一種消息通信模式：發(fā)送者 (pub) 發(fā)送消息，訂閱者 (sub) 接收消息，發(fā)布者可以向指定的渠道 (channel) 發(fā)送消息，訂閱者如果訂閱了該頻道的話就能收到消息，從而實(shí)現(xiàn)多個(gè)客戶端的通信效果。

訂閱的命令是SUBSCRIBE channel[channel ...]，可以訂閱一個(gè)或多個(gè)頻道，當(dāng)有新消息通過(guò)PUBLISH命令發(fā)送給頻道時(shí)，訂閱者就能收到消息，就好像這樣

開(kāi)啟兩個(gè)客戶端，一個(gè)訂閱了頻道channel1，另一個(gè)通過(guò)PUBLISH發(fā)送消息后，訂閱的那個(gè)就能收到了，靠這種模式就能實(shí)現(xiàn)不同客戶端之間的通信。

當(dāng)然，關(guān)于這種通信模式有哪些妙用場(chǎng)景我們就不展開(kāi)了，大家可以自己去網(wǎng)上查閱一下，我們的主角還是Redisson，熱身完畢，該上主菜了。

Redisson源碼

在使用Redisson加鎖之前，需要先獲取一個(gè)RLock實(shí)例對(duì)象，有了這個(gè)對(duì)象就可以調(diào)用lock、tryLock方法來(lái)完成加鎖的功能

Config config = new Config();
config.useSingleServer()
  .setPassword("")
  .setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);
// RLock對(duì)象
RLock lock = redisson.getLock("myLock");

配置好對(duì)應(yīng)的host，然后就可以創(chuàng)建一個(gè)RLock對(duì)象。RLock是一個(gè)接口，具體的同步器需要實(shí)現(xiàn)該接口，當(dāng)我們調(diào)用redisson.getLock()時(shí)，程序會(huì)初始化一個(gè)默認(rèn)的同步執(zhí)行器RedissonLock

這里面初始化了幾個(gè)參數(shù)，

commandExecutor：異步的Executor執(zhí)行器，Redisson中所有的命令都是通過(guò)...Executor 執(zhí)行的 ；

id：唯一ID，初始化的時(shí)候是用UUID創(chuàng)建的；

internalLockLeaseTime：等待獲取鎖時(shí)間，這里讀的是配置類中默認(rèn)定義的，時(shí)間為30秒；

同時(shí)，圖片里我還標(biāo)注了一個(gè)方法getEntryName，返回的是 “ID ：鎖名稱” 的字符串，代表的是當(dāng)前線程持有對(duì)應(yīng)鎖的一個(gè)標(biāo)識(shí)，這些參數(shù)有必要留個(gè)印象，后面的源碼解析中經(jīng)常會(huì)出現(xiàn)。

說(shuō)完了初始化的東西，我們就可以開(kāi)始學(xué)習(xí)加鎖和解鎖的源碼了。

加鎖

Redisson的加鎖方法有兩個(gè)，tryLock和lock，使用上的區(qū)別在于tryLock可以設(shè)置鎖的過(guò)期時(shí)長(zhǎng)leaseTime和等待時(shí)長(zhǎng)waitTime，核心處理的邏輯都差不多，我們先從tryLock講起。

tryLock

代碼有點(diǎn)長(zhǎng)啊。。。整成圖片不太方便，直接貼上來(lái)吧，

/**
 * @param waitTime 等待鎖的時(shí)長(zhǎng) 
 * @param leaseTime 鎖的持有時(shí)間 
 * @param unit 時(shí)間單位
 * @return
 * @throws InterruptedException
 */
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {    
        // 剩余的等待鎖的時(shí)間
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        
        final long threadId = Thread.currentThread().getId();
        // 嘗試獲取鎖，如果沒(méi)取到鎖，則返回鎖的剩余超時(shí)時(shí)間
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // ttl為null，說(shuō)明可以搶到鎖了，返回true
        if (ttl == null) {
            return true;
        }
        
        // 如果waitTime已經(jīng)超時(shí)了，就返回false，代表申請(qǐng)鎖失敗
        time -= (System.currentTimeMillis() - current);
        if (time <= 0) {
            acquireFailed(threadId);
            return false;
        }
        
        current = System.currentTimeMillis();
        // 訂閱分布式鎖, 解鎖時(shí)進(jìn)行通知，看，這里就用到了我們上面說(shuō)的發(fā)布-訂閱了吧
        final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        // 阻塞等待鎖釋放，await()返回false，說(shuō)明等待超時(shí)了
        if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
                    @Override
                    public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
                        if (subscribeFuture.isSuccess()) {
                         // 等待都超時(shí)了，直接取消訂閱
                            unsubscribe(subscribeFuture, threadId);
                        }
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }

        try {
            time -= (System.currentTimeMillis() - current);
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
            }
         // 進(jìn)入死循環(huán)，反復(fù)去調(diào)用tryAcquire嘗試獲取鎖，跟上面那一段拿鎖的邏輯一樣
            while (true) {
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }

                time -= (System.currentTimeMillis() - currentTime);
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }

                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= (System.currentTimeMillis() - currentTime);
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
            }
        } finally {
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

代碼還是挺長(zhǎng)的，不過(guò)流程也就兩步，要么線程拿到鎖返回成功；要么沒(méi)拿到鎖并且等待時(shí)間還沒(méi)過(guò)就繼續(xù)循環(huán)拿鎖，同時(shí)監(jiān)聽(tīng)鎖是否被釋放。

拿鎖的方法是tryAcquire，傳入的參數(shù)分別是鎖的持有時(shí)間，時(shí)間單位以及代表當(dāng)前線程的ID，跟進(jìn)代碼查看調(diào)用棧，它會(huì)調(diào)到一個(gè)叫做tryAcquireAsync的方法：

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
    return get(tryAcquireAsync(leaseTime, unit, threadId));
}

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
        // 如果有設(shè)置鎖的等待時(shí)長(zhǎng)的話，就直接調(diào)用tryLockInnerAsync方法獲取鎖
        if (leaseTime != -1) {
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
        // 沒(méi)有設(shè)置等待鎖的時(shí)長(zhǎng)的話，加多一個(gè)監(jiān)聽(tīng)器，也就是調(diào)用lock.lock()會(huì)跑的邏輯，后面會(huì)說(shuō)
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        ttlRemainingFuture.addListener(new FutureListener<Long>() {
            @Override
            public void operationComplete(Future<Long> future) throws Exception {
                if (!future.isSuccess()) {
                    return;
                }

                Long ttlRemaining = future.getNow();
                // lock acquired
                if (ttlRemaining == null) {
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }

我們繼續(xù)跟，看看tryLockInnerAsync方法的源碼：

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);

    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
              "if (redis.call('exists', KEYS[1]) == 0) then " +
                  "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                  "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
              "end; " +
              "return redis.call('pttl', KEYS[1]);",
                Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
String getLockName(long threadId) {
    return id + ":" + threadId;
}

這里就是底層的調(diào)用棧了，直接操作命令，整合成lua腳本后，調(diào)用netty的工具類跟redis進(jìn)行通信，從而實(shí)現(xiàn)獲取鎖的功能。

這段腳本命令還是有點(diǎn)意思的，簡(jiǎn)單解讀一下：

• 先用exists key命令判斷是否鎖是否被占據(jù)了，沒(méi)有的話就用hset命令寫入，key為鎖的名稱，field為“客戶端唯一ID:線程ID”，value為1；
• 鎖被占據(jù)了，判斷是否是當(dāng)前線程占據(jù)的，是的話value值加1；
• 鎖不是被當(dāng)前線程占據(jù)，返回鎖剩下的過(guò)期時(shí)長(zhǎng)；

命令的邏輯并不復(fù)雜，但不得不說(shuō)，作者的設(shè)計(jì)還是很有心的，用了redis的Hash結(jié)構(gòu)存儲(chǔ)數(shù)據(jù)，如果發(fā)現(xiàn)當(dāng)前線程已經(jīng)持有鎖了，就用hincrby命令將value值加1，value的值將決定釋放鎖的時(shí)候調(diào)用解鎖命令的次數(shù)，達(dá)到實(shí)現(xiàn)鎖的可重入性效果。

每一步命令對(duì)應(yīng)的邏輯我都在下面的圖中標(biāo)注了，大家可以讀一下：

我們繼續(xù)跟代碼吧，根據(jù)上面的命令可以看出，如果線程拿到鎖的話，tryLock方法會(huì)直接返回true，萬(wàn)事大吉。

拿不到的話，就會(huì)返回鎖的剩余過(guò)期時(shí)長(zhǎng)，這個(gè)時(shí)長(zhǎng)有什么作用呢？我們回到tryLock方法中死循環(huán)的那個(gè)地方：

這里有一個(gè)針對(duì)waitTime和key的剩余過(guò)期時(shí)間大小的比較，取到二者中比較小的那個(gè)值，然后用Java的Semaphore信號(hào)量的tryAcquire方法來(lái)阻塞線程。

那么Semaphore信號(hào)量又是由誰(shuí)控制呢，何時(shí)才能release呢。這里又需要回到上面來(lái)看，各位看官應(yīng)該還記得，我們上面貼的tryLock代碼中還有這一段：

current = System.currentTimeMillis();
// 訂閱分布式鎖, 解鎖時(shí)進(jìn)行通知
final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);

訂閱的邏輯顯然是在subscribe方法里，跟著方法的調(diào)用鏈，它會(huì)進(jìn)入到PublishSubscribe.Java中：

這段代碼的作用在于將當(dāng)前線程的threadId添加到一個(gè)AsyncSemaphore中，并且設(shè)置一個(gè)redis的監(jiān)聽(tīng)器，這個(gè)監(jiān)聽(tīng)器是通過(guò)redis的發(fā)布、訂閱功能實(shí)現(xiàn)的。

一旦監(jiān)聽(tīng)器收到redis發(fā)來(lái)的消息，就從中獲取與當(dāng)前thread相關(guān)的，如果是鎖被釋放的消息，就立馬通過(guò)操作Semaphore（也就是調(diào)用release方法）來(lái)讓剛才阻塞的地方釋放。

釋放后線程繼續(xù)執(zhí)行，仍舊是判斷是否已經(jīng)超時(shí)。如果還沒(méi)超時(shí)，就進(jìn)入下一次循環(huán)再次去獲取鎖，拿到就返回true，沒(méi)有拿到的話就繼續(xù)流程。

這里說(shuō)明一下，之所以要循環(huán)，是因?yàn)殒i可能會(huì)被多個(gè)客戶端同時(shí)爭(zhēng)搶，線程阻塞被釋放之后的那一瞬間很可能還是拿不到鎖，但是線程的等待時(shí)間又還沒(méi)過(guò)，這個(gè)時(shí)候就需要重新跑循環(huán)去拿鎖。

這就是tryLock獲取鎖的整個(gè)過(guò)程了，畫一張流程圖的話表示大概是這樣：

lock

除了tryLock，一般我們還經(jīng)常直接調(diào)用lock來(lái)獲取鎖，lock的拿鎖過(guò)程跟tryLock基本是一致的，區(qū)別在于lock沒(méi)有手動(dòng)設(shè)置鎖過(guò)期時(shí)長(zhǎng)的參數(shù)，該方法的調(diào)用鏈也是跑到tryAcquire方法來(lái)獲取鎖的，不同的是，它會(huì)跑到這部分的邏輯：

這段代碼做了兩件事：

1、預(yù)設(shè)30秒的過(guò)期時(shí)長(zhǎng)，然后去獲取鎖

2、開(kāi)啟一個(gè)監(jiān)聽(tīng)器，如果發(fā)現(xiàn)拿到鎖了，就開(kāi)啟定時(shí)任務(wù)不斷去刷新該鎖的過(guò)期時(shí)長(zhǎng)

刷新過(guò)期時(shí)長(zhǎng)的方法是scheduleExpirationRenewal，貼一下源碼吧：

private void scheduleExpirationRenewal(final long threadId) {
 // expirationRenewalMap是一個(gè)ConcurrentMap，存儲(chǔ)標(biāo)志為"當(dāng)前線程ID:key名稱"的任務(wù)
        if (expirationRenewalMap.containsKey(getEntryName())) {
            return;
        }

        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                // 檢測(cè)鎖是否存在的lua腳本，存在的話就用pexpire命令刷新過(guò)期時(shí)長(zhǎng)
                RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                            "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                            "return 1; " +
                        "end; " +
                        "return 0;",
                          Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
                
                future.addListener(new FutureListener<Boolean>() {
                    @Override
                    public void operationComplete(Future<Boolean> future) throws Exception {
                        expirationRenewalMap.remove(getEntryName());
                        if (!future.isSuccess()) {
                            log.error("Can't update lock " + getName() + " expiration", future.cause());
                            return;
                        }
                        
                        if (future.getNow()) {
                            // reschedule itself
                            scheduleExpirationRenewal(threadId);
                        }
                    }
                });
            }
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

        if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
            task.cancel();
        }
    }

代碼的流程比較簡(jiǎn)單，大概就是開(kāi)啟一個(gè)定時(shí)任務(wù)，每隔internalLockLeaseTime / 3的時(shí)間（這個(gè)時(shí)間是10秒）就去檢測(cè)鎖是否還被當(dāng)前線程持有，是的話就重新設(shè)置過(guò)期時(shí)長(zhǎng)internalLockLeaseTime，也就是30秒的時(shí)間。

而這些定時(shí)任務(wù)會(huì)存儲(chǔ)在一個(gè)ConcurrentHashMap對(duì)象expirationRenewalMap中，存儲(chǔ)的key就為“線程ID:key名稱”，如果發(fā)現(xiàn)expirationRenewalMap中不存在對(duì)應(yīng)當(dāng)前線程key的話，定時(shí)任務(wù)就不會(huì)跑，這也是后面解鎖中的一步重要操作。

上面這段代碼就是Redisson中所謂的”看門狗“程序，用一個(gè)異步線程來(lái)定時(shí)檢測(cè)并執(zhí)行的，以防手動(dòng)解鎖之前就過(guò)期了。

其他的邏輯就跟tryLock()基本沒(méi)什么兩樣啦，大家看一下就知道了

解鎖

有拿鎖的方法，自然也就有解鎖。Redisson分布式鎖解鎖的上層調(diào)用方法是unlock()，默認(rèn)不用傳任何參數(shù)

@Override
    public void unlock() {
     // 發(fā)起釋放鎖的命令請(qǐng)求
        Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId()));
        if (opStatus == null) {
            throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                    + id + " thread-id: " + Thread.currentThread().getId());
        }
        if (opStatus) {
         // 成功釋放鎖，取消"看門狗"的續(xù)時(shí)線程
            cancelExpirationRenewal();
        }
    }

解鎖相關(guān)的命令操作在unlockInnerAsync方法中定義，

又是一大串的lua腳本，比起前面加鎖那段腳本的命令稍微復(fù)雜了點(diǎn)，不過(guò)沒(méi)關(guān)系，我們簡(jiǎn)單梳理一下，命令的邏輯大概是這么幾步：

1、判斷鎖是否存在，不存在的話用publish命令發(fā)布釋放鎖的消息，訂閱者收到后就能做下一步的拿鎖處理；

2、鎖存在但不是當(dāng)前線程持有，返回空置nil；

3、當(dāng)前線程持有鎖，用hincrby命令將鎖的可重入次數(shù)-1，然后判斷重入次數(shù)是否大于0，是的話就重新刷新鎖的過(guò)期時(shí)長(zhǎng)，返回0，否則就刪除鎖，并發(fā)布釋放鎖的消息，返回1；

當(dāng)線程完全釋放鎖后，就會(huì)調(diào)用cancelExpirationRenewal()方法取消"看門狗"的續(xù)時(shí)線程

void cancelExpirationRenewal() {
 // expirationRenewalMap移除對(duì)應(yīng)的key，就不會(huì)執(zhí)行當(dāng)前線程對(duì)應(yīng)的"看門狗"程序了
    Timeout task = expirationRenewalMap.remove(getEntryName());
    if (task != null) {
        task.cancel();
    }
}

這就是釋放鎖的過(guò)程了，怎么樣，是不是還是比較簡(jiǎn)單的，閱讀起來(lái)比加鎖那份代碼舒服多了，當(dāng)然啦，簡(jiǎn)單歸簡(jiǎn)單，為了方便你們理清整個(gè)分布式鎖的過(guò)程，我當(dāng)然還是費(fèi)心費(fèi)力的給你們畫流程圖展示下啦（就沖這點(diǎn)，是不是該給我來(lái)個(gè)三連啊，哈哈）：

RedLock

以上就是Redisson分布式鎖的原理講解，總的來(lái)說(shuō)，就是簡(jiǎn)單的用lua腳本整合基本的set命令實(shí)現(xiàn)鎖的功能，這也是很多Redis分布式鎖工具的設(shè)計(jì)原理。除此之外，Redisson還支持用"RedLock算法"來(lái)實(shí)現(xiàn)鎖的效果，這個(gè)工具類就是RedissonRedLock。

用法也很簡(jiǎn)單，創(chuàng)建多個(gè)Redisson Node, 由這些無(wú)關(guān)聯(lián)的Node就可以組成一個(gè)完整的分布式鎖

RLock lock1 = Redisson.create(config1).getLock(lockKey);
RLock lock2 = Redisson.create(config2).getLock(lockKey);
RLock lock3 = Redisson.create(config3).getLock(lockKey);

RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
try {
   redLock.lock();
} finally {
   redLock.unlock();
}

RedLock算法原理方面我就不細(xì)說(shuō)了，大家有興趣可以看我之前的文章，或者是網(wǎng)上搜一下，簡(jiǎn)單的說(shuō)就是能一定程度上能有效防止Redis實(shí)例單點(diǎn)故障的問(wèn)題，但并不完全可靠，不管是哪種設(shè)計(jì)，光靠Redis本身都是無(wú)法保證鎖的強(qiáng)一致性的。

還是那句話，魚和熊掌不可兼得，性能和安全方面也往往如此，Redis強(qiáng)大的性能和使用的方便足以滿足日常的分布式鎖需求，如果業(yè)務(wù)場(chǎng)景對(duì)鎖的安全隱患無(wú)法忍受的話，最保底的方式就是在業(yè)務(wù)層做冪等處理。

總結(jié)

看了本文的源碼解析，相信各位看官對(duì)Redisson分布式鎖的設(shè)計(jì)也有了足夠的了解，當(dāng)然啦，雖然是講解源碼，我們的主要精力還是放在分布式鎖的原理上，一些無(wú)關(guān)流程的代碼就沒(méi)有帶大家字斟酌句的解讀了，大家有興趣的話可以自己去閱讀看看，源碼中很多地方都展示了一些基礎(chǔ)并發(fā)工具和網(wǎng)絡(luò)通信的妙用之處，學(xué)習(xí)一下還是挺有收獲的。

最后我還是想吐槽一下，Redisson的注釋是真的少啊。。。。。。

如果您覺(jué)得文章有用的話，歡迎點(diǎn)個(gè)贊支持一下，這將是對(duì)我創(chuàng)作的最好鼓勵(lì)！

作者：鄙人薛某，一個(gè)不拘于技術(shù)的互聯(lián)網(wǎng)人，喜歡用通俗易懂的語(yǔ)言來(lái)解構(gòu)后端技術(shù)的知識(shí)點(diǎn)，想看更多精彩文章的可以關(guān)注我的公眾號(hào)，微信搜索【鄙人薛某】即可關(guān)注