手把手带你撸zookeeper源码-zookeeper中follower启动的时候会做什么?

傷城~ 2022-11-25 10:23 129阅读 0赞

上篇文章 [手把手带你撸zookeeper源码-zookeeper通信序列化协议][zookeeper_-zookeeper]，简单说了一下zookeeper的jute序列化通信协议，写了一个简单的demo，以及如何避免粘包和拆包的，也顺带写了一点有关dubbo的序列化协议和如何自己自定义序列化协议

回到上上篇文章 [手把手带你撸zookeeper源码-zookeeper确定好角色后会做什么？][zookeeper_-zookeeper 1]主要分析了zookeer中leader启动的时候都会做些什么事，分析到了leader启动的时候会对server.x=zk1:2888:3888，中的2888端口进行监听，等待其他的follower来进行链接，如果在等待了ticket \* initLimit时间之内还有follower还没和leader进行链接，leader便不再等待，就直接对外提供服务

这篇文章主要分析一下follower启动的时候都会做什么事情？如何向leader发起注册链接的，如何从leader进行数据同步的？

回到QuorumPeer中的run方法，找到如下代码

case FOLLOWING:
                        try {
                            LOG.info("FOLLOWING");
                            setFollower(makeFollower(logFactory));
                            follower.followLeader();
                        } catch (Exception e) {
                            LOG.warn("Unexpected exception",e);
                        } finally {
                            follower.shutdown();
                            setFollower(null);
                            setPeerState(ServerState.LOOKING);
                        }
                        break;

这段代码就是当选举完leader之后，zk知道直接时follower角色之后，会进行follower角色的一系列初始化操作，我们先进入makeFollower方法

protected Follower makeFollower(FileTxnSnapLog logFactory) throws IOException {
            return new Follower(this, new FollowerZooKeeperServer(logFactory, 
                    this,new ZooKeeperServer.BasicDataTreeBuilder(), this.zkDb));
        }
         
        protected Leader makeLeader(FileTxnSnapLog logFactory) throws IOException {
            return new Leader(this, new LeaderZooKeeperServer(logFactory,
                    this,new ZooKeeperServer.BasicDataTreeBuilder(), this.zkDb));
        }
        
        protected Observer makeObserver(FileTxnSnapLog logFactory) throws IOException {
            return new Observer(this, new ObserverZooKeeperServer(logFactory,
                    this, new ZooKeeperServer.BasicDataTreeBuilder(), this.zkDb));
        }

我这里一下子粘贴了三个方法，第一个makeFollower方法，就是在当前启动的角色是follower进行实例化一个Follower对象，第二个方法makeLeader之前分析过，就是当前启动的角色是leader时进行实例化一个Leader对象，第三个makeObserver方法，就是当前启动的角色是Observer时进行实例化一个Observer对象，这块唯一不同的地方就是，传递的ZookeeperServer的子类不一样，我们先看一下ZookeeperServer的类UML图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZXJlbGk_size_16_color_FFFFFF_t_70][]

Leader角色创建了LeaderZookeeperServer

follower角色创建了FollowerZookeeperSever

observer角色创建了ObserverZookeeperServer

这个是很关键的点，因为在不同的zk角色启动的时候会调用setupRequestProcessors()这个方法，然后初始化自己的调用链processor，这块我们之后再分析

初始化完Follower对象之后，接下来会调用Follower.followerLeader的方法

// 查找leader所在服务器
                QuorumServer leaderServer = findLeader();            
                try {
                    //向leader发起连接
                    connectToLeader(leaderServer.addr, leaderServer.hostname);
                    //向leader进行注册, 经过三次握手
                    long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO);
    
                    long newEpoch = ZxidUtils.getEpochFromZxid(newEpochZxid);
                    if (newEpoch < self.getAcceptedEpoch()) {
                        LOG.error("Proposed leader epoch " + ZxidUtils.zxidToString(newEpochZxid)
                                + " is less than our accepted epoch " + ZxidUtils.zxidToString(self.getAcceptedEpoch()));
                        throw new IOException("Error: Epoch of leader is lower");
                    }
                    syncWithLeader(newEpochZxid);
                    QuorumPacket qp = new QuorumPacket();
                    while (this.isRunning()) {
                        readPacket(qp); //从leader读数据
                        processPacket(qp);
                    }
                } catch (Exception e) {
                }

