Dubbo一致性哈希负载均衡的源码和Bug，了解一下？

本文是对于Dubbo负载均衡策略之一的一致性哈希负载均衡的详细分析。对源码逐行解读、根据实际运行结果，配以丰富的图片，多是东半球讲一致性哈希算法在Dubbo中的实现最详细的文章了。

文中所示源码，没有特别标注的地方，均为2.7.4.1版本。

在撰写本文的过程当中，发现了Dubbo2.7.0版本以后的一个bug。会致使性能问题，且目前还未解决，若是大家的负载均衡配置的是一致性哈希或者考虑使用一致性哈希的话，能够了解一下。

本文目录

第一节：哈希算法

本小节主要是为了介绍一致性哈希算法作铺垫。简单的介绍了哈希算法解决了什么问题，带来了什么问题。

第二节：一致性哈希算法

本小节主要是经过做图对一致性哈希进行了简单的介绍。介绍了一致性哈希是怎么解决哈希算法带来的问题，怎么解决数据倾斜的问题。

第三节：一致性哈希算法在Dubbo中的应用

本小节是全文重点，经过一个"骚"操做，对Dubbo一致性哈希算法的源码进行了十分详细的剖析。从整个类到类里面的每一个方法进行了详尽的分析，打印了大量的日志，配合图片，方便读者理解。

第四节：我又发现了一个Bug

本小节主要是介绍我在研究Dubbo一致性哈希负载均衡时遇到的一个问题，深刻研究以后发现多是一个Bug。这一小节就是比较详尽的介绍了这个Bug现象、缘由以及个人解决方案。

第五节：加入节点，画图分析

本小节对具体的案例进行了分析，并配以图片，相信能帮助读者更加深入的理解一致性哈希算法在Dubbo中的应用。

第六节：一致性哈希的应用场景

本小节主要介绍几个应用场景。使用Duboo框架，在什么样的需求可使用一致性哈希算法作负载均衡。

PS：前1、二节主要是进行了背景知识的简单铺垫，若是你了解相关背景知识，能够直接从第三节看起。本文的重点是第三到第五节。若是你只想知道Bug是什么，能够直接阅读第四节。

另：阅读本文须要对Dubbo有必定的了解。文章很长，建议收藏慢慢阅读。必定会有收获的。

哈希算法

在介绍一致性哈希算法以前，咱们看看哈希算法，以及它解决了什么问题，带来了什么问题。

如上图所示，假设0,1,2号服务器都存储的有用户信息，那么当咱们须要获取某用户信息时，由于咱们不知道该用户信息存放在哪一台服务器中，因此须要分别查询0,1,2号服务器。这样获取数据的效率是极低的。

对于这样的场景，咱们能够引入哈希算法。

仍是上面的场景，但前提是每一台服务器存放用户信息时是根据某一种哈希算法存放的。因此取用户信息的时候，也按照一样的哈希算法取便可。

假设咱们要查询用户号为100的用户信息，通过某个哈希算法，好比这里的userId mod n，即100 mod 3结果为1。因此用户号100的这个请求最终会被1号服务器接收并处理。

这样就解决了无效查询的问题。

可是这样的方案会带来什么问题呢？

扩容或者缩容时，会致使大量的数据迁移。最少也会影响百分之50的数据。

为了说明问题，咱们加入一台服务器3。服务器的数量n就从3变成了4。仍是查询用户号为100的用户信息时，100 mod 4结果为0。这时，请求就被0号服务器接收了。

当服务器数量为3时，用户号为100的请求会被1号服务器处理。

当服务器数量为4时，用户号为100的请求会被0号服务器处理。

因此，当服务器数量增长或者减小时，必定会涉及到大量数据迁移的问题。可谓是牵一发而动全身。

对于上述哈希算法其优势是简单易用，大多数分库分表规则就采起的这种方式。通常是提早根据数据量，预先估算好分区数。

其缺点是因为扩容或收缩节点致使节点数量变化时，节点的映射关系须要从新计算，会致使数据进行迁移。因此扩容时一般采用翻倍扩容，避免数据映射所有被打乱，致使全量迁移的状况，这样只会发生50%的数据迁移。

