RocketMQ源码解析：ProcessQueue的作用

冷不防 2023-10-02 11:27 4阅读 0赞

### 一、背景 ###

今天准备写了关于Consumer获取消息的流程，即`pullMessage()`。结果在看到拉取消息的报文包装类`PullRequest`时，它其中有个队列消费快照`ProcessQueue`，进入源码一看，卧槽，500多行代码。看来有必要专门分析一下它了，也就有了今天的故事。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]

### 二、对ProcessQueue的疑问？ ###

我们发现在消息消费的过程中很多地方都用到了ProcessQueue，比如：PullMessageService、DefaultMQPushConsumerImpl、ConsumerMessageOrderlyService。  
我们思考一下：ProcessQueue是干什么的？它提供了哪些能力/方法？哪些地方用到了它？因为什么用到它？

### 三、ProcessQueue介绍 ###

#### 1、ProcessQueue简介 ####

源码上关于ProcessQueue的注释：Queue consumption snapshot。  
言简意赅，就很nice。  
它是一个队列消费快照，消息拉取的时候，会将实际消息体、拉取相关的操作存放在其中。比如消费进度，消费等等功能的底层核心数据保存都是有ProcessQueue提供。

另外在拉取消息的时候使用的是PullRequest去请求，其结构如下：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_19_color_FFFFFF_t_70_g_se_x_16]  
可以看到，ProcessQueue和一个MessageQueue是对应的，MessageQueue用来表示拉取回来的消息元数据信息：topic、brokerName、queueId。

**ProcessQueue几个重要的成员变量：**

/**
     * 读写锁：对消息的操作都要使用
     */
    private final ReadWriteLock lockTreeMap = new ReentrantReadWriteLock();
    /**
     * 临时存放消息用的
     */
    private final TreeMap<Long, MessageExt> msgTreeMap = new TreeMap<Long, MessageExt>();
    /**
     * 消息总数量
     */
    private final AtomicLong msgCount = new AtomicLong();
    /**
     * 整个ProcessQueue处理单元的总消息长度
     */
    private final AtomicLong msgSize = new AtomicLong();
    private final Lock lockConsume = new ReentrantLock();
    /**
     * A subset of msgTreeMap, will only be used when orderly consume
     * 一个临时的TreeMap，仅在顺序消费模式下使用
     */
    private final TreeMap<Long, MessageExt> consumingMsgOrderlyTreeMap = new TreeMap<Long, MessageExt>();
    private final AtomicLong tryUnlockTimes = new AtomicLong(0);
    /**
     * 整个ProcessQueue处理单元的offset最大边界
     */
    private volatile long queueOffsetMax = 0L;
    /**
     * 是否正在消费
     */
    private volatile boolean consuming = false;
    /**
     * broker端还有多少条消息没被处理（拉取消息的那一刻）
     */
    private volatile long msgAccCnt = 0;

**注意：每次对msgTreeMap、msgCount、msgSize进行操作时，都会加上读写锁lockTreeMap。**

#### 2、包含的方法 ####

我们主要看下图框中的方法：  
![请添加图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_15_color_FFFFFF_t_70_g_se_x_16]

##### 1）isLockExpired() #####

该方法是在处理消息时，判断拥有的锁是否过期，默认过期时间30s。