这个方法会做一些列的事情，我们一步步分析

第一件事

QuorumServer leaderServer = findLeader();

根据自己本地zoo.cfg中配置的server.x和保存leader sid做对比，然后获取到leader server地址

//向leader发起连接
    connectToLeader(leaderServer.addr, leaderServer.hostname);

找到了leader server之后就开始向leader发起链接请求

protected void connectToLeader(InetSocketAddress addr, String hostname)
                throws IOException, ConnectException, InterruptedException {
            sock = new Socket();        
            sock.setSoTimeout(self.tickTime * self.initLimit);
            // 最多可以重复连接5次，如果超过5此都没连接成功, 则当前follower放弃和leader进行连接
            for (int tries = 0; tries < 5; tries++) {
                try {
                    sock.connect(addr, self.tickTime * self.syncLimit);
                    sock.setTcpNoDelay(nodelay);
                    break;
                } catch (IOException e) {
                    if (tries == 4) {
                        LOG.error("Unexpected exception",e);
                        throw e;
                    } else {
                        LOG.warn("Unexpected exception, tries="+tries+
                                ", connecting to " + addr,e);
                        sock = new Socket();
                        sock.setSoTimeout(self.tickTime * self.initLimit);
                    }
                }
                Thread.sleep(1000);
            }
    
            self.authLearner.authenticate(sock, hostname);
            // 通过socket获取输入输出流，并包装为jute序列化协议对象,
            // 接下来可以直接通过jute进行数据的序列化、反序列化读取数据和发送数据
            leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(
                    sock.getInputStream()));
            bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
            leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);
        }

这个方法就是向leader发起链接的代码，就是socket之间的链接，每个follower会发起5此向leader链接的请求，如果一直失败，超过5此以后，就不再向发起链接了

leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(
                    sock.getInputStream()));
    bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
    leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);

最后的代码就是如果和leader链接成功之后，会把socket中的输入输出流封装为jute序列化协议对象leaderIs读数据和leaderOs写数据，之后如果要发送数据或者读取数据直接拿来使用即可

接下来registerWithLeader()方法会向leader发起三次握手建立一个following链接

protected long registerWithLeader(int pktType) throws IOException{
        	long lastLoggedZxid = self.getLastLoggedZxid();
            QuorumPacket qp = new QuorumPacket();                
            qp.setType(pktType);
            qp.setZxid(ZxidUtils.makeZxid(self.getAcceptedEpoch(), 0));
            
            /*
             * Add sid to payload
             * 包装当前的sid为learnerInfo对象
             */
            LearnerInfo li = new LearnerInfo(self.getId(), 0x10000);
            ByteArrayOutputStream bsid = new ByteArrayOutputStream();
            BinaryOutputArchive boa = BinaryOutputArchive.getArchive(bsid);
            boa.writeRecord(li, "LearnerInfo");
            qp.setData(bsid.toByteArray());
            //发送sid和协议版本号给leader,
            writePacket(qp, true);
            readPacket(qp);        
            final long newEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
            // 第一次接收到leader发送过来的数据，leader发送过来的协议版本号
    		if (qp.getType() == Leader.LEADERINFO) {
            	// we are connected to a 1.0 server so accept the new epoch and read the next packet
            	leaderProtocolVersion = ByteBuffer.wrap(qp.getData()).getInt();
            	byte epochBytes[] = new byte[4];
            	final ByteBuffer wrappedEpochBytes = ByteBuffer.wrap(epochBytes);
            	if (newEpoch > self.getAcceptedEpoch()) {
            		wrappedEpochBytes.putInt((int)self.getCurrentEpoch());
            		self.setAcceptedEpoch(newEpoch);
            	} else if (newEpoch == self.getAcceptedEpoch()) {
                    wrappedEpochBytes.putInt(-1);
            	} else {
            		throw new IOException("Leaders epoch, " + newEpoch + " is less than accepted epoch, " + self.getAcceptedEpoch());
            	}
            	QuorumPacket ackNewEpoch = new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);
            	// 发送ack
            	writePacket(ackNewEpoch, true);
                return ZxidUtils.makeZxid(newEpoch, 0);
            } else {
            	if (newEpoch > self.getAcceptedEpoch()) {
            		self.setAcceptedEpoch(newEpoch);
            	}
                if (qp.getType() != Leader.NEWLEADER) {
                    LOG.error("First packet should have been NEWLEADER");
                    throw new IOException("First packet should have been NEWLEADER");
                }
                return qp.getZxid();
            }
        }