假设这是一个缓存服务，数据的迁移会致使在迁移的时间段内，有缓存是失效的。缓存失效，可怕啊。还记得我以前的文章吗，《当周杰伦把QQ音乐干翻的时候，做为程序猿我看到了什么？》就是讲缓存击穿、缓存穿透、缓存雪崩的场景和对应的解决方案。

一致性哈希算法

为了解决哈希算法带来的数据迁移问题，一致性哈希算法应运而生。

对于一致性哈希算法，官方说法以下：

一致性哈希算法在1997年由麻省理工学院提出，是一种特殊的哈希算法，在移除或者添加一个服务器时，可以尽量小地改变已存在的服务请求与处理请求服务器之间的映射关系。一致性哈希解决了简单哈希算法在分布式哈希表( Distributed Hash Table，DHT) 中存在的动态伸缩等问题。

什么意思呢？我用大白话加画图的方式给你简单的介绍一下。

一致性哈希，你能够想象成一个哈希环，它由0到2^32-1个点组成。A，B，C分别是三台服务器，每一台的IP加端口通过哈希计算后的值，在哈希环上对应以下：

当请求到来时，对请求中的某些参数进行哈希计算后，也会得出一个哈希值，此值在哈希环上也会有对应的位置，这个请求会沿着顺时针的方向，寻找最近的服务器来处理它，以下图所示：

一致性哈希就是这么个东西。那它是怎么解决服务器的扩容或收缩致使大量的数据迁移的呢？

看一下当咱们使用一致性哈希算法时，加入服务器会发什么事情。

当咱们加入一个D服务器后，假设其IP加端口，通过哈希计算后落在了哈希环上图中所示的位置。

这时影响的范围只有图中标注了五角星的区间。这个区间的请求从原来的由C服务器处理变成了由D服务器请求。而D到C，C到A，A到B这个区间的请求没有影响，加入D节点后，A、B服务器是无感知的。

因此，在一致性哈希算法中，若是增长一台服务器，则受影响的区间仅仅是新服务器(D)在哈希环空间中，逆时针方向遇到的第一台服务器(B)之间的区间，其它区间(D到C，C到A，A到B)不会受到影响。

在加入了D服务器的状况下，咱们再假设一段时间后，C服务器宕机了：

当C服务器宕机后，影响的范围也是图中标注了五角星的区间。C节点宕机后，B、D服务器是无感知的。

因此，在一致性哈希算法中，若是宕机一台服务器，则受影响的区间仅仅是宕机服务器(C)在哈希环空间中，逆时针方向遇到的第一台服务器(D)之间的区间，其它区间(C到A，A到B，B到D)不会受到影响。

综上所述，在一致性哈希算法中，无论是增长节点，仍是宕机节点，受影响的区间仅仅是增长或者宕机服务器在哈希环空间中，逆时针方向遇到的第一台服务器之间的区间，其它区间不会受到影响。

是否是很完美？

不是的，理想和现实的差距是巨大的。

一致性哈希算法带来了什么问题？

当节点不多的时候可能会出现这样的分布状况，A服务会承担大部分请求。这种状况就叫作数据倾斜。

怎么解决数据倾斜呢？加入虚拟节点。

怎么去理解这个虚拟节点呢？

首先一个服务器根据须要能够有多个虚拟节点。假设一台服务器有n个虚拟节点。那么哈希计算时，可使用IP+端口+编号的形式进行哈希值计算。其中的编号就是0到n的数字。因为IP+端口是同样的，因此这n个节点都是指向的同一台机器。

以下图所示：

在没有加入虚拟节点以前，A服务器承担了绝大多数的请求。可是假设每一个服务器有一个虚拟节点（A-1，B-1，C-1），通过哈希计算后落在了如上图所示的位置。那么A服务器的承担的请求就在必定程度上（图中标注了五角星的部分）分摊给了B-一、C-1虚拟节点，实际上就是分摊给了B、C服务器。

一致性哈希算法中，加入虚拟节点，能够解决数据倾斜问题。

当你在面试的过程当中，若是听到了相似于数据倾斜的字眼。那大几率是在问你一致性哈希算法和虚拟节点。

在介绍了相关背景后，咱们能够去看看一致性哈希算法在Dubbo中的应用了。