public boolean isLockExpired() {
        
            return (System.currentTimeMillis() - this.lastLockTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;
        }

public final static long REBALANCE_LOCK_MAX_LIVE_TIME =
                Long.parseLong(System.getProperty("rocketmq.client.rebalance.lockMaxLiveTime", "30000"));

**那么该方法是在哪些地方使用到了呢？**  
![在这里插入图片描述][db22956fbd9844ff844fb7d1f5d84f07.png]  
从上图我们可以看出，isLockExpired()方法在顺序消费时使用，用来判断向broker请求消息时，对ProcessQueue锁定时间是否超过阈值(默认30000ms，可以通过rocketmq.client.rebalance.lockMaxLiveTime设置)，如果没有超时，代表还是持有锁，超过这个时间则失效。

##### 2）isPullExpired() #####

判断拉取消息是否过期，默认两分钟（可以通过rocketmq.client.pull.pullMaxIdleTime设置）

public boolean isLockExpired() {
        
            return (System.currentTimeMillis() - this.lastLockTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME;
        }

private final static long PULL_MAX_IDLE_TIME = Long.parseLong(System.getProperty("rocketmq.client.pull.pullMaxIdleTime", "120000"));

**那么该方法是在哪些地方使用到了呢？**  
在RebalanceImpl\#updateProcessQueueTableInRebalance()方法中，如下所示：

if (mq.getTopic().equals(topic)) {
        
        if (!mqSet.contains(mq)) {
        
            .......
        } else if (pq.isPullExpired()) {
        
        .......
        }
    }

在负载均衡更新ProcessQueueTable时调用，如果拉取失效，ProcessQueue将被丢弃。

##### 3）cleanExpiredMsg() #####

清理过期的消息，规则如下：

*  最多16条消息一处理；
 *  消息在客户端存在`超过15分钟就被认为已过期`，然后从本地缓存中移除，以10s的延时消息方式发送会Broker。

\*\*注意：\*\*顺序消费不清理过期消息，进并发消费模式才清理过期消息。

public void cleanExpiredMsg(DefaultMQPushConsumer pushConsumer) {
        
        // 顺序消费不清理过期消息
        if (pushConsumer.getDefaultMQPushConsumerImpl().isConsumeOrderly()) {
        
            return;
        }
    
        // 每次最多清理16条消息
        int loop = msgTreeMap.size() < 16 ? msgTreeMap.size() : 16;
        for (int i = 0; i < loop; i++) {
        
            MessageExt msg = null;
            try {
        
                // 处理消息之前，先拿到读锁
                this.lockTreeMap.readLock().lockInterruptibly();
                try {
        
                    // 临时存放消息的treeMap不为空，并且 判断当前时间 - TreeMap里第一条消息的开始消费时间 > 15分钟
                    if (!msgTreeMap.isEmpty() && System.currentTimeMillis() - Long.parseLong(MessageAccessor.getConsumeStartTimeStamp(msgTreeMap.firstEntry().getValue())) > pushConsumer.getConsumeTimeout() * 60 * 1000) {
        
                        // 把第一条消息拿出来
                        msg = msgTreeMap.firstEntry().getValue();
                    } else {
        
    
                        break;
                    }
                } finally {
        
                    // 释放读锁
                    this.lockTreeMap.readLock().unlock();
                }
            } catch (InterruptedException e) {
        
                log.error("getExpiredMsg exception", e);
            }
    
            try {
        
    
                // 把过期消息以延时消息方式重新发给broker，10s之后才能消费。
                pushConsumer.sendMessageBack(msg, 3);
                log.info("send expire msg back. topic={}, msgId={}, storeHost={}, queueId={}, queueOffset={}", msg.getTopic(), msg.getMsgId(), msg.getStoreHost(), msg.getQueueId(), msg.getQueueOffset());
                try {
        
                    // 获取写锁
                    this.lockTreeMap.writeLock().lockInterruptibly();
                    try {
        
                        if (!msgTreeMap.isEmpty() && msg.getQueueOffset() == msgTreeMap.firstKey()) {
        
                            try {
        
                                // 将过期消息从本地缓存中的消息列表中移除掉，  Collections.singletonList表示只有一个元素的List集合
                                removeMessage(Collections.singletonList(msg));
                            } catch (Exception e) {
        
                                log.error("send expired msg exception", e);
                            }
                        }
                    } finally {
        
                        // 释放写锁
                        this.lockTreeMap.writeLock().unlock();
                    }
                } catch (InterruptedException e) {
        
                    log.error("getExpiredMsg exception", e);
                }
            } catch (Exception e) {
        
                log.error("send expired msg exception", e);
            }
        }
    }

**方法被调用的地方：**

// 启动定时任务（每15分钟跑一次），清理过期消息
    public void start() {
        
        this.cleanExpireMsgExecutors.scheduleAtFixedRate(new Runnable() {
        
    
            @Override
            public void run() {
        
                cleanExpireMsg();
            }
    
        }, this.defaultMQPushConsumer.getConsumeTimeout(), this.defaultMQPushConsumer.getConsumeTimeout(), TimeUnit.MINUTES);
    }
    
    // 针对每个ProcessQueue每次最多清理16条过期消息
    private void cleanExpireMsg() {
        
            Iterator<Map.Entry<MessageQueue, ProcessQueue>> it =
                this.defaultMQPushConsumerImpl.getRebalanceImpl().getProcessQueueTable().entrySet().iterator();
            while (it.hasNext()) {
        
                Map.Entry<MessageQueue, ProcessQueue> next = it.next();
                ProcessQueue pq = next.getValue();
                pq.cleanExpiredMsg(this.defaultMQPushConsumer);
            }
        }

并发消费ConsumeMessageConcurrentlyService\#start()方法中启动了一个定时任务，每15分钟调用一次cleanExpiredMsg()清理过期消息：超过15分钟未被消费，针对每个ProcessQueue每次最多清理16条消息。

##### 4）putMessage() #####

用来初始化属性：msgTreeMap（存放拉取到的消息）、queueOffsetMax（消费到的最大offset）、msgSize（ProcessQueue中的总消息长度）、msgCount（ProcessQueue中的总消息数量）、msgAccCnt（表示broker端还有多少消息未被消费）。

public boolean putMessage(final List<MessageExt> msgs) {
        
        // 这个只有在顺序消费的时候才会遇到，并发消费不会用到
        boolean dispatchToConsume = false;
        try {
        
            this.lockTreeMap.writeLock().lockInterruptibly();
            try {
        
                //有效消息数量
                int validMsgCnt = 0;
                for (MessageExt msg : msgs) {
        
                    // 把传过来的消息都都放在msgTreeMap中，以消息在queue中的offset作为key，msg做为value
                    MessageExt old = msgTreeMap.put(msg.getQueueOffset(), msg);
                    // 正常情况，说明原本msgTreeMap中不包含此条消息
                    if (null == old) {
        
                        validMsgCnt++;
                        // 将最后一个消息的offset赋值给queueOffsetMax
                        this.queueOffsetMax = msg.getQueueOffset();
                        // 把当前消息的长度加到msgSize中
                        msgSize.addAndGet(msg.getBody().length);
                    }
                }
                // 增加有效消息数量
                msgCount.addAndGet(validMsgCnt);
    
                // msgTreeMap不为空(含有消息)，并且不是正在消费状态
                // // 这个值在放消息的时候会设置为true，在顺序消费模式，取不到消息则设置为false
                if (!msgTreeMap.isEmpty() && !this.consuming) {
        
                    // 将ProcessQueue置为正在被消费状态
                    // 有消息，且为未消费状态，则顺序消费模式可以消费
                    dispatchToConsume = true;
                    this.consuming = true;
                }
    
                if (!msgs.isEmpty()) {
        
                    // 拿到最后一条消息
                    MessageExt messageExt = msgs.get(msgs.size() - 1);
                    // 获取broker端（拉取消息时）queue里最大的offset，maxOffset会存在每条消息里
                    String property = messageExt.getProperty(MessageConst.PROPERTY_MAX_OFFSET);
                    // 计算broker端还有多少条消息没有被消费
                    if (property != null) {
        
                        // broker端的最大偏移量 - 当前ProcessQueue中处理的最大消息偏移量
                        long accTotal = Long.parseLong(property) - messageExt.getQueueOffset();
                        if (accTotal > 0) {
        
                            this.msgAccCnt = accTotal;
                        }
                    }
                }
            } finally {
        
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (InterruptedException e) {
        
            log.error("putMessage exception", e);
        }
    
        return dispatchToConsume;
    }

**方法被调用的地方：**

> MQClientInstance\#start()方法中调用pullMessageService\#start()，  
> PullMessageService本身是一个ServiceThread，它的run()方法中会调用DefaultMQPushConsumerImpl\#pullMessage()方法，  
> 然后在其中new出的PullCallBack的onSuccess()方法中：
> 
>  *  当传入的pullResult的pullStatus是FOUND时，并且PullRequest的msgFoundList不为空时 会调用该putMessage()方法。

![请添加图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
在拉取消息后处理消息时，在ProcessQueue中通过putMessage()方法，往红黑树msgTreeMap存储一分消息数据，并统计消息总数量、消息总长度、broker端还有多少消息未被拉取。返回结果dispatchToConsume如果为true，表示消费集合有数据，顺序消息会据此构造消费请求继续处理。

##### 5）getMaxSpan() #####

返回processQueue中处理的一批消息中最大offset和最小offset之间的差距。

public long getMaxSpan() {
        
        try {
        
            this.lockTreeMap.readLock().lockInterruptibly();
            try {
        
                if (!this.msgTreeMap.isEmpty()) {
        
                    // msgTreeMap中最后一个消息的offset - 第一个消息的offset
                    return this.msgTreeMap.lastKey() - this.msgTreeMap.firstKey();
                }
            } finally {
        
                this.lockTreeMap.readLock().unlock();
            }
        } catch (InterruptedException e) {
        
            log.error("getMaxSpan exception", e);
        }
    
        return 0;
    }

**方法被使用的地方：**  
DefaultMQPushConsumerImpl\#pullMessage()方法中，会根据这个判断当前还有多少消息未处理，如果大于某个值(默认2000)，则进行流控处理：将PullRequest先存起来，默认过50ms再执行拉消息逻辑。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]

##### 6）removeMessage() #####

该方法将从ProcessQueue中移除部分消息，`并行消费`模式中使用。

public long removeMessage(final List<MessageExt> msgs) {
        
        // 当前消费进度的下一个offset
        long result = -1;
        final long now = System.currentTimeMillis();
        try {
        
            // 获取写锁
            this.lockTreeMap.writeLock().lockInterruptibly();
            // 最后消费的时间
            this.lastConsumeTimestamp = now;
            try {
        
                if (!msgTreeMap.isEmpty()) {
        
                    result = this.queueOffsetMax + 1;
                    // 删除的消息个数
                    int removedCnt = 0;
                    // 遍历消息，将其从Treemap中移除
                    for (MessageExt msg : msgs) {
        
                        MessageExt prev = msgTreeMap.remove(msg.getQueueOffset());
                        // 不为null，说明删除成功
                        if (prev != null) {
        
                            // 已经移除的消息数量
                            removedCnt--;
                            // ProcessQueue中消息的长度 - 当前消息长度
                            msgSize.addAndGet(0 - msg.getBody().length);
                        }
                    }
                    // ProcessQueue中的消息数量 - 删除的消息数量，即加上removedCnt（是一个负数）
                    msgCount.addAndGet(removedCnt);
    
                    // 如果还有消息存在，则使用msgTreeMap中的第一消息的offset（即最小offset）作为消费进度
                    if (!msgTreeMap.isEmpty()) {
        
                        result = msgTreeMap.firstKey();
                    }
                }
            } finally {
        
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (Throwable t) {
        
            log.error("removeMessage exception", t);
        }
    
        return result;
    }

**方法被调用的地方：**  
ConsumeMessageConcurrentlyService\#processConsumeResult()中  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 3]  
removeMessage()方法在并发消费后进行调用，消息处理完之后将ProcessQueue中msgTreeMap红黑树的这批消息移除。

#### 以下的方法全部是在顺序消费模式下使用 ####

##### 7）takeMessages() #####

该方法在顺序消费模式下使用，取到消息后，就会调用我们定义的MessageListener进行消费。  
调用takeMessages从msgTreeMap中获取消息后：其内部会将消息都放在一个临时的TreeMap中，然后进行顺序消费。

public List<MessageExt> takeMessages(final int batchSize) {
        
        List<MessageExt> result = new ArrayList<MessageExt>(batchSize);
        final long now = System.currentTimeMillis();
        try {
        
            this.lockTreeMap.writeLock().lockInterruptibly();
            this.lastConsumeTimestamp = now;
            try {
        
                if (!this.msgTreeMap.isEmpty()) {
        
                    // 从 msgTreeMap中获取batchSize条数据，每次都返回offset最小的那条消息并从msgTreeMap中移除
                    for (int i = 0; i < batchSize; i++) {
        
                        Map.Entry<Long, MessageExt> entry = this.msgTreeMap.pollFirstEntry();
                        if (entry != null) {
        
                            // 把消息放到返回列表
                            result.add(entry.getValue());
                            // 把消息的offset和消息体msg，放到顺序消费TreeMap中
                            consumingMsgOrderlyTreeMap.put(entry.getKey(), entry.getValue());
                        } else {
        
                            break;
                        }
                    }
                }
    
                // 没有取到消息，说明不需要消费，即将consuming置为FALSE。
                if (result.isEmpty()) {
        
                    consuming = false;
                }
            } finally {
        
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (InterruptedException e) {
        
            log.error("take Messages exception", e);
        }
    
        return result;
    }

**方法被调用的地方：**  
ConsumeMessageOrderlyService.ConsumeRequest\#run()方法中：

List<MessageExt> msgs = this.processQueue.takeMessages(consumeBatchSize);

递归往上层找，找到最终调用的地方：

> 在ConsumeMessageOrderlyService\#submitConsumeRequest()方法中会new一个ConsumeRequest（是一个Runnable实现类），然后将其放到线程中执行；即将消息丢入消费服务线程中进行消费  
> 而submitConsumeRequest()方法被调用的场景有两种：  
> 1. 集群消费模式下调用submitConsumeRequestLater方法（位于ConsumeMessageService的两个实现类中）重新提交client端消费请求时，  
> 2. DefaultMQPushConsumerImpl\#pullMessage()中拉取消息成功之后，将消息丢入消费服务线程中进行消费。

##### 8）commit() #####

在顺序消费模式下，调用takeMessages从msgTreeMap中获取消息：其内部会将消息都放在一个临时的TreeMap中，然后进行消费。消费完消息之后，我们需要调用commit()方法将这个临时的TreeMap清除。

public long commit() {
        
        try {
        
            this.lockTreeMap.writeLock().lockInterruptibly();
            try {
        
                // 获取临时TreeMap中最后一个消息的offset
                Long offset = this.consumingMsgOrderlyTreeMap.lastKey();
                // 消费完成之后，减去该批次的消息数量
                msgCount.addAndGet(0 - this.consumingMsgOrderlyTreeMap.size());
    
                // 维护ProcessQueue中的总消息长度
                for (MessageExt msg : this.consumingMsgOrderlyTreeMap.values()) {
        
                    // 减去每条已消费消息的长度
                    msgSize.addAndGet(0 - msg.getBody().length);
                }
                // 清除临时TreeMap中的所有消息
                this.consumingMsgOrderlyTreeMap.clear();
                // 返回下一个消费进度
                if (offset != null) {
        
                    return offset + 1;
                }
            } finally {
        
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (InterruptedException e) {
        
            log.error("commit exception", e);
        }
    
        return -1;
    }

**方法被调用的地方：**  
ConsumeMessageOrderlyService\#processConsumeResult()方法中

if (context.isAutoCommit()) {
        
        switch (status) {
        
            case SUCCESS:
                //清空msgTreeMapTemp
                commitOffset = consumeRequest.getProcessQueue().commit();
                ......
        }
    } else {
        
        switch (status) {
        
            case COMMIT:
                //清空msgTreeMapTemp
                commitOffset = consumeRequest.getProcessQueue().commit();
                break;
         }
    }

##### 9）rollback() #####

该方法把临时TreeMap中的消息全部放到msgTreeMap中，等待下一次消费。

public void rollback() {
        
        try {
        
            this.lockTreeMap.writeLock().lockInterruptibly();
            try {
        
                // 把临时TreeMap中的消息全部放到msgTreeMap中，等待下一次消费
                this.msgTreeMap.putAll(this.consumingMsgOrderlyTreeMap);
                this.consumingMsgOrderlyTreeMap.clear();
            } finally {
        
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (InterruptedException e) {
        
            log.error("rollback exception", e);
        }
    }

**方法被调用的地方：**  
ConsumerMessageOrderlyService\#processConsumeResult()中

case ROLLBACK:
    consumeRequest.getProcessQueue().rollback();

在顺序消费客户端处理消息后，如果消息处理结果的状态为ROLLBACK，此时调用ProcessQueue\#rollback方法；将msgTreeMapTmp中的消息重新写回msgTreeMap中；`ROLLBACK状态在顺序消费中已不建议使用`。

##### 10）makeMessageToConsumeAgain() #####

顺序消费模式下，使消息可以被重新消费。

public void makeMessageToConsumeAgain(List<MessageExt> msgs) {
        
        try {
        
            this.lockTreeMap.writeLock().lockInterruptibly();
            try {
        
                for (MessageExt msg : msgs) {
        
                    // 从临时TreeMap中取出消息，
                    this.consumingMsgOrderlyTreeMap.remove(msg.getQueueOffset());
                    //再将消息放到msgTreeMap中
                    this.msgTreeMap.put(msg.getQueueOffset(), msg);
                }
            } finally {
        
                this.lockTreeMap.writeLock().unlock();
            }
        } catch (InterruptedException e) {
        
            log.error("makeMessageToCosumeAgain exception", e);
        }
    }

**方法被调用的地方：**  
ConsumerMessageOrderlyService\#processConsumeResult()

case SUSPEND_CURRENT_QUEUE_A_MOMENT:
        this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size());
        if (checkReconsumeTimes(msgs)) {
        
            consumeRequest.getProcessQueue().makeMessageToConsumeAgain(msgs);
            .....
            }

makeMessageToCosumeAgain()方法在顺序消费模式下，当客户端返回消息状态为SUSPEND\_CURRENT\_QUEUE\_A\_MOMENT时调用；其将消息从msgTreeMapTemp移除，并将该批消息重新放入msgTreeMap，使这批消息可以被再次消费。

### 三、总结 ###

ProcessQueue作为MessageQueue在Consumer端的镜像，从负载均衡、消息拉取、消费状态处理、offset提交等功能层面，控制着整个消费的脉搏，尤其是在顺序消费中。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/47b3aad95a6c4dce8f7097f9f16ae294.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_19_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/33c6602d50b2413c8fa6c66ff74d4a30.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_19,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_15_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/67900a1b054b4e0f9fda2eb9cbd43f8a.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_15,color_FFFFFF,t_70,g_se,x_16
[db22956fbd9844ff844fb7d1f5d84f07.png]: https://img-blog.csdnimg.cn/db22956fbd9844ff844fb7d1f5d84f07.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/7e3d616940004caa90626ca1cd7556e4.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/d3ae332e80904d3ea428a1d53336a53e.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 3]: https://img-blog.csdnimg.cn/681d072ffa474279af80fd2650b3a150.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16