有两种情况，如果zookeeper刚刚启动或者是刚新加入进来的follower，本地没有数据。第二种就是集群中某一台宕机了，本地中保存的有自己最新的zxid和可以接受的acceptEpoch(epoch之前有提到过，就是一旦有leader变更，epoch就会加一)

上次握手大概时这样子的

第一次: follower启动的时候，会根据epoch来生成一个zxid和epoch(leader版本号)，还有sid以及协议版本号(0x10000)发送给leader

第二次: follower接收到leader返回回来的信息，然后读取leader发送回来的epoch(当前leader的版本)，如果leader发送过来的epoch比自己本地的epoch要大，则更新自己本地的epoch，和leader保持同步，如果时相等的，那么最后一步的ack可以不用再次发送epoch给leader，如果leader发送过来的epoch小于本地的epoch，则会抛出异常，此时有可能因为zookeeper集群脑裂，导致链接上了一个版本较低的leader

第三次: follower会发送一个ack给leader，把自己最新的zxid发送给leader

才想一下最终把zxid发送给leader干啥？

我们可以想想，如果我当前的zk是从故障中恢复的，那么此时我本地的数据肯定时落后于leader的，我把我的zxid发送给leader，那么leader会根据自己的数据来判断一下有多少数据需要同步给我，需不需要全量同步？

其实zk故障重启之后如何同步数据有好几种情况，我们在下面代码中来详细说

回到Follower.followerLeader()方法中，我们继续下面的代码

// 校验一下leader的zxid是否小于我们的, 这种情肯定不会发生，只是做个安全检查
                    long newEpoch = ZxidUtils.getEpochFromZxid(newEpochZxid);
                    if (newEpoch < self.getAcceptedEpoch()) {
                        LOG.error("Proposed leader epoch " + ZxidUtils.zxidToString(newEpochZxid)
                                + " is less than our accepted epoch " + ZxidUtils.zxidToString(self.getAcceptedEpoch()));
                        throw new IOException("Error: Epoch of leader is lower");
                    }

这块代码上面写了注释, 就是做一个安全检查

又一个关键代码来了

syncWithLeader(newEpochZxid);

看方法名字: 同步leader，我们看看代码，上上面的代码我们把本地的最新zxid发送到leader了