一致性哈希算法在Dubbo中的应用

通过《一文讲透Dubbo负载均衡之最小活跃数算法》这篇文章咱们知道Dubbo中负载均衡的实现是经过org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance中的doSelect抽象方法实现的，一致性哈希负载均衡的实现类以下所示： org.apache.dubbo.rpc.cluster.loadbalance.ConsistentHashLoadBalance

因为一致性哈希实现类看起来稍微有点抽象，不太好演示，因此我想到了一个"骚"操做。前面的文章说过LoadBalance是一个SPI接口：

既然是一个SPI接口，那咱们能够本身扩展一个如出一辙的算法，只是在算法里面加入一点输出语句方便咱们观察状况。怎么扩展SPI接口就不描述了，只要记住代码里面的输出语句都是额外加的，此外没有任何改动便可，以下:

整个类以下图片所示，请先看完整个类，有一个总体的概念后，我会进行方法级别的分析。

图片很长，其中我加了不少注释和输出语句，能够点开大图查看，必定会帮你更加好的理解一致性哈希在Dubbo中的应用：

把代码也贴在这里

public class WhyConsistentHashLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "consistenthash";

    /** * Hash nodes name */
    public static final String HASH_NODES = "hash.nodes";

    /** * Hash arguments name */
    public static final String HASH_ARGUMENTS = "hash.arguments";

    private final ConcurrentMap<String, WhyConsistentHashLoadBalance.ConsistentHashSelector<?>> selectors = 
                                    new ConcurrentHashMap<String, WhyConsistentHashLoadBalance.ConsistentHashSelector<?>>();

    @SuppressWarnings("unchecked")
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String methodName = RpcUtils.getMethodName(invocation);
        String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
        System.out.println("从selectors中获取value的key=" + key);
        //获取invokers的hashcode
        int identityHashCode = System.identityHashCode(invokers);
        WhyConsistentHashLoadBalance.ConsistentHashSelector<T> selector = 
                                                (WhyConsistentHashLoadBalance.ConsistentHashSelector<T>) selectors.get(key);
        //若是invokers是一个新的List对象，意味着服务提供者数量发生了变化，可能新增也可能减小了。
        //此时selector.identityHashCode!=identityHashCode条件成立
        //若是是第一次调用此时selector == null条件成立
        if (selector == null || selector.identityHashCode != identityHashCode) {
            System.out.println("是新的invokers:" + identityHashCode + ",原:" + (selector == null ? "null" : selector.identityHashCode));
            //建立新的ConsistentHashSelector
            selectors.put(key, new WhyConsistentHashLoadBalance.ConsistentHashSelector<T>(invokers, methodName, identityHashCode));
            selector = (WhyConsistentHashLoadBalance.ConsistentHashSelector<T>) selectors.get(key);
            
            System.out.println("哈希环构建完成,详情以下:");
            for (Map.Entry<Long, Invoker<T>> entry : selector.virtualInvokers.entrySet()) {
                System.out.println("key(哈希值)=" + entry.getKey() + ",value(虚拟节点)=" + entry.getValue());
            }
        }
        //调用ConsistentHashSelector的select方法选择Invoker
        System.out.println("开始调用ConsistentHashSelector的select方法选择Invoker");
        return selector.select(invocation);
    }

    private static final class ConsistentHashSelector<T> {
        //使用TreeMap存储Invoker的虚拟节点
        private final TreeMap<Long, Invoker<T>> virtualInvokers;
        //虚拟节点数
        private final int replicaNumber;
        //hashCode
        private final int identityHashCode;
        //请求中的参数下标。
        //须要对请求中对应下标的参数进行哈希计算
        private final int[] argumentIndex;

        ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
            this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
            this.identityHashCode = identityHashCode;
            URL url = invokers.get(0).getUrl();
            System.out.println("CHS中url为=" + url);
            //即便启动多个invoker，每一个invoker对应的url上的虚拟节点数配置的都是同样的
            //这里默认是160个。本文中的示例代码设置为4个。
            this.replicaNumber = url.getMethodParameter(methodName, HASH_NODES, 160);
            //全部输出语句都是我加的，CHS是ConsistentHashSelector的缩写
            System.out.println("CHS中url上的【hash.nodes】为=" + replicaNumber);
            //获取参与哈希计算的参数下标值，默认对第一个参数进行哈希运算
            //本文中的示例代码使用默认配置，因此这里的index长度为1。
            String[] index = COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, HASH_ARGUMENTS, "0"));
            System.out.println("CHS中url上的【hash.arguments】为=" + Arrays.toString(index));
            //for循环，对argumentIndex进行赋值操做。
            argumentIndex = new int[index.length];
            for (int i = 0; i < index.length; i++) {
                argumentIndex[i] = Integer.parseInt(index[i]);
            }
            System.out.println("CHS中argumentIndex数组为=" + Arrays.toString(argumentIndex));
            //本文中启动了2个服务提供者，因此invokers=2
            for (Invoker<T> invoker : invokers) {
                //获取每一个invoker的地址
                String address = invoker.getUrl().getAddress();
                System.out.println("CHS中invoker的地址为=" + address);
                for (int i = 0; i < replicaNumber / 4; i++) {
                    //对address+i进行md5运算获得一个长度为16的字节数组
                    byte[] digest = md5(address + i);
                    System.out.println("CHS中对" + address + i + "进行md5计算");
                    //对digest部分字节进行4次hash运算获得四个不一样的long型正整数
                    for (int h = 0; h < 4; h++) {
                        //h=0时，取digest中下标为0~3的4个字节进行位运算
                        //h=1时，取digest中下标为4~7的4个字节进行位运算
                        //h=2,h=3时过程同上
                        long m = hash(digest, h);
                        System.out.println("CHS中对digest进行第"+h+"次hash计算后的值:"+m+"，当前invoker="+invoker);
                        //将hash到invoker的映射关系存储到virtualInvokers中，
                        //virtualInvokers须要提供高效的查询操做，所以选用TreeMap做为存储结构
                        virtualInvokers.put(m, invoker);
                    }
                }
            }
        }

        public Invoker<T> select(Invocation invocation) {
            String key = toKey(invocation.getArguments());
            System.out.println("CHS的select方法根据argumentIndex取出invocation中参与hash计算的key="+key);
            byte[] digest = md5(key);
            //取digest数组的前四个字节进行hash运算,再将hash值传给selectForKey方法，
            //寻找合适的Invoker
            long hash = hash(digest, 0);
            System.out.println("CHS的select方法中key=" + key + "通过哈希计算后hash=" + hash);
            return selectForKey(hash);        
        }

        //根据argumentIndex将参数转化为key。
        private String toKey(Object[] args) {
            StringBuilder buf = new StringBuilder();
            for (int i : argumentIndex) {
                if (i >= 0 && i < args.length) {
                    buf.append(args[i]);
                }
            }
            return buf.toString();
        }

        private Invoker<T> selectForKey(long hash) {
            //到TreeMap中查找第一个节点值大于或等于当前hash的Invoker
            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
            //若是hash大于Invoker在圆环上最大的位置，此时entry=null，
            //须要将TreeMap的头节点赋值给entry
            if (entry == null) {
                entry = virtualInvokers.firstEntry();
            }
            System.out.println("CHS的selectForKey方法根据key="+hash+"选择出来的invoker="+entry.getValue());
            return entry.getValue();
        }

        private long hash(byte[] digest, int number) {
            return (((long) (digest[3 + number * 4] & 0xFF) << 24)
                    | ((long) (digest[2 + number * 4] & 0xFF) << 16)
                    | ((long) (digest[1 + number * 4] & 0xFF) << 8)
                    | (digest[number * 4] & 0xFF))
                    & 0xFFFFFFFFL;
        }

        private byte[] md5(String value) {
            MessageDigest md5;
            try {
                md5 = MessageDigest.getInstance("MD5");
            } catch (NoSuchAlgorithmException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            md5.reset();
            byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
            md5.update(bytes);
            return md5.digest();
        }

    }
}

复制代码

改造以后，咱们先把程序跑起来，有了输出就好分析了。

服务端代码以下：

其中的端口是须要手动修改的，我分别启动服务在20881和20882端口。

项目中provider.xml配置以下：

consumer.xml配置以下:

而后，启动在20881和20882端口分别启动两个服务端。客户端消费以下：

运行结果输出以下，能够先看个大概的输出，下面会对每一部分输出进行逐一的解读。

好了，用例也跑起来了，日志也有了。接下来开始结合代码和日志进行方法级别的分析。

首先是doSelect方法的入口：

从上图咱们知道了，第一次调用须要对selectors进行put操做，selectors的key是接口中定义的方法，value是ConsistentHashSelector内部类。

ConsistentHashSelector经过调用其构造函数进行初始化的。invokers(服务端)做为参数传递到了构造函数中，构造函数里面的逻辑，就是把服务端映射到哈希环上的过程，请看下图，结合代码，仔细分析输出数据：

从上图能够看出，当ConsistentHashSelector的构造方法调用完成后，8个虚拟节点在哈希环上已经映射完成。两台服务器，每一台4个虚拟节点组成了这8个虚拟节点。

doSelect方法继续执行，并打印出每一个虚拟节点的哈希值和对应的服务端，请仔细品读下图：