protected void syncWithLeader(long newLeaderZxid) throws IOException, InterruptedException{
            QuorumPacket ack = new QuorumPacket(Leader.ACK, 0, null, null);
            QuorumPacket qp = new QuorumPacket();
            long newEpoch = ZxidUtils.getEpochFromZxid(newLeaderZxid);
            
            boolean snapshotNeeded = true;
    
            readPacket(qp);
            LinkedList<Long> packetsCommitted = new LinkedList<Long>();
            LinkedList<PacketInFlight> packetsNotCommitted = new LinkedList<PacketInFlight>();
            synchronized (zk) {
                // 如果和leader之间有不同的数据, 则不需要快照
                if (qp.getType() == Leader.DIFF) {
                    LOG.info("Getting a diff from the leader 0x{}", Long.toHexString(qp.getZxid()));
                    // 如果当前follower宕机，然后恢复重启，此时会落后leader一部分数据，然后去同步宕机之后的数据即可
                    snapshotNeeded = false;
                }
                // 从leader获取快照信息
                else if (qp.getType() == Leader.SNAP) {
                    LOG.info("Getting a snapshot from leader 0x" + Long.toHexString(qp.getZxid()));
                    // The leader is going to dump the database
                    // clear our own database and read
                    // 清空本地的数据文件和内存数据库，从leader中读取快照数据，进行反序列化
                    // 可能当前的zk服务器新加入集群的，此时会从leader进行全量同步数据
                    zk.getZKDatabase().clear();
                    zk.getZKDatabase().deserializeSnapshot(leaderIs);
                    // 读取签名
                    String signature = leaderIs.readString("signature");
                    if (!signature.equals("BenWasHere")) {
                        LOG.error("Missing signature. Got " + signature);
                        throw new IOException("Missing signature");                   
                    }
                    zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
                } else if (qp.getType() == Leader.TRUNC) {
                    //we need to truncate the log to the lastzxid of the leader
                    // 根据leader的lastzxid对本地日志进行截断
                    // 有这样的一种场景，当前zk服务器原来是leader，然后有客户端发送过来数据，写入本地日志文件，还没来得及发送给follower
                    // 此时leader宕机了，然后原有的集群中某个follower会被选为leader，此时相当于就丢掉了一条数据
                    // 当挂掉的leader宕机恢复之后，会作为一个follower加入集群中，此时回和leader进行同步，因为此时的leader中是没有这条数据的
                    // 所以会把宕机之前最后的一条数据给删除掉
                    LOG.warn("Truncating log to get in sync with the leader 0x"
                            + Long.toHexString(qp.getZxid()));
                    boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());
                    if (!truncated) {
                        System.exit(13);
                    }
                    zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
                }
                else {
                    LOG.error("Got unexpected packet from leader "
                            + qp.getType() + " exiting ... " );
                    System.exit(13);
    
                }
                // 创建会话跟踪器
                zk.createSessionTracker();
                
                long lastQueued = 0;
    
                boolean writeToTxnLog = !snapshotNeeded;
                // we are now going to start getting transactions to apply followed by an UPTODATE
                outerLoop:
                while (self.isRunning()) {
                    readPacket(qp);
                    switch(qp.getType()) {
                    case Leader.PROPOSAL:
                        PacketInFlight pif = new PacketInFlight();
                        pif.hdr = new TxnHeader();
                        pif.rec = SerializeUtils.deserializeTxn(qp.getData(), pif.hdr);
                        if (pif.hdr.getZxid() != lastQueued + 1) {
                        LOG.warn("Got zxid 0x"
                                + Long.toHexString(pif.hdr.getZxid())
                                + " expected 0x"
                                + Long.toHexString(lastQueued + 1));
                        }
                        lastQueued = pif.hdr.getZxid();
                        packetsNotCommitted.add(pif);
                        break;
                    case Leader.COMMIT:
                        if (!writeToTxnLog) {
                            pif = packetsNotCommitted.peekFirst();
                            if (pif.hdr.getZxid() != qp.getZxid()) {
                                LOG.warn("Committing " + qp.getZxid() + ", but next proposal is " + pif.hdr.getZxid());
                            } else {
                                zk.processTxn(pif.hdr, pif.rec);
                                packetsNotCommitted.remove();
                            }
                        } else {
                            packetsCommitted.add(qp.getZxid());
                        }
                        break;
                    case Leader.INFORM:
                       
                        PacketInFlight packet = new PacketInFlight();
                        packet.hdr = new TxnHeader();
                        packet.rec = SerializeUtils.deserializeTxn(qp.getData(), packet.hdr);
                        lastQueued = packet.hdr.getZxid();
                        if (!writeToTxnLog) {
                            // Apply to db directly if we haven't taken the snapshot
                            zk.processTxn(packet.hdr, packet.rec);
                        } else {
                            packetsNotCommitted.add(packet);
                            packetsCommitted.add(qp.getZxid());
                        }
                        break;
                    case Leader.UPTODATE:   // 标识同步leader数据已经同步完了，可以接收客户端发起的请求了
                        if (isPreZAB1_0) {
                            zk.takeSnapshot();
                            self.setCurrentEpoch(newEpoch);
                        }
                        self.cnxnFactory.setZooKeeperServer(zk);                
                        break outerLoop;    // 跳出循环
                    case Leader.NEWLEADER: 
                        File updating = new File(self.getTxnFactory().getSnapDir(),
                                            QuorumPeer.UPDATING_EPOCH_FILENAME);
                        if (!updating.exists() && !updating.createNewFile()) {
                            throw new IOException("Failed to create " +
                                                  updating.toString());
                        }
                        if (snapshotNeeded) {
                            zk.takeSnapshot();
                        }
                        self.setCurrentEpoch(newEpoch);
                        if (!updating.delete()) {
                            throw new IOException("Failed to delete " +
                                                  updating.toString());
                        }
                        writeToTxnLog = true; //Anything after this needs to go to the transaction log, not applied directly in memory
                        isPreZAB1_0 = false;
                        writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);
                        break;
                    }
                }
            }
    }