说明一下：上面图中的哈希环是没有考虑比例的，仅仅是展示了两个服务器在哈希环上的相对位置。并且为了演示说明方便，仅仅只有8个节点。假设咱们有4台服务器，每台服务器的虚拟节点是默认值(160)，这个状况下哈希环上一共有160*4=640个节点。

哈希环映射完成后，接下来的逻辑是把此次请求通过哈希计算后，映射到哈希环上，并顺时针方向寻找遇到的第一个节点，让该节点处理该请求：

还记得地址为468e8565的A服务器是什么端口吗？前面的图片中有哦，该服务对应的端口是20882。

最后咱们看看输出结果:

和咱们预期的一致。整个调用就算是完成了。

再对两个方法进行一个补充说明。

第一个方法是selectForKey，这个方法里面逻辑以下图所示：

虚拟节点都存储在TreeMap中。顺时针查询的逻辑由TreeMap保证。看一下下面的Demo你就明白了。

第二个方法是hash方法，其中的& 0xFFFFFFFFL的目的以下:

&是位运算符，而0xFFFFFFFFL转换为四字节表现后，其低32位全是1，因此保证了哈希环的范围是[0,Integer.MAX_VALUE]：

因此这里咱们能够改造这个哈希环的范围，假设咱们改成100000。十进制的100000对于的16进制为186A0。因此咱们改造后的哈希算法为：

再次调用后能够看到，计算后的哈希值都在10万之内。可是分布极不均匀，说明修改数据后这个哈希算法不是一个优秀的哈希算法：

以上，就是对一致性哈希算法在Dubbo中的实现的解读。须要特殊说明一下的是，和上周分享的最小活跃数负载均衡算法不一样，一致性哈希负载均衡策略和权重没有任何关系。

我又发现了一个BUG

在上篇文章中，我介绍了Dubbo 2.6.5版本以前，最小活跃数算法的两个bug。很不幸，此次我又发现了Dubbo 2.7.4.1版本，一致性哈希负载均衡策略的一个bug，我提交了issue，截止目前还未解决。

issue地址以下： github.com/apache/dubb…

我在这里详细说一下这个Bug现象、缘由和个人解决方案。

现象以下，咱们调用三次服务端：

输出日志以下（有部分删减）：

能够看到，在三次调用的过程当中并无发生服务的上下线操做，可是每一次调用都从新进行了哈希环的映射。而咱们预期的结果是应该只有在第一次调用的时候进行哈希环的映射，若是没有服务上下线的操做，后续请求根据已经映射好的哈希环进行处理。

上面输出的缘由是因为每次调用的invokers的identityHashCode发生了变化：

咱们看一下三次调用invokers的状况：

通过debug咱们能够看出由于每次调用的invokers地址值不是同一个，因此System.identityHashCode(invokers)方法返回的值都不同。

接下来的问题就是为何每次调用的invokers地址值都不同呢？

通过Debug以后，能够找到这个地方： org.apache.dubbo.rpc.cluster.RouterChain#route

问题就出在这个TagRouter中： org.apache.dubbo.rpc.cluster.router.tag.TagRouter#filterInvoker

因此，在TagRouter中的stream操做，改变了invokers，致使每次调用时其 System.identityHashCode(invokers)返回的值不同。因此每次调用都会进行哈希环的映射操做，在服务节点多，虚拟节点多的状况下会有必定的性能问题。