这块代码有点多，我也只是截取了一部分，今天主要分析一下上面的代码，这个方法截取到了487行，剩余的代码我们之后分析

QuorumPacket ack = new QuorumPacket(Leader.ACK, 0, null, null);
            QuorumPacket qp = new QuorumPacket();
            long newEpoch = ZxidUtils.getEpochFromZxid(newLeaderZxid);
            boolean snapshotNeeded = true;
    
            readPacket(qp);

这块很简单，就是从leader中读取发送过来的数据

if (qp.getType() == Leader.DIFF) {
                    LOG.info("Getting a diff from the leader 0x{}", Long.toHexString(qp.getZxid()));
                    // 如果当前follower宕机，然后恢复重启，此时会落后leader一部分数据，然后去同步宕机之后的数据即可
                    snapshotNeeded = false;
                }
                // 从leader获取快照信息
                else if (qp.getType() == Leader.SNAP) {
                    LOG.info("Getting a snapshot from leader 0x" + Long.toHexString(qp.getZxid()));
                    // The leader is going to dump the database
                    // clear our own database and read
                    // 清空本地的数据文件和内存数据库，从leader中读取快照数据，进行反序列化
                    // 可能当前的zk服务器新加入集群的，此时会从leader进行全量同步数据
                    zk.getZKDatabase().clear();
                    zk.getZKDatabase().deserializeSnapshot(leaderIs);
                    // 读取签名
                    String signature = leaderIs.readString("signature");
                    if (!signature.equals("BenWasHere")) {
                        LOG.error("Missing signature. Got " + signature);
                        throw new IOException("Missing signature");                   
                    }
                    zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
                } else if (qp.getType() == Leader.TRUNC) {
                    //we need to truncate the log to the lastzxid of the leader
                    // 根据leader的lastzxid对本地日志进行截断
                    // 有这样的一种场景，当前zk服务器原来是leader，然后有客户端发送过来数据，写入本地日志文件，还没来得及发送给follower
                    // 此时leader宕机了，然后原有的集群中某个follower会被选为leader，此时相当于就丢掉了一条数据
                    // 当挂掉的leader宕机恢复之后，会作为一个follower加入集群中，此时回和leader进行同步，因为此时的leader中是没有这条数据的
                    // 所以会把宕机之前最后的一条数据给删除掉
                    LOG.warn("Truncating log to get in sync with the leader 0x"
                            + Long.toHexString(qp.getZxid()));
                    boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());
                    if (!truncated) {
                        // not able to truncate the log
                        LOG.error("Not able to truncate the log "
                                + Long.toHexString(qp.getZxid()));
                        System.exit(13);
                    }
                    zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
                }
                else {
                    LOG.error("Got unexpected packet from leader "
                            + qp.getType() + " exiting ... " );
                    System.exit(13);
    
                }

大家也可以看着我写的注释进行理解，几个分支我们单独分析一下

// 如果和leader之间有不同的数据, 则不需要快照
                if (qp.getType() == Leader.DIFF) {
                    LOG.info("Getting a diff from the leader 0x{}", Long.toHexString(qp.getZxid()));
                    // 如果当前follower宕机，然后恢复重启，此时会落后leader一部分数据，然后去同步宕机之后的数据即可
                    snapshotNeeded = false;
                }