到这一步，问题又发生了变化。这个TagRouter怎么来的呢？

若是了解Dubbo 2.7.x版本新特性的朋友可能知道，标签路由是Dubbo2.7引入的新功能。

经过加载下面的配置加载了RouterFactrory：

META-INF\dubbo\internal\org.apache.dubbo.rpc.cluster.RouterFactory（Dubbo 2.7.0版本以前）

META-INF\dubbo\internal\com.alibaba.dubbo.rpc.cluster.RouterFactory（Dubbo 2.7.0以前）

下面是Dubbo 2.6.7(2.6.x的最后一个版本)和Dubbo 2.7.0版本该文件的对比:

能够看到确实是在Dubbo2.7.0以后引入了TagRouter。

至此，Dubbo 2.7.0版本以后，一致性哈希负载均衡算法的Bug的前因后果也介绍清楚了。目前该Bug还未解决。

解决方案是什么呢？特别简单，把获取identityHashCode的方法从System.identityHashCode(invokers)修改成invokers.hashCode()便可。

此方案是我提的issue里面的评论，这里System.identityHashCode和 hashCode之间的联系和区别就不进行展开讲述了，不清楚的你们能够自行了解一下。

改完以后，咱们再看看运行效果：

能够看到第二次调用的时候并无进行哈希环的映射操做，而是直接取到了值，进行调用。

加入节点，画图分析

最后，我再分析一种状况。在A、B、C三个服务器(2088一、2088二、20883端口)都在正常运行，哈希映射已经完成的状况下，咱们再启动一个D节点（20884端口），这时的日志输出和对应的哈希环变化状况以下:

根据日志做图以下：

根据输出日志和上图再加上源码，你再细细回味一下。我我的以为仍是讲的很是详细了，多是东半球讲一致性哈希算法在Dubbo中的实现最详细的文章了。

一致性哈希的应用场景

当你们谈到一致性哈希算法的时候，首先的第一印象应该是在缓存场景下的使用，由于在一个优秀的哈希算法加持下，其上下线节点对总体数据的影响(迁移)都是比较友好的。

可是想一下为何Dubbo在负载均衡策略里面提供了基于一致性哈希的负载均衡策略？它的实际使用场景是什么？

我最开始也想不明白。我想的是在Dubbo的场景下，假设需求是想要一个用户的请求一直让一台服务器处理，那咱们能够采用一致性哈希负载均衡策略，把用户号进行哈希计算，能够实现这样的需求。可是这样的需求未免有点太牵强了，适用场景略小。

直到有天晚上，我睡觉以前，电光火石之间忽然想到了一个稍微适用的场景了。

若是需求是须要保证某一类请求必须顺序处理呢？

若是你用其余负载均衡策略，请求分发到了不一样的机器上去，就很难保证请求的顺序处理了。好比A，B请求要求顺序处理，如今A请求先发送，被负载到了A服务器上，B请求后发送，被负载到了B服务器上。而B服务器因为性能好或者当前没有其余请求或者其余缘由极有可能在A服务器还在处理A请求以前就把B请求处理完成了。这样不符合咱们的要求。

这时，一致性哈希负载均衡策略就上场了，它帮咱们保证了某一类请求都发送到固定的机器上去执行。好比把同一个用户的请求发送到同一台机器上去执行，就意味着把某一类请求发送到同一台机器上去执行。因此咱们只须要在该机器上运行的程序中保证顺序执行就好了，好比你加一个队列。

一致性哈希算法+队列，能够实现顺序处理的需求。

最后说一句

这是Dubbo负载均衡算法的第二篇文章，上周写了一篇《一文讲透Dubbo负载均衡之最小活跃数算法》，也是很是详细，能够看看哦。

才疏学浅，不免会有纰漏，若是你发现了错误的地方，还请你留言给我指出来，我对其加以修改。

若是你以为文章还不错，你的转发、分享、赞扬、点赞、留言就是对我最大的鼓励。

感谢您的阅读，个人订阅号里全是原创，十分欢迎并感谢您的关注。

以上。

原创不易，欢迎转发，求个关注，赏个"在看"吧。