这个分支的情况就是如果zk集群中某个follower宕机了，然后进行了重启，那么此时他直接取leader中取同步它宕机期间未同步的数据即可，同步数据在下面的代码中

if (qp.getType() == Leader.SNAP) {
                    LOG.info("Getting a snapshot from leader 0x" + Long.toHexString(qp.getZxid()));
                    // The leader is going to dump the database
                    // clear our own database and read
                    // 清空本地的数据文件和内存数据库，从leader中读取快照数据，进行反序列化
                    // 可能当前的zk服务器新加入集群的，此时会从leader进行全量同步数据
                    zk.getZKDatabase().clear();
                    zk.getZKDatabase().deserializeSnapshot(leaderIs);
                    // 读取签名
                    String signature = leaderIs.readString("signature");
                    if (!signature.equals("BenWasHere")) {
                        LOG.error("Missing signature. Got " + signature);
                        throw new IOException("Missing signature");                   
                    }
                    zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
                }

第二个分支，这应该有两种情况下会发生，第一个情况就是如果在现有zk集群中新加入进来follower或者observer，那么此时它们刚开始是没有数据的，此时肯定会进行全量的同步，直接把leader中的快照全部同步过来。另一个情况有可能是这样的，就是leader判断当前follower和leader之间的数据相差太大，评估一下增量同步比快照全量同步的代价要大，也可能会触发全量快照同步，这块代码主要是先清空一下本地的内存数据库

if (qp.getType() == Leader.TRUNC) {
                    //we need to truncate the log to the lastzxid of the leader
                    // 根据leader的lastzxid对本地日志进行截断
                    // 有这样的一种场景，当前zk服务器原来是leader，然后有客户端发送过来数据，写入本地日志文件，还没来得及发送给follower
                    // 此时leader宕机了，然后原有的集群中某个follower会被选为leader，此时相当于就丢掉了一条数据
                    // 当挂掉的leader宕机恢复之后，会作为一个follower加入集群中，此时回和leader进行同步，因为此时的leader中是没有这条数据的
                    // 所以会把宕机之前最后的一条数据给删除掉
                    LOG.warn("Truncating log to get in sync with the leader 0x"
                            + Long.toHexString(qp.getZxid()));
                    boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());
                    if (!truncated) {
                        // not able to truncate the log
                        LOG.error("Not able to truncate the log "
                                + Long.toHexString(qp.getZxid()));
                        System.exit(13);
                    }
                    zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
                }

第三种情况很特殊，只有是leader挂了之后才会出现这种情况

大家知道zk集群写数据保持了顺序一致性，并且是2PC的过半写机制，此时就导致了这种情况

比方说，现在客户端写入一条数据，然后请求最终会转发给leader，leader回先把数据写入到本地日志文件，然后在发送proposal给其他的follower，但是在写入本地磁盘成功之后，还没来得及发送给其他的follower proposal消息，忽然leader宕机或者网络出现了问题导致其他的follower认为leader宕机了。因为zk集群肯定是会从follower中重新选举一个leader，而所有的follower都没有收到这个proposal。此时新的leader选举出来了。旧的leader宕机之后可能运维重启，也可能网络恢复了。此时它发现集群中已经有了leader，那么它会作为一个follower节点加入到集群中取。此时follower会把自己的最新zxid发送给leader，leader发现自己根本就没有这个zxid，然后旧leader必须把这个zxid对应的数据从日志和内存中给删除掉，然后再继续从leader中同步因为宕机或者网络原因导致没有同步到的最新数据。大概就是这种情况，大家可以好好思考一下

这篇文章先写到这里吧，下篇文章再继续分析接下来的代码

[zookeeper_-zookeeper]: https://blog.csdn.net/liereli/article/details/107826354
[zookeeper_-zookeeper 1]: https://blog.csdn.net/liereli/article/details/107774543
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZXJlbGk_size_16_color_FFFFFF_t_70]: /images/20221124/64c814342cf44f31a9bd9acd1aba6db8.png