1.基本概念

• 硬件的监控

+ 机器CPU
+ 内存
+ 磁盘
+ 网络

• 业务的监控

+ 可能出现的问题


    - 数据库的主从延迟变长
    - 接口响应时间变长
    - 系统中出现大量错误
+ 监控的指标如何选择？
+ 采集指标有哪些方法和途径
+ 指标采集到如何处理和展示

2.监控指标如何选择（监控什么？）

• 谷歌分布式系统监控，4个黄金信号量：

+ 延迟


    - 请求的响应时间
    - 访问数据库和缓存的时间
+ 通信量 吞吐量
+ 错误


    - 网络服务错误 4xx 5xx
    - error
+ 饱和度


    - CPU使用率
    - 内存使用率
    - 磁盘使用率
    - 缓存数据库连接数

• RED指标体系\

+ R 代表请求量（Request rate）\
+ E 代表错误（Error）\
+ D 代表响应时间（Duration\

• 根据业务特色进行监控

+ 数据库主从延迟数据
+ 下次队列对滴情况
+ 缓存的命中率

\

3.如何采集数据指标

采集方式

• Agent

+ 在具体的机器上启动监控服务
+ 收集数据，发送给监控系统


    - 连接memcached客户端
    - JMX 内存信息、GC信息、kafka

• 代码中埋点

+ 将计算调用资源或者服务耗时、调用量，并发送给监控服务器

• 日志

+ 通过成熟的日志采集工具


    - Apache Flume
    - Fluentd\
    - Filebeat

4.监控数据的处理与存储

• 1.首先使用消息队列承接数据
• 2.处理数据

+ 把数据写入到 Elasticsearch（反索引），然后通过 Kibana 展示数据
+ 流式处理的中间件，比如 Spark、Storm\


    - 解析日志
    - 聚合运算
    - 可以存储在时序数据库中
    - 绘制报表

• 3.形成报表

+ 访问趋势报表，整体运行情况\
+ 性能报表，程序中的阿米点\
+ 资源报表，具体资源的报表\

5.总结

• 监控指标

+ 耗时
+ 请求量
+ 错误数

• 数据采集方式

+ Agent
+ 埋点
+ 日志

• 监控体系

+ 趋势报表 woter
+ 性能报表 bamai
+ 资源报表 odin

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

本文正在参与 「掘力星计划」 ，赢取创作大礼包，挑战创作激励金。​

作者主页：Java李杨勇 简历模板、学习资料、面试题库、技术互助【关注我，都给你】

在我们日常开发项目中经常用到正则表达式/比如邮箱/电话手机号/域名/ip等)都会经常用到
其实一个字符串就是一个简单的正则表达式，例如 **Hello World** 正则表达式匹配 “Hello World” 字符串。
.（点号）也是一个正则表达式，它匹配任何一个字符如：”a” 或 “1”。
下表列出了一些正则表达式的实例及描述：

java 正则表达式和 Perl 的是最为相似的。

java.util.regex 包主要包括以下三个类：

• Pattern 类：

pattern 对象是一个正则表达式的编译表示。Pattern 类没有公共构造方法。要创建一个 Pattern 对象，你必须首先调用其公共静态编译方法，它返回一个 Pattern 对象。该方法接受一个正则表达式作为它的第一个参数。

• Matcher 类：

Matcher 对象是对输入字符串进行解释和匹配操作的引擎。与Pattern 类一样，Matcher 也没有公共构造方法。你需要调用 Pattern 对象的 matcher 方法来获得一个 Matcher 对象。

• PatternSyntaxException：

PatternSyntaxException 是一个非强制异常类，它表示一个正则表达式模式中的语法错误。
以下实例中使用了正则表达式 .runoob. 用于查找字符串中是否包了 runoob 子串：

1
2
3
4
5
6
7
8
9

java复制代码import java.util.regex.*;
class RegexExample1{
public static void main(String[] args){
  String content = "I am noob " + "from runoob.com."; 
  String pattern = ".*runoob.*"; 
  boolean isMatch = Pattern.matches(pattern, content);
  System.out.println("字符串中是否包含了 'runoob' 子字符串? " + isMatch);
  } 
}

最终打印字符串中是否包含了 ‘runoob’ 子字符串? true

正则表达式语法大全

在其他语言中，\ 表示：我想要在正则表达式中插入一个普通的（字面上的）反斜杠，请不要给它任何特殊的意义。

在 Java 中，\ 表示：我要插入一个正则表达式的反斜线，所以其后的字符具有特殊的意义。
所以，在其他的语言中（如 Perl），一个反斜杠 \ 就足以具有转义的作用，而在 Java 中正则表达式中则需要有两个反斜杠才能被解析为其他语言中的转义作用。也可以简单的理解在 Java 的正则表达式中，两个 \ 代表其他语言中的一个 \，这也就是为什么表示一位数字的正则表达式是 \d，而表示一个普通的反斜杠是 \。

1
2

less复制代码System.out.print("\");    // 输出为 \
System.out.print("\\");  // 输出为 \

注意：根据 Java Language Specification 的要求，Java 源代码的字符串中的反斜线被解释为 Unicode 转义或其他字符转义。因此必须在字符串字面值中使用两个反斜线，表示正则表达式受到保护，不被 Java 字节码编译器解释。例如，当解释为正则表达式时，字符串字面值 “\b” 与单个退格字符匹配，而 “\b” 与单词边界匹配。字符串字面值 “(hello)” 是非法的，将导致编译时错误；要与字符串 (hello) 匹配，必须使用字符串字面值 “(hello)“。

作者主页：Java李杨勇 简历模板、学习资料、面试题库、技术互助【关注我，都给你】

好了、今天就分享到这儿吧，我是小奥、下期见~~

打卡 文章 更新 80/ 100天

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

小知识，大挑战！本文正在参与「程序员必备小知识」创作活动。

本文已参与 「掘力星计划」 ，赢取创作大礼包，挑战创作激励金。

前面的文章咱们讲了 MyBatis 批量插入的 3 种方法：循环单次插入、MyBatis Plus 批量插入、MyBatis 原生批量插入，详情请点击《MyBatis 批量插入数据的 3 种方法！》。
​

但之前的文章也有不完美之处，原因在于：使用 「循环单次插入」的性能太低，使用「MyBatis Plus 批量插入」性能还行，但要额外的引入 MyBatis Plus 框架，使用「MyBatis 原生批量插入」性能最好，但在插入大量数据时会导致程序报错，那么，今天咱们就会提供一个更优的解决方案。

原生批量插入的“坑”

首先，我们来看一下 MyBatis 原生批量插入中的坑，当我们批量插入 10 万条数据时，实现代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

java复制代码import com.example.demo.model.User;
import com.example.demo.service.impl.UserServiceImpl;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.ArrayList;
import java.util.List;

@SpringBootTest
class UserControllerTest {

    // 最大循环次数
    private static final int MAXCOUNT = 100000;

    @Autowired
    private UserServiceImpl userService;
    
    /**
     * 原生自己拼接 SQL，批量插入
     */
    @Test
    void saveBatchByNative() {
        long stime = System.currentTimeMillis(); // 统计开始时间
        List<User> list = new ArrayList<>();
        for (int i = 0; i < MAXCOUNT; i++) {
            User user = new User();
            user.setName("test:" + i);
            user.setPassword("123456");
            list.add(user);
        }
        // 批量插入
        userService.saveBatchByNative(list);
        long etime = System.currentTimeMillis(); // 统计结束时间
        System.out.println("执行时间：" + (etime - stime));
    }
}

核心文件 UserMapper.xml 中的实现代码如下：

1
2
3
4
5
6
7
8
9
10
11

xml复制代码<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.example.demo.mapper.UserMapper">
    <insert id="saveBatchByNative">
        INSERT INTO `USER`(`NAME`,`PASSWORD`) VALUES
        <foreach collection="list" separator="," item="item">
            (#{item.name},#{item.password})
        </foreach>
    </insert>

</mapper>

当我们开心地运行以上程序时，就出现了以下的一幕：

沃，程序竟然报错了！
​

这是因为使用 MyBatis 原生批量插入拼接的插入 SQL 大小是 4.56M，而默认情况下 MySQL 可以执行的最大 SQL 为 4M，那么在程序执行时就会报错了。

解决方案

以上的问题就是因为批量插入时拼接的 SQL 文件太大了，所以导致 MySQL 的执行报错了。那么我们第一时间想到的解决方案就是将大文件分成 N 个小文件，这样就不会因为 SQL 太大而导致执行报错了。也就是说，我们可以将待插入的 List 集合分隔为多个小 List 来执行批量插入的操作，而这个操作过程就叫做 List 分片。

有了处理思路之后，接下来就是实践了，那如何对集合进行分片操作呢？
​

分片操作的实现方式有很多种，这个我们后文再讲，接下来我们使用最简单的方式，也就是 Google 提供的 Guava 框架来实现分片的功能。

分片 Demo 实战

要实现分片功能，第一步我们先要添加 Guava 框架的支持，在 pom.xml 中添加以下引用：

1
2
3
4
5
6
7

java复制代码<!-- google guava 工具类 -->
<!-- https://mvnrepository.com/artifact/com.google.guava/guava -->
<dependency>
  <groupId>com.google.guava</groupId>
  <artifactId>guava</artifactId>
  <version>31.0.1-jre</version>
</dependency>

接下来我们写一个小小的 demo，将以下 7 个人名分为 3 组（每组最多 3 个），实现代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

java复制代码import com.google.common.collect.Lists;

import java.util.Arrays;
import java.util.List;

/**
 * Guava 分片
 */
public class PartitionByGuavaExample {
    // 原集合
    private static final List<String> OLD_LIST = Arrays.asList(
            "唐僧,悟空,八戒,沙僧,曹操,刘备,孙权".split(","));

    public static void main(String[] args) {
        // 集合分片
        List<List<String>> newList = Lists.partition(OLD_LIST, 3);
        // 打印分片集合
        newList.forEach(i -> {
            System.out.println("集合长度：" + i.size());
        });
    }
}

以上程序的执行结果如下：

从上述结果可以看出，我们只需要使用 Guava 提供的 Lists.partition 方法就可以很轻松的将一个集合进行分片了。
​

原生批量插入分片实现

那接下来，就是改造我们的 MyBatis 批量插入代码了，具体实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

java复制代码@Test
void saveBatchByNativePartition() {
    long stime = System.currentTimeMillis(); // 统计开始时间
    List<User> list = new ArrayList<>();
    // 构建插入数据
    for (int i = 0; i < MAXCOUNT; i++) {
        User user = new User();
        user.setName("test:" + i);
        user.setPassword("123456");
        list.add(user);
    }
    // 分片批量插入
    int count = (int) Math.ceil(MAXCOUNT / 1000.0); // 分为 n 份，每份 1000 条
    List<List<User>> listPartition = Lists.partition(list, count);
    // 分片批量插入
    for (List<User> item : listPartition) {
        userService.saveBatchByNative(item);
    }
    long etime = System.currentTimeMillis(); // 统计结束时间
    System.out.println("执行时间：" + (etime - stime));
}

执行以上程序，最终的执行结果如下：

从上图可以看出，之前批量插入时的异常报错不见了，并且此实现方式的执行效率竟比 MyBatis Plus 的批量插入的执行效率要高，MyBatis Plus 批量插入 10W 条数据的执行时间如下：

总结

本文我们演示了 MyBatis 原生批量插入时的问题：可能会因为插入的数据太多从而导致运行失败，我们可以通过分片的方式来解决此问题，分片批量插入的实现步骤如下：

1. 计算出分片的数量（分为 N 批）；
2. 使用 Lists.partition 方法将集合进行分片（分为 N 个集合）；
3. 循环将分片的集合进行批量插入的操作。

关注公众号「Java中文社群」查看更多 MyBatis 和 Spring Boot 的系列文章。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

RDB和AOF

持久化流程

要有下面五个过程：

1. 客户端向服务端发送写操作(数据在客户端的内存中)。
2. 数据库服务端接收到写请求的数据(数据在服务端的内存中)。
3. 服务端调用write这个系统调用，将数据往磁盘上写(数据在系统内存的缓冲区中)。
4. 操作系统将缓冲区中的数据转移到磁盘控制器上(数据在磁盘缓存中)。
5. 磁盘控制器将数据写到磁盘的物理介质中(数据真正落到磁盘上)。

这5个过程是在理想条件下一个正常的保存流程，但是在大多数情况下，我们的机器等等都会有各种各样的故障，这里划分了两种情况：

• Redis数据库发生故障，只要在上面的第三步执行完毕，那么就可以持久化保存，剩下的两步由操作系统替我们完成。
• 操作系统发生故障，必须上面5步都完成才可以。

在这里只考虑了保存的过程可能发生的故障，其实保存的数据也有可能发生损坏，需要一定的恢复机制，不过在这里就不再延伸了。现在主要考虑的是redis如何来实现上面5个保存磁盘的步骤。它提供了两种策略机制，也就是RDB和AOF。

RDB

RDB(Redis DataBase)方式是在指定的时间间隔内将内存中数据集快照持久化到磁盘以二进制文件dump.rdb进行保存

开启RDB持久化（默认开启）

自定义配置的快照规则：

save：这里是用来配置触发 Redis的 RDB 持久化条件，也就是什么时候将内存中的数据保存到硬盘。比如”save m n”。表示m秒内数据集存在n次修改时，自动触发bgsave。

默认如下配置：

save 3600 1 ：表示3600秒钟内至少1个键被更改则进行快照。

save 300 100 ：表示300秒内至少100个键被更改则进行快照。

save 60 10000 ：表示60秒内至少10000个键被更改则进行快照。

如果不需要持久化，那么你可以注释掉所有的 save 行来停用保存功能或设置为save “” 。

RDB文件保存过程

• redis调用fork,现在有了子进程和父进程。
• 父进程继续处理client请求，子进程负责将内存内容写入到临时文件。由于os的写时复制机制（copy on write)父子进程会共享相同的物理页面，当父进程处理写请求时os会为父进程要修改的页面创建副本，而不是写共享的页面。所以子进程的地址空间内的数据是fork时刻整个数据库的一个快照。
• 当子进程将快照写入临时文件完毕后，用临时文件替换原来的快照文件，然后子进程退出。

save、bgsave、自动化

save触发方式

该命令会阻塞当前Redis服务器，执行save命令期间，Redis不能处理其他命令，直到RDB过程完成为止。具体流程如下：

执行完成时候如果存在老的RDB文件，就把新的替代掉旧的。我们的客户端可能都是几万或者是几十万，这种方式显然不可取。

save命令执行一个同步保存操作，将当前 Redis 实例的所有数据快照(snapshot)以 RDB 文件的形式保存到硬盘。

1
2

java复制代码127.0.0.1:6379> save
OK

bgsave触发方式

执行该命令时，Redis会在后台异步进行快照操作，快照同时还可以响应客户端请求。具体流程如下：

具体操作是Redis进程执行fork操作创建子进程，RDB持久化过程由子进程负责，完成后自动结束。阻塞只发生在fork阶段，一般时间很短。基本上 Redis 内部所有的RDB操作都是采用 bgsave 命令。

Lastsave命令返回最近一次 Redis 成功将数据保存到磁盘上的时间，以 UNIX 时间戳格式表示。

1
2
3
4

java复制代码127.0.0.1:6379> bgsave
Background saving started
127.0.0.1:6379> lastsave
(integer) 1632295819

自动触发方式

自动触发是由我们的配置文件来完成的。在redis.conf配置文件中，里面有如下配置，我们可以去设置：

1. save：这里是用来配置触发 Redis的 RDB 持久化条件，也就是什么时候将内存中的数据保存到硬盘。比如“save m n”。表示m秒内数据集存在n次修改时，自动触发bgsave。
2. stop-writes-on-bgsave-error ：默认值为yes。当启用了RDB且最后一次后台保存数据失败，Redis是否停止接收数据。这会让用户意识到数据没有正确持久化到磁盘上，否则没有人会注意到灾难发生了。如果Redis重启了，那么又可以重新开始接收数据了
3. rdbcompression ；默认值是yes。对于存储到磁盘中的快照，可以设置是否进行压缩存储。
4. rdbchecksum ：默认值是yes。在存储快照后，我们还可以让redis使用CRC64算法来进行数据校验，但是这样做会增加大约10%的性能消耗，如果希望获取到最大的性能提升，可以关闭此功能。
5. dbfilename ：设置快照的文件名，默认是 dump.rdb
6. dir：设置快照文件的存放路径，这个配置项一定是个目录，而不能是文件名。

save与bgsave区别

因为第三种方式是配置的，所以我们对前两种进行一个对比：

RDB优点

• RDB文件紧凑，全量备份，整个Redis数据库将只包含一个文件，非常适合用于进行备份和灾难恢复。
• 生成RDB文件的时候，redis主进程会fork()一个子进程来处理所有保存工作，主进程不需要进行任何磁盘IO操作。
• RDB 在恢复大数据集时的速度比 AOF 的恢复速度要快。

RDB缺点

• 如果你需要尽量避免在服务器故障时丢失数据，那么RDB不适合你。 虽然 Redis 允许你设置不同的保存点（save point）来控制保存 RDB 文件的频率， 但是，因为RDB文件需要保存整个数据集的状态， 所以它并不是一个轻松的操作。 因此你可能会至少5分钟才保存一次RDB文件。 在这种情况下， 一旦发生故障停机，你就可能会丢失好几分钟的数据。
• 每次保存 RDB 的时候，Redis 都要 fork() 出一个子进程，并由子进程来进行实际的持久化工作。 在数据集比较庞大时，fork()可能会非常耗时，造成服务器在某某毫秒内停止处理客户端； 如果数据集非常巨大，并且 CPU 时间非常紧张的话，那么这种停止时间甚至可能会长达整整一秒。 虽然 AOF 重写也需要进行 fork() ，但无论 AOF 重写的执行间隔有多长，数据的耐久性都不会有任何损失。

如果数据相对来说比较重要，希望将损失降到最小，则可以使用AOF方式进行持久化。

AOF

AOF(Append Only File).redis会将每一个收到的写命令都通过write函数追加到文件中。通俗的理解就是日志记录。

每当有一个写命令过来时，就直接保存在我们的AOF文件中。

开启AOF持久化

第一步：修改redis.conf文件
将

1
2
3
4
5
6
7

java复制代码appendonly no
##默认持久化文件名
appendfilename "appendonly.aof"
# appendfsync always  ##每次操作都会立即写入aof文件中
##每秒持久化一次(默认配置)
appendfsync everysec
# appendfsync no   ##不主动进行同步操作，默认30s一次

修改为

1

java复制代码appendonly yes

第二步：指定redis.conf文件启动

1
2

java复制代码[root@172 redis-6.2.5]# ./src/redis-server  /app/redis/redis-6.2.5/redis.conf
[root@172 redis-6.2.5]#

AOF文件保存过程

redis会将每一个收到的写命令都通过write函数追加到文件中(默认是 appendonly.aof)。

当redis重启时会通过重新执行文件中保存的写命令来在内存中重建整个数据库的内容。当然由于os会在内核中缓存write做的修改，所以可能不是立即写到磁盘上。这样aof方式的持久化也还是有可能会丢失部分修改。

保存策略

我们可以通过配置文件告诉redis我们想要通过fsync函数强制os写入到磁盘的时机。有三种方式如下（默认是：每秒fsync一次）

• appendfsync always ：每次收到写命令就立即强制写入磁盘，最慢的，但是保证完全的持久化，不推荐使用
• appendfsync everysec ：每秒钟强制写入磁盘一次，在性能和持久化方面做了很好的折中，推荐
• appendfsync no ：完全依赖os，性能最好,持久化没保证

aof 的方式也同时带来了另一个问题。持久化文件会变的越来越大。例如我们调用incr test命令100次，文件中必须保存全部的100条命令，其实有99条都是多余的。因为要恢复数据库的状态其实文件中保存一条set test 100就够了。

文件重写原理

AOF的方式也同时带来了另一个问题。持久化文件会变的越来越大。为了压缩aof的持久化文件。redis提供了bgrewriteaof命令。将内存中的数据以命令的方式保存到临时文件中，同时会fork出一条新进程来将文件重写。

• redis调用fork ，现在有父子两个进程
• 子进程根据内存中的数据库快照，往临时文件(新的aof文件) 中写入重建数据库状态的命令
• 父进程继续处理client请求，除了把写命令写入到原来的aof文件中。同时把收到的写命令缓存起来。这样就能保证如果子进程重写失败的话并不会出问题。
• 当子进程把快照内容写入已命令方式写到临时文件中后，子进程发信号通知父进程。然后父进程把缓存的写命令也写入到临时文件。
• 现在父进程可以使用临时文件替换老的aof文件，并重命名，后面收到的写命令也开始往新的aof文件中追加。

需要注意到是重写aof文件的操作，并没有读取旧的aof文件，而是将整个内存中的数据库内容用命令的方式重写了一个新的aof文件,这点和快照有点类似。

AOF优点

• AOF可以更好的保护数据不丢失，一般AOF会每隔1秒，通过一个后台线程执行一次fsync操作，最多丢失1秒钟的数据。
• AOF 文件是一个只进行追加操作的日志文件（append only log），没有任何磁盘寻址的开销，写入性能非常高，文件不容易破损，因此对AOF文件的写入不需要进行seek，即使日志因为某些原因而包含了未写入完整的命令（比如写入时磁盘已满，写入中途停机，等等），redis-check-aof工具也可以轻易地修复这种问题。
• Redis 可以在 AOF 文件体积变得过大时，自动地在后台对 AOF 进行重写，也不会影响客户端的读写： 重写后的新 AOF 文件包含了恢复当前数据集所需的最小命令集合。 整个重写操作是绝对安全的，因为 Redis 在创建新 AOF 文件的过程中，会继续将命令追加到现有的 AOF 文件里面，即使重写过程中发生停机，现有的 AOF 文件也不会丢失。 而一旦新 AOF 文件创建完毕，Redis 就会从旧 AOF 文件切换到新 AOF 文件，并开始对新 AOF文件进行追加操作
• AOF 文件有序地保存了对数据库执行的所有写入操作，这个特性非常适合做灾难性的误删除的紧急恢复。 这些写入操作以 Redis 协议的格式保存， 因此 AOF 文件的内容非常容易被人读懂， 对文件进行分析（parse）也很轻松。 导出（export） AOF 文件也非常简单： 举个例子， 如果你不小心执行了 FLUSHALL 命令， 但只要 AOF 文件未被重写， 那么只要停止服务器， 移除 AOF 文件末尾的 FLUSHALL 命令， 并重启 Redis ， 就可以将数据集恢复到 FLUSHALL 执行之前的状态。

AOF缺点

• 对于相同的数据集来说，AOF 文件的体积通常要大于 RDB 文件的体积。
• AOF开启后，支持的写QPS会比RDB支持的写QPS低，因为AOF一般会配置成每秒fsync一次日志文件， 在一般情况下， 每秒 fsync 的性能依然非常高， 而关闭fsync可以让 AOF 的速度和 RDB 一样快， 即使在高负荷之下也是如此。 不过在处理巨大的写入载入时，RDB 可以提供更有保证的最大延迟时间
• AOF 在过去曾经发生过这样的 bug ： 因为个别命令的原因，导致 AOF 文件在重新载入时，无法将数据集恢复成保存时的原样。 （举个例子，阻塞命令 BRPOPLPUSH 就曾经引起过这样的 bug） 测试套件里为这种情况添加了测试： 它们会自动生成随机的、复杂的数据集， 并通过重新载入这些数据来确保一切正常。 虽然这种 bug 在 AOF 文件中并不常见， 但是对比来说， RDB 几乎是不可能出现这种 bug 的。

RDB和AOF总结

• rdb持久化
  • 是快照同步方式（按照周期性进行数据的持久化）
  • 数据可能在某时间点上宕机后存在数据丢失
  • 持久化数据效率低，二进制方式恢复数据速度快
• aof持久化
  • 是增量日志同步方式（对行为进行操作持久化数据）
  • 数据也会存在某时间节点丢失数据
  • 持久化数据效率高，以命令方式恢复数据速度慢
• 选择
  • 一般来说， 如果想达到数据安全性， 你应该同时使用两种持久化功能。
  • 如果你非常关心你的数据， 但仍然可以承受数分钟以内的数据丢失， 那么你可以只使用 RDB 持久化。
  • 最多只会丢失一秒钟的数据使用AOF

redis分布式缓存系列

• redis分布式缓存（一）一一 redis安装(linux和docker)

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

小知识，大挑战！本文正在参与「程序员必备小知识」创作活动

本文已参与「掘力星计划」，赢取创作大礼包，挑战创作激励金。

点赞再看，养成习惯。微信搜索【一条coding】关注这个在互联网摸爬滚打的程序员。

本文收录于github-技术专家修炼，里面有我的学习路线、系列文章、面试题库、自学资料、电子书等。

哈喽，大家好，我是一条~

今天给大家带来排序算法的讲解。相信无论是初学者学习还是大厂面试，都少不了排序算法这关，即使没学过算法，对冒泡排序也不会陌生。

今天，一条就带大家彻底跨过「排序算法」这道坎，保姆级教程建议收藏。⭐️

本文配套源码地址：《八大排序》源码，提取码：5ehp

准备
==

古语云：“兵马未动，粮草先行”。想跟着一条一块把「排序算法」弄明白的,建议先准备好以下代码模板。

📢 观看本教程需知道基本循环语法、两数交换、双指针等前置知识。

📚 建议先看完代码和逐步分析后再尝试自己写。

• 新建一个Java工程,本文全篇也基于Java语言实现代码。
• 建立如下目录结构

• 在MainTest测试类中编写测试模板。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

java复制代码/**
 * 测试类
 * Author：一条
 * Date：2021/09/23
 */
public class MainTest {
    public static void main(String[] args) {
        //待排序序列
        int[] array={6,10,4,5,2,8};
        //调用不同排序算法
				// BubbleSort.sort(array);

        // 创建有100000个随机数据的数组
        int[] costArray=new int[100000];
        for (int i = 0; i < 100000; i++) {
            // 生成一个[0,100000) 的一个数
            costArray[i] = (int) (Math.random() * 100000);
        }

        Date start = new Date();
        //过长，先注释掉逐步打印
				//BubbleSort.sort(costArray);
        Date end = new Date();
        System.out.println("耗时："+(end.getTime()-start.getTime())/1000+"s");
    }
}

该段代码内容主要有两个功能：

• 调用不同的排序算法进行测试
• 测试不同排序算法将10w个数排好序需要的时间。更加具象的理解时间复杂度的不同

冒泡排序

基本思想

通过对乱序序列从前向后遍历,依次比较相邻元素的值，若发现逆序则交换，使值较大的元素逐渐从前移向后部。

像水底下的气泡一样逐渐向上冒一样。

动图讲解

18.17.49.gif)

代码实现

不理解的小伙伴可以用debug模式逐步分析。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

java复制代码/**
 * 冒泡排序
 * Author：一条
 * Date：2021/09/23
 */
public class BubbleSort{
    public static int[] sort(int[] array){
        for (int i = 0; i < array.length; i++) {
            for (int j = 0; j < array.length-1; j++) {
              //依次比较，将最大的元素交换到最后
                if (array[j]>array[j+1]){
                  // 用临时变量temp交换两个值
                    int temp=array[j];
                    array[j]=array[j+1];
                    array[j+1]=temp;
                }
            }
          //输出每一步的排序结果
            System.out.println(Arrays.toString(array));
        }
        return array;
    }
}

输出结果

逐步分析

1. 初始数组：[6,10,4,5,2,8]
2. 6拿出来和后一个10比较，6<10，不用交换。- > j++;
3. 10拿出来和后一个4比较,10>4,交换。- > [6,4,10,5,2,8]
4. 依次执行j++与后一个比较交换。
5. 第一层i循环完，打印第一行- > [6, 4, 5, 2, 8, 10]，此时最后一位10在正确位置上。 - > i++
6. 从4开始，继续比较交换，倒数第二位8回到正确位置。
7. 如上循环下去 - > ……
8. 最终结果 - > [2, 4, 5, 6, 8, 10]

这时再回去看动图理解。

耗时测试

记得先注释掉排序类逐步打印代码。

时间复杂度：O(n^2)

算法优化

优化点一

外层第一次遍历完，最后一位已经是正确的，j就不需要再比较，所以结束条件应改为j-i-1;。

优化点二

因为排序的过程中，各元素不断接近自己的位置，如果一趟比较下来没有进行过交换，就说明序列有序，因此要在排序过程中设置一个标志flag判断元素是否进行过交换。从而减少不必要的比较。

优化代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

java复制代码public static int[] sortPlus(int[] array){
        System.out.println("优化冒泡排序开始----------");
        for (int i = 0; i < array.length; i++) {
            boolean flag=false;
            for (int j = 0; j < array.length-i-1; j++) {
                if (array[j]>array[j+1]){
                    flag=true;
                    int temp=array[j];
                    array[j]=array[j+1];
                    array[j+1]=temp;
                }
            }
            if (flag==false){
                break;
            }
//            System.out.println(Arrays.toString(array));
        }
        return array;
    }

优化测试

通过基础测试看到当序列已经排好序，即不发生交换后终止循环。

耗时测试由27s优化到17s。

选择排序

基本思想

选择排序和冒泡排序很像，是从乱序序列的数据中，按指定的规则选出某一元素，再依规定交换位置后达到排序的目的。

动图讲解

18.52.31.gif)

代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

java复制代码public class SelectSort {
    public static int[] sort(int[] array) {
        System.out.println("选择排序开始----------");
        for (int i = 0; i < array.length; i++) {
          //每个值只需与他后面的值进行比较，所以从开始
            for (int j = i; j < array.length; j++) {
              //注意此处是哪两个值比较
                if (array[i]>array[j]){
                    int temp=array[i];
                    array[i]=array[j];
                    array[j]=temp;
                }
            }
            System.out.println(Arrays.toString(array));
        }
        return array;
    }
}

输出结果

逐步分析

• 初始数组：[6,10,4,5,2,8]
• 拿出6与10比较，不交换 - > j++
• 6与2比较，交换 - > j++
• 注意此时是拿2继续比较，都不交换，确定第一位（最小的数）为2 - > i++
• 循环下去，依次找到第一小，第二小，……的数
• 最终结果 - > [2, 4, 5, 6, 8, 10]

这时再回去看动图理解。

耗时测试

时间复杂度：O(n^2)

算法优化

上诉代码中使用交换的方式找到较小值，还可以通过移动的方式，即全部比较完只交换一次。

这种对空间的占有率会有些增益，但对时间的增益几乎没有，可忽略，亦不再演示。

插入排序

基本思想

把n个乱序的元素看成为一个有序表和一个无序表，开始时有序表中只包含一个元素，无序表中包含有n-1个元素，排序过程中通过不断往有序表插入元素，获取一个局部正确解，逐渐扩大有序序列的长度，直到完成排序。

动图讲解

19.20.05.gif)

代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

java复制代码/**
 * 插入排序
 * Author：一条
 * Date：2021/09/23
 */
public class InsertSort {
    public static void sort(int[] array) {
        for (int i = 1; i < array.length; i++) {
            //插入有序序列，且将有序序列扩大
            for (int j = i; j > 0; j--) {
                if (array[j]>array[j-1]){
                    int temp=array[j];
                    array[j]=array[j-1];
                    array[j-1]=temp;
                }
            }
//            System.out.println(Arrays.toString(array));
        }
    }
}

输出结果

耗时测试

算法优化

见下方希尔排序，就是希尔对插入排序的优化。

希尔排序

希尔排序是插入排序的一个优化，思考往[2,3,4,5,6]中插入1，需要将所有元素的位置都移动一遍，也就是说在某些极端情况下效率不高，也称该算法不稳定。

希尔排序是插入排序经过改进之后的一个更高效的版本，也称为缩小增量排序。

基本思想

希尔排序是把记录按下标的一定增量分组，对每组使用插入排序；

随着增量逐渐减少，每组包含的关键词越来越多，当增量减至1时，整个序列恰被分成一组，算法便终止。

和插入排序一样，从局部到全部，希尔排序是局部再局部。

动图讲解

代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

java复制代码/**
 * 希尔排序
 * Author：一条
 * Date：2021/09/23
 */
public class ShellSort {
    public static void sort(int[] array) {
        System.out.println("希尔排序开始--------");
        //gap初始增量=length/2  逐渐缩小：gap/2
        for (int gap = array.length/2; gap > 0 ; gap/=2) {
            //插入排序 交换法
            for (int i = gap; i < array.length ; i++) {
                int j = i;
                while(j-gap>=0 && array[j]<array[j-gap]){
                    //插入排序采用交换法
                    int temp = array[j];
                    array[j]=array[j-gap];
                    array[j-gap]=temp;
                    j-=gap;
                }
            }
            System.out.println(Arrays.toString(array));
        }
    }
}

输出结果

耗时测试

算法优化

无

快速排序

快速排序（Quicksort）是对冒泡排序的一种改进，相比冒泡排序，每次的交换都是跳跃式的。

基本思想

将要排序的数据分割成独立的两部分，其中一部分的所有数据都比另外一部分的所有数据都要小，然后再按此方法对这两部分数据分别进行快速排序，整个排序过程可以递归进行，以此达到整个数据变成有序序列。

体现出分治的思想。

动图讲解

02.49.17.gif)

代码实现

思路如下：

• 首先在这个序列中找一个数作为基准数，为了方便可以取第一个数。
• 遍历数组，将小于基准数的放置于基准数左边，大于基准数的放置于基准数右边。此处可用双指针实现。
• 此时基准值把数组分为了两半，基准值算是已归位（找到排序后的位置）。
• 利用递归算法，对分治后的子数组进行排序。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

java复制代码public class QuickSort {
    public static void sort(int[] array) {
        System.out.println("快速排序开始---------");
        mainSort(array, 0, array.length - 1);
    }

    private static void mainSort(int[] array, int left, int right) {
        if(left > right) {
            return;
        }
        //双指针
        int i=left;
        int j=right;
        //base就是基准数
        int base = array[left];
        //左边小于基准，右边大于基准
        while (i<j) {
            //先看右边，依次往左递减
            while (base<=array[j]&&i<j) {
                j--;
            }
            //再看左边，依次往右递增
            while (base>=array[i]&&i<j) {
                i++;
            }
            //交换
            int temp = array[j];
            array[j] = array[i];
            array[i] = temp;
        }
        //最后将基准为与i和j相等位置的数字交换
        array[left] = array[i];
        array[i] = base;
        System.out.println(Arrays.toString(array));
        //递归调用左半数组
        mainSort(array, left, j-1);
        //递归调用右半数组
        mainSort(array, j+1, right);
    }
}

输出结果

逐步分析

• 将6作为基准数，利用左右指针使左边的数<6，右边的数>6。
• 对左右两边递归，即左边用5作为基准数继续比较。
• 直到left > right结束递归。

耗时测试

快速排序为什么这么快？

算法优化

优化一

三数取中（median-of-three）：我们目前是拿第一个数作为基准数，对于部分有序序列，会浪费循环，可以用三数取中法优化，感性的小伙伴可自行了解。

优化二

快速排序对于长序列非常快，但对于短序列不如插入排序。可以综合使用。

堆排序

此章节对基础知识要求较高，初学者可跳过。

基本思想

堆排序是利用堆这种数据结构而设计的一种排序算法，堆排序是一种**选择排序，**它的最坏，最好，平均时间复杂度均为O(nlogn)，它也是不稳定排序。首先简单了解下堆结构。

堆

堆是具有以下性质的完全二叉树：

• 每个结点的值都大于或等于其左右孩子结点的值，称为大顶堆；
• 每个结点的值都小于或等于其左右孩子结点的值，称为小顶堆。

主要利用堆顶元素最大或最小的特性，通过不断构建大顶堆，交换堆顶和堆尾，断尾重构的方式实现排序。

动图讲解

代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

java复制代码public class HeapSort {
    public static void sort(int[] array) {
        //创建堆
        for (int i = (array.length - 1) / 2; i >= 0; i--) {
            //从第一个非叶子结点从下至上，从右至左调整结构
            adjustHeap(array, i, array.length);
        }

        //调整堆结构+交换堆顶元素与末尾元素
        for (int i = array.length - 1; i > 0; i--) {
            //将堆顶元素与末尾元素进行交换
            int temp = array[i];
            array[i] = array[0];
            array[0] = temp;

            //重新对堆进行调整
            adjustHeap(array, 0, i);
        }
    }

    /**
     * 调整堆
     * @param array 待排序列
     * @param parent 父节点
     * @param length 待排序列尾元素索引
     */
    private static void adjustHeap(int[] array, int parent, int length) {
        //将temp作为父节点
        int temp = array[parent];
        //左孩子
        int lChild = 2 * parent + 1;
        while (lChild < length) {
            //右孩子
            int rChild = lChild + 1;
            // 如果有右孩子结点，并且右孩子结点的值大于左孩子结点，则选取右孩子结点
            if (rChild < length && array[lChild] < array[rChild]) {
                lChild++;
            }
            // 如果父结点的值已经大于孩子结点的值，则直接结束
            if (temp >= array[lChild]) {
                break;
            }
            // 把孩子结点的值赋给父结点
            array[parent] = array[lChild];
            //选取孩子结点的左孩子结点,继续向下筛选
            parent = lChild;
            lChild = 2 * lChild + 1;
        }
        array[parent] = temp;
        System.out.println(Arrays.toString(array));
    }
}

输出结果

逐步分析

1. 构建初始堆，将待排序列构成一个大顶堆(或者小顶堆)，升序大顶堆，降序小顶堆；
2. 将堆顶元素与堆尾元素交换，并断开(从待排序列中移除)堆尾元素。
3. 重新构建堆。
4. 重复2~3，直到待排序列中只剩下一个元素(堆顶元素)。

耗时测试

算法优化

优化点关键就在于我们以什么手法找到顶部元素该有的位置，有兴趣同学可以继续研究。

归并排序

基本思想

归并排序（MERGE-SORT）是利用归并的思想实现的排序方法，采用经典的分治（divide-and-conquer）策略。

将乱序序列不断的分成一半，排好序再拼回去，用递归实现。

难点在于如何归并两个排好序的数组？

我们可以开辟一个临时数组，使用快慢指针来辅助我们的归并。

虽然需要额外空间的来完成这个排序。但是现在计算机的内存都比较大，时间的效率要比空间的效率重要的多。

所以空间换时间也是优化算法时的一个方向。

动图讲解

15.20.57.gif)

代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

java复制代码public class MergeSort {
    public static void sort(int[] array){
        divide(array,0,array.length-1);
    }

    private static int[] divide(int[] array, int left, int right) {
        int mid = (left+right)/2;
        if(left<right){
            divide(array,left,mid);
            divide(array,mid+1,right);
            //左右归并
            merge(array,left,mid,right);
        }
        System.out.println(Arrays.toString(array));
        return array;
    }

    public static void merge(int[] array, int left, int mid, int right) {
        int[] temp = new int[right-left+1];
        int i= left;
        int j = mid+1;
        int k=0;
        // 把较小的数先移到新数组中
        while(i<=mid && j<=right){
            if(array[i]<array[j]){
                temp[k++] = array[i++];
            }else{
                temp[k++] = array[j++];
            }
        }
        // 把左边剩余的数移入数组
        while(i<=mid){
            temp[k++] = array[i++];
        }
        // 把右边边剩余的数移入数组
        while(j<=right){
            temp[k++] = array[j++];
        }
        // 把新数组中的数覆盖nums数组
        for(int x=0;x<temp.length;x++){
            array[x+left] = temp[x];
        }
    }
}

输出结果

耗时测试

算法优化

计数排序

从这里开始都是非比较排序。

基本思想

假如输入一个数x，如果我们可以找到比该数小的数有几个，那么就可以直接将x放入到对应的输出数组的位置。
比如测试序列中的x=5，，比5小的有2个，那么毫无疑问5就该排在第三位。

动图讲解

16.05.17.gif)

代码实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

java复制代码public class CountSort {
    public static void sort(int[] array) {
        System.out.println("计数排序开始--------");
        //计算最大值和最小值，用于确定count[]的长度
        int max = array[0], min = array[0];
        for (int i : array) {
            if (i > max) {
                max = i;
            }
            if (i < min) {
                min = i;
            }
        }
        //count[]用于存储每个元素出现的次数
        int[] count = new int[max - min + 1];

        //统计出现的频次
        for (int i : array) {
            count[i - min] += 1;//数的位置 上+1
        }

        //创建最终返回的数组，和原始数组长度相等，但是排序完成的
        int[] result = new int[array.length];
        int index = 0;//记录最终数组的下标

        //先循环每一个元素  在计数排序器的下标中
        for (int i = 0; i < count.length; i++) {
            //遍历循环出现的次数
            for (int j = 0; j < count[i]; j++) {//count[i]:这个数出现的频率
                result[index++] = i + min;//以为原来减少了min现在加上min，值就变成了原来的值
            }
            System.out.println(Arrays.toString(result));
        }
    }
}

输出结果

逐步分析

• 就是将原始数组中的数值出现的频率（次数）记录在新数组下标中。
• 遍历出现的次数，依次放入新数组。

耗时测试

说实话，这个速度都惊到我了。计数排序的时间复杂度是O(n),缺点是限制值域为[0,k]的整数。

算法优化

正常计数排序是从0开始的，本文实现的代码从min开始，已优化。

总结

本文并没有具体介绍时间复杂度，因为我放一个数字在这里你根本理解不了他们的差距。

用100000长度的数组测试八大排序算法，化抽象为具体。当数据量很大的时候 nlogn的优势将会比n^2越来越大，当n=10^5的时候，nlogn的算法要比n^2的算法快6000倍，什么概念呢，就是如果我们要处理一个数据集，用nlogn的算法要处理一天的话，用n^2的算法将要处理6020天。这就基本相当于是15年。

一个优化改进的算法可能比一个比一个笨的算法速度快了许多，这就是大厂考查算法和我们学习算法的原因。

古语云：乘众人之智，则无不任也；用众人之力，则无不胜也。为了攻克算法这座大山，我制定了组队刷题计划，每天至少1道leetcode算法题。

驽马十驾，功在不舍。

如果你刚刚大一，每天坚持学习，你将会至少比别人多刷1200道题，那么毕业时你的工资就可能是别人的3-4倍。

如果你是职场人，每天提升自己，升职加薪，成为技术专家指日可待。

只要你愿意去奋斗，始终走在拼搏的路上，那你的人生，最坏的结果，也不过是大器晚成。

点此加入计划

如果链接被屏蔽，或者有权限问题，可以私聊作者解决。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

小知识，大挑战！本文正在参与“程序员必备小知识”创作活动。

Doris

优点

1.同时支持 高并发点查询和高吞吐的Ad-hoc查询

2.同时支持 批量导入和近实时mini-batch导入

3.同时支持 明细和聚合查询

4.兼容Mysql协议和标准SQL

5.支持Rollup Table 和Rollup Table的智能查询路由

6.支持多表Join和灵活的表达式查询

7.支持Schema 在线变更

8.支持和Hash和Range二级分区

缺点

1.Doris 目前不支持如 update 或 insert 等操作单条数据的 DML 语句

2.不支持实时导入

3.不支持精确去重的物化索引

4.RollUP表需要逐个手动建立

使用场景

1.Doris的聚合模型主要用于固定模式的报表类查询场景，实现原理和Mesa完全一致

2.无精确去重的OLAP查询

3.百毫秒的高性能OLAP查询

4.大结果集查询

5.明细查询

kylin

优点

1. 支持标准SQL，提供JDBC/ODBC接口
2. 通过预计算Cube显著降低查询时的计算量，数据集越大效果越明显
3. 支持精确去重计数，并且由于预计算，查询去重指标的速度很快
4. 可以支持比较高的查询并发

缺点

1. 需要学习Cube的定义和优化，学习成本较高
2. SQL需要具有一定的结构，不支持AdHoc查询
3. HBase没有二级索引，过滤的性能稍逊色
4. 支持的维度数量不宜过多（一般不超过20个），否则Cube的计算和存储开销会明显增加

使用场景

1.固化查询：指标提取、多维分析、dashboard等

2.查询模式比较固定、SQL表达

3.数据规模大、指标数量多、去重指标要求精确

4.查询并发度高对响应时间要求比较严苛

Doris相比Kylin的优势：

1、Doris中比较独特的聚合函数是Replace函数，这个聚合函数能够保证相同Keys的记录只保留最新的Value，可以借助这个Replace函数来实现点更新即支持update操作 ； kylin 不支持update操作；

2、Doris同时支持高并发点查询和高吞吐的Ad-hoc查询；Kylin 不支持Ad-hoc查询；

3、Doris 支持明细查询；Kylin不支持指标的明细查询；

4、Doris 兼容mysql协议；Kylin不兼容mysql协议；

5、Doris中和Kylin的Cuboid等价的概念是RollUp表，但是Doris的RollUp表定义更灵活，维度列和指标列可以自由选择；而Kylin只可以选择维度列，每个cuboid都必须包含所有指标列；

Presto

优点

• Presto 与 Hive 对比， 能够处理PB级别的海量数据分析，但是Presto是基于内存运算的，减少没必要的硬盘IO， 所以更快
• 能够连接多个数据源， 跨数据源连表查
• 效率10倍

缺点

稳定性差

Presto并不是把PB级别的数据都放在内存中计算。根据场景，对于聚合操作：count、avg等，是边读数据边计算再清内存再读数据再计算。对于连表查，可能会产生大量的临时数据，速度会变慢

使用场景

支持在线数据查询，包括hive、mysql等。支持跨数据源查询。

Presto主要用来处理响应时间小于1秒到几分钟的场景

ClickHouse

优点

ClickHouse最大的特点就是快，快，快，重要的话说三遍！

与Hadoop、Spark这些巨无霸组件相比，ClickHouse很轻量级，其特点：

• 列式存储数据库，数据压缩
• 关系型、支持SQL
• 分布式并行计算，把单机性能压榨到极限
• 高可用
• 数据量级在PB级别
• 实时数据更新
• 索引

缺点

• 缺少高频率，低延迟的修改或删除已存在数据的能力。仅能用于批量删除或修改数据。
• 没有完整的事务支持
• 不支持二级索引
• 有限的SQL支持，join实现与众不同
• 不支持窗口功能
• 元数据管理需要人工干预维护

使用场景

ClickHouse在单表查询性能上独领风骚，远超过其他的OLAP数据库；

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

小知识，大挑战！本文正在参与“程序员必备小知识”创作活动。

详细介绍了Java线程的6中状态，以及状态之间的转换。

本文基于JDK1.8。

1 线程状态(生命周期)

1.1 源码中的状态

关于Java线程的状态，网上说法很多，有五种、六种甚至七种，本文采用Java官方的线程状态分类。

实际上，官方一共给出了六种线程状态，我们来看看源码便可知：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

java复制代码    public class Thread implements Runnable {
        //线程的状态以枚举的方式定义在Thread类的内部
        /**
         * A thread state.  A thread can be in one of the following states:
         * <ul>
         * <li>{@link #NEW}<br>
         *     A thread that has not yet started is in this state.
         *     </li>
         * <li>{@link #RUNNABLE}<br>
         *     A thread executing in the Java virtual machine is in this state.
         *     </li>
         * <li>{@link #BLOCKED}<br>
         *     A thread that is blocked waiting for a monitor lock
         *     is in this state.
         *     </li>
         * <li>{@link #WAITING}<br>
         *     A thread that is waiting indefinitely for another thread to
         *     perform a particular action is in this state.
         *     </li>
         * <li>{@link #TIMED_WAITING}<br>
         *     A thread that is waiting for another thread to perform an action
         *     for up to a specified waiting time is in this state.
         *     </li>
         * <li>{@link #TERMINATED}<br>
         *     A thread that has exited is in this state.
         *     </li>
         * </ul>
         *
         * <p>
         * A thread can be in only one state at a given point in time.
         * These states are virtual machine states which do not reflect
         * any operating system thread states.
         *
         * @since   1.5
         * @see #getState
         */
        public enum State {
            /**
             * Thread state for a thread which has not yet started.
             */
            NEW,

            /**
             * Thread state for a runnable thread.  A thread in the runnable
             * state is executing in the Java virtual machine but it may
             * be waiting for other resources from the operating system
             * such as processor.
             */
            RUNNABLE,

            /**
             * Thread state for a thread blocked waiting for a monitor lock.
             * A thread in the blocked state is waiting for a monitor lock
             * to enter a synchronized block/method or
             * reenter a synchronized block/method after calling
             * {@link Object#wait() Object.wait}.
             */
            BLOCKED,

            /**
             * Thread state for a waiting thread.
             * A thread is in the waiting state due to calling one of the
             * following methods:
             * <ul>
             *   <li>{@link Object#wait() Object.wait} with no timeout</li>
             *   <li>{@link #join() Thread.join} with no timeout</li>
             *   <li>{@link LockSupport#park() LockSupport.park}</li>
             * </ul>
             *
             * <p>A thread in the waiting state is waiting for another thread to
             * perform a particular action.
             * <p>
             * For example, a thread that has called <tt>Object.wait()</tt>
             * on an object is waiting for another thread to call
             * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
             * that object. A thread that has called <tt>Thread.join()</tt>
             * is waiting for a specified thread to terminate.
             */
            WAITING,

            /**
             * Thread state for a waiting thread with a specified waiting time.
             * A thread is in the timed waiting state due to calling one of
             * the following methods with a specified positive waiting time:
             * <ul>
             *   <li>{@link #sleep Thread.sleep}</li>
             *   <li>{@link Object#wait(long) Object.wait} with timeout</li>
             *   <li>{@link #join(long) Thread.join} with timeout</li>
             *   <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
             *   <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
             * </ul>
             */
            TIMED_WAITING,

            /**
             * Thread state for a terminated thread.
             * The thread has completed execution.
             */
            TERMINATED;
        }
    }

从源码可以看出，Java语言定义了6种线程状态，作为内部枚举保存在Thread类中。这里我建议大家采用这六种状态，当然也可以按照自己的理解来。下面来解释一下每一种状态。

1.2 状态解释

在任意一个时间点，一个线程只能有且只有其中的一种状态，这6种状态分别如下：

1. 新建（NEW）：创建后尚未启动的线程处于这种状态。
2. 运行（RUNNABLE）：调用start()方法,RUNNABLE包括了操作系统线程状态中的Running和Ready，也就是处于此状态的线程有可能正在执行，也有可能正在等待着CPU为它分配执行时间(该线程已经获取了除CPU资源外的其他资源，等待获取CPU 资源后才会真正处于运行状态)。

官方为什么不将这两种状态分开呢？有种可能是：线程在这Running和Ready两种状态之间的时长太短了，现代cpu采用轮询式时间片调度，大部分线程Running和Ready的时间都非常短暂，因此考虑将这两种状态合并为RUNNABLE状态。

3. 无限期等待（WAITING）：处于这种状态的线程不会被分配CPU执行时间，它们要等待被其他线程显式地唤醒。以下方法会让线程陷入无限期的等待状态：

1. 没有设置Timeout参数的Object.wait（）方法。
2. 没有设置Timeout参数的Thread.join（）方法。
3. LockSupport.park（）方法。

4. 限期等待（TIMED_WAITING）：处于这种状态的线程也不会被分配CPU执行时间，不过无须等待被其他线程显式地唤醒，在一定时间之后它们会由系统自动唤醒。以下方法会让线程进入限期等待状态：

1. Thread.sleep(long millis)方法。
2. 设置了Timeout参数的Object.wait（）方法。
3. 设置了Timeout参数的Thread.join（）方法。
4. LockSupport.parkNanos（）方法。
5. LockSupport.parkUntil（）方法。

5. 阻塞（BLOCKED）：线程被阻塞了，“阻塞状态”与“等待状态”的区别是：“阻塞状态”在等待着获取到一个排他锁，这个事件将在另外一个线程获得锁的时候可能发生，比如synchronized之外；而“等待状态”则是在获得锁之后，主动释放锁，进入等待一段时间，或者等待唤醒动作的发生。
6. 结束（TERMINATED）：已终止线程的线程状态，线程已经结束执行。

补充：
Java将操作系统中的运行和就绪两个状态合并称为运行状态。

阻塞状态是线程阻塞在进入synchronized关键字修饰的方法或代码块（获取锁）时的状态，但是阻塞在java.concurrent包中Lock接口的线程状态却是等待状态，因为java.concurrent包中Lock接口对于阻塞的实现均使用了LockSupport类中的相关方法。

2 线程状态转换

上述6种状态在遇到特定事件发生的时候将会互相转换，它们的转换关系如下图：

上图状态的转换和方法已经很明朗了，下面重点说说几种状态转换，以及相关方法补充。

2.1 进入等待/超时等待

2.1.1 进入等待状态

1. LockSupport.park()。
2. 发生Io阻塞。
3. suspend()，方法已过时。
4. public final void wait() 释放锁
5. public final void join() 释放锁

2.1.1.1 wait方法的介绍

wait方法属于object类，wait()方法使当前线程暂停执行并释放锁，让其他线程可以进入synchronized数据块，当前线程被放入对象等待队列中。Wait()方法必须被包含在对应的synchronized语句中，无论是wait()方法还是notify()方法都需要获取目标对象的一个监视器。

当调用notify()方法后，将从对象的等待队列中移走一个任意的线程并放到锁标志等待池中，只有锁标志等待池中线程才可能够获取锁标志；如果等待队列中没有线程，则notify()不起作用。notifyAll()则从对象等待池中移走所有等待那个对象的线程并放到锁标志等待池中。

被唤醒的线程并不会立即执行而是尝试获得锁,执行唤醒方法的线程也并不会立即释放锁。

2.1.1.2 join方法的介绍

join方法在内部使用wait()方法进行等待，底层用wait()来实现，可以释放锁。join方法上加了synchronuzed关键字，因此使用wait没有问题。

join方法的主要作用就是同步，它可以使得线程之间的并行执行变为串行执行，有些类似于同步的运行效果。在A线程中调用了B线程的join()方法时，表示只有当B线程执行完毕时，A线程才能继续执行。如果有多个线程,除了A线程之外的其他线程正常竞争cpu和锁。

join方法中如果传入参数，则表示这样的意思：如果A线程中调用B线程的join(10)，则表示A线程会等待B线程执行10毫秒，10毫秒过后，A、B线程并行执行。需要注意的是，jdk规定，join(0)的意思不是A线程等待B线程0秒，而是A线程等待B线程无限时间，直到B线程执行完毕，即join(0)等价于join()。(其实join()中调用的是join(0))。主线程中调用join，则主线程等待， 其他多个线程之间并不需要互相等待。

join()与synchronized的区别是：join在内部调用wait()方法进行等待，而synchronized关键字使用的是”对象监视器”原理作为同步。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

java复制代码    /**
     * join方法
     * @throws InterruptedException
     */
    public final void join() throws InterruptedException {
        //内部调用join(long millis)方法,参数为0 ,表示无限等待
        join(0);
    }

    /**
     * 等待millis毫秒
     * @param millis
     * @throws InterruptedException
     */
    public final synchronized void join(long millis)
            throws InterruptedException {
        long base = System.currentTimeMillis();
        long now = 0;

        if (millis < 0) {
            throw new IllegalArgumentException("timeout value is negative");
        }
        //millis等于0 表示无限期等待
        if (millis == 0) {
            while (isAlive()) {
                //调用wait方法
                wait(0);
            }
        } 
        //否则,限时等待
        else {
            while (isAlive()) {
                long delay = millis - now;
                if (delay <= 0) {
                    break;
                }
                //调用wait方法
                wait(delay);
                now = System.currentTimeMillis() - base;
            }
        }
    }

2.1.2 进入超时等待

1. public final void join(long millis) –超时等待 释放锁
2. LockSupport.parkNanos
3. LockSupport.parkUntil
4. wait(long timeout); –超时等待 释放锁
5. public static void sleep(long millis); —超时等待 不释放锁

2.1.2.1 sleep方法的介绍

sleep方法使当前线程（即调用该方法的线程）暂停执行一段时间，让其他线程有机会继续执行，但它并不释放对象锁。也就是说如果有synchronized同步快，其他线程仍然不能访问共享数据。注意该方法要捕捉异常。sleep会让其他所有线程都有同等的cpu争夺权力！

睡眠时被中断，则会在sleep()处抛出InterruptedException 异常。

在调用Thread.sleep(long millis）时为millis 参数传递了一个负数， 则会抛出IllegalArgumentException 异常.

2.1.2.2 LockSupport类简介

LockSupport是一个线程阻塞工具类，所有的方法都是静态方法，可以让线程在任意位置阻塞，当然阻塞之后肯定得有唤醒的方法。

常用方法：

park不需要获取某个对象的锁。因为中断的时候park不会抛出InterruptedException异常，所以需要在park之后自行判断中断状态，然后做额外的处理。

2.1.2.3 过期的suspend和resume方法

suspend方法被不推荐使用。不推荐使用suspend()去挂起线程的原因，是因为suspend()在导致线程暂停的同时，并不会去释放任何锁资源。其他线程都无法访问被它占用的锁。直到对应的线程执行resume()方法后，被挂起的线程才能继续，从而其它被阻塞在这个锁的线程才可以继续执行。如果resume()操作出现在suspend()之前执行，那么线程将一直处于挂起状态，同时一直占用锁，这就容易产生死锁。

而且，对于被挂起的线程，它的线程状态居然还是RUNNABLE

1
2
3
4
5
6
7
8

java复制代码        //创建子线程,在其内部调用suspend让其"阻塞"
        Thread thread = new Thread(() -> Thread.currentThread().suspend());
        thread.start();
        //主线程睡眠三秒,让子线程充分运行
        Thread.currentThread().sleep(3000);
        //获取子线程状态,发现还是RUNNABLE状态
        Thread.State state = thread.getState();
        System.out.println(state);

2.4 进入RUNNABLE状态

1. TIMED_WAITING状态结束
2. WAITING状态被唤醒
3. BLOCKED状态获得锁
4. public static void yield();
5. 线程的cpu时间片使用完毕还未执行完任务，重回就绪态，但此时不会释放锁。

2.4.1 yield方法的介绍

yield方法又称为“线程礼让”，使调用线程释放cpu执行权，让自己和其他多个线程重新争夺cpu执行权。

线程直接回到RUNNABLE状态，而不是让线程处于阻塞态，因此也有可能是当前让步的线程又进入到“运行状态”继续运行。

yield会尽量让同等级和高等级的线程具有更大的争夺权，而sleep会让所有线程都有同等的争夺权力，但它们并不是绝对的。毕竟java线程最终是调用操作系统的资源生成的，充满了不确定性。

yield方法不会释放持有已获得的锁，只是释放cpu的执行权。

如果有什么不懂或者需要交流，各位可以留言。另外，希望收藏、关注一下，我将不间断更新Java各种教程！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

虽然说Linux在安全方面确实相当于windows要更加可靠一些，但一般使用其作为服务器的我们，也不能大意，也是需要严格限制网络传输过程中的出入规则。上篇文章我们有聊到统计网络的信息，这篇文章来学习一下比较著名的防火墙iptables,它已经有十几年的历史了，算是不折不扣的Linux系统的功臣。

一、命令介绍

iptables 命令可以制定一些规则，规定其它电脑可以使用哪些端口来连接你的电脑（对应入），以及你的电脑可以连接哪些端口（对应出）,如果粒度更细，甚至也可以通过 IP 地址来进行过滤。

比较常见的操作有，拦截Mysql的出入网络，那么可以用 iptables 封锁 3306 端口等。一般来说系统都会默认安装iptables,如果没有安装，安装下面的指令完成安装。

1
2
3
4
5

bash复制代码# in Centos
$sudo yum install iptables

#in Ubuntu/Debian
$sudo apt install iptables

在接下来的实际操作时，需要注意的是，使用这个命令，需要使用到root权限。

二、操作实例

iptables -L
显示所有的防火墙规则。

1

bash复制代码$iptables -L

图上可以清晰的看到，对于我的服务器，明显防火墙规则分了好几个区域，包括：

• Chain INPUT : 对应控制“进入”的网络传输的规则
• Chain FORWARD : 对应控制“转发”的网络传输的规则
• Chain OUTPUT : 对应控制“出去”的网络传输的规则
• Chain DOCKER : 对应控制Docker的网络传输的规则,包括其它以Docker开头的块区域，都是Docker具体细分的规则，比如用户，隔离等等。

这个命令让我们认识到了目前的防火墙过滤的规则，那么我们CRUD这些规则呢？

清除所有的规则

1
2
3

bash复制代码$iptables -F 
$iptables -X 
$iptables -Z

屏蔽IP

1
2
3
4

bash复制代码# 屏蔽单个 IP
$iptables -I INPUT -s 121.45.6.7 -j DROP
# 封整个段
$iptables -I INPUT -s 121.0.0.0/8 -j DROP

删除指定规则

1
2
3
4

bash复制代码# 将所有 iptables 以序号标记显示
$iptables -L -n --line-numbers
#执行删除 INPUT 里序号为 10 的规则
$iptables -D INPUT 10

其功能很强大，可以阅读官方语法深入操作。

三、扩展

在使用过程中，iptables 的配置和使用规则相当繁复，普通用户学习成本上是相当的高。另外一个命令可以帮助到我们，那就是UFW（Uncomplicated Firewall）,顾名思义，就是简单的防火墙。很多Linux的发行版本并不支持，可以去官方UFW去学习使用。 进一步说，如果你的系统支持图形界面，建议使用GUFW。

iptables已经有这么多年的历史了，从 Linux 3.13 开始，官方实际上已经推出了新的命令nftables以取代前者，这个新命令是新的防火墙子系统 / 包过滤引擎,提供了一个更简单的 Kernel ABI (Application Binary Interface)，减少重复代码，改进错误报告，更有效支持过滤规则。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

一、背景

在现网环境，一些使用Redis集群的业务随着业务量的上涨，往往需要进行节点扩容操作。

之前有了解到运维同学对一些节点数比较大的Redis集群进行扩容操作后，业务侧反映集群性能下降，具体表现在访问时延增长明显。

某些业务对Redis集群访问时延比较敏感，例如现网环境对模型实时读取，或者一些业务依赖读取Redis集群的同步流程，会影响业务的实时流程时延。业务侧可能无法接受。

为了找到这个问题的根因，我们对某一次的Redis集群迁移操作后的集群性能下降问题进行排查。

1.1 问题描述

这一次具体的Redis集群问题的场景是：某一个Redis集群进行过扩容操作。业务侧使用Hiredis-vip进行Redis集群访问，进行MGET操作。

业务侧感知到访问Redis集群的时延变高。

1.2 现网环境说明

• 目前现网环境部署的Redis版本多数是3.x或者4.x版本;
• 业务访问Redis集群的客户端品类繁多，较多的使用Jedis。本次问题排查的业务使用客户端Hiredis-vip进行访问;
• Redis集群的节点数比较大，规模是100+;
• 集群之前存在扩容操作。

1.3 观察现象

因为时延变高，我们从几个方面进行排查：

• 带宽是否打满；
• CPU是否占用过高；
• OPS是否很高；

通过简单的监控排查，带宽负载不高。但是发现CPU表现异常：

1.3.1 对比OPS和CPU负载

观察业务反馈使用的MGET和CPU负载，我们找到了对应的监控曲线。

从时间上分析，MGET和CPU负载高并没有直接关联。业务侧反馈的是MGET的时延普遍增高。此处看到MGET的OPS和CPU负载是错峰的。

此处可以暂时确定业务请求和CPU负载暂时没有直接关系，但是从曲线上可以看出：在同一个时间轴上，业务请求和cpu负载存在错峰的情况，两者间应该有间接关系。

1.3.2 对比Cluster指令OPS和CPU负载

由于之前有运维侧同事有反馈集群进行过扩容操作，必然存在slot的迁移。

考虑到业务的客户端一般都会使用缓存存放Redis集群的slot拓扑信息，因此怀疑Cluster指令会和CPU负载存在一定联系。

我们找到了当中确实有一些联系：

此处可以明显看到：某个实例在执行Cluster指令的时候，CPU的使用会明显上涨。

根据上述现象，大致可以进行一个简单的聚焦：

• 业务侧执行MGET，因为一些原因执行了Cluster指令；
• Cluster指令因为一些原因导致CPU占用较高影响其他操作；
• 怀疑Cluster指令是性能瓶颈。

同时，引申几个需要关注的问题：

为什么会有较多的Cluster指令被执行？

为什么Cluster指令执行的时候CPU资源比较高？

为什么节点规模大的集群迁移slot操作容易“中招”？

二、问题排查

2.1 Redis热点排查

我们对一台现场出现了CPU负载高的Redis实例使用perf top进行简单的分析：

从上图可以看出来，函数（ClusterReplyMultiBulkSlots）占用的CPU资源高达 51.84%，存在异常。

2.1.1 ClusterReplyMultiBulkSlots实现原理

我们对clusterReplyMultiBulkSlots函数进行分析：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

java复制代码void clusterReplyMultiBulkSlots(client *c) {
    /* Format: 1) 1) start slot
     *            2) end slot
     *            3) 1) master IP
     *               2) master port
     *               3) node ID
     *            4) 1) replica IP
     *               2) replica port
     *               3) node ID
     *           ... continued until done
     */
 
    int num_masters = 0;
    void *slot_replylen = addDeferredMultiBulkLength(c);
 
    dictEntry *de;
    dictIterator *di = dictGetSafeIterator(server.cluster->nodes);
    while((de = dictNext(di)) != NULL) {
        /*注意：此处是对当前Redis节点记录的集群所有主节点都进行了遍历*/
        clusterNode *node = dictGetVal(de);
        int j = 0, start = -1;
 
        /* Skip slaves (that are iterated when producing the output of their
         * master) and  masters not serving any slot. */
        /*跳过备节点。备节点的信息会从主节点侧获取。*/
        if (!nodeIsMaster(node) || node->numslots == 0) continue;
        for (j = 0; j < CLUSTER_SLOTS; j++) {
            /*注意：此处是对当前节点中记录的所有slot进行了遍历*/
            int bit, i;
            /*确认当前节点是不是占有循环终端的slot*/
            if ((bit = clusterNodeGetSlotBit(node,j)) != 0) {
                if (start == -1) start = j;
            }
            /*简单分析，此处的逻辑大概就是找出连续的区间，是的话放到返回中；不是的话继续往下递归slot。
              如果是开始的话，开始一个连续区间，直到和当前的不连续。*/
            if (start != -1 && (!bit || j == CLUSTER_SLOTS-1)) {
                int nested_elements = 3; /* slots (2) + master addr (1). */
                void *nested_replylen = addDeferredMultiBulkLength(c);
 
                if (bit && j == CLUSTER_SLOTS-1) j++;
 
                /* If slot exists in output map, add to it's list.
                 * else, create a new output map for this slot */
                if (start == j-1) {
                    addReplyLongLong(c, start); /* only one slot; low==high */
                    addReplyLongLong(c, start);
                } else {
                    addReplyLongLong(c, start); /* low */
                    addReplyLongLong(c, j-1);   /* high */
                }
                start = -1;
 
                /* First node reply position is always the master */
                addReplyMultiBulkLen(c, 3);
                addReplyBulkCString(c, node->ip);
                addReplyLongLong(c, node->port);
                addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN);
 
                /* Remaining nodes in reply are replicas for slot range */
                for (i = 0; i < node->numslaves; i++) {
                    /*注意：此处遍历了节点下面的备节点信息，用于返回*/
                    /* This loop is copy/pasted from clusterGenNodeDescription()
                     * with modifications for per-slot node aggregation */
                    if (nodeFailed(node->slaves[i])) continue;
                    addReplyMultiBulkLen(c, 3);
                    addReplyBulkCString(c, node->slaves[i]->ip);
                    addReplyLongLong(c, node->slaves[i]->port);
                    addReplyBulkCBuffer(c, node->slaves[i]->name, CLUSTER_NAMELEN);
                    nested_elements++;
                }
                setDeferredMultiBulkLength(c, nested_replylen, nested_elements);
                num_masters++;
            }
        }
    }
    dictReleaseIterator(di);
    setDeferredMultiBulkLength(c, slot_replylen, num_masters);
}
 
/* Return the slot bit from the cluster node structure. */
/*该函数用于判断指定的slot是否属于当前clusterNodes节点*/
int clusterNodeGetSlotBit(clusterNode *n, int slot) {
    return bitmapTestBit(n->slots,slot);
}
 
/* Test bit 'pos' in a generic bitmap. Return 1 if the bit is set,
 * otherwise 0. */
/*此处流程用于判断指定的的位置在bitmap上是否为1*/
int bitmapTestBit(unsigned char *bitmap, int pos) {
    off_t byte = pos/8;
    int bit = pos&7;
    return (bitmap[byte] & (1<<bit)) != 0;
}
typedef struct clusterNode {
    ...
    /*使用一个长度为CLUSTER_SLOTS/8的char数组对当前分配的slot进行记录*/
    unsigned char slots[CLUSTER_SLOTS/8]; /* slots handled by this node */
    ...
} clusterNode;

每一个节点（ClusterNode）使用位图（char slots[CLUSTER_SLOTS/8]）存放slot的分配信息。

简要说一下BitmapTestBit的逻辑：clusterNode->slots是一个长度为CLUSTER_SLOTS/8的数组。CLUSTER_SLOTS是固定值16384。数组上的每一个位分别代表一个slot。此处的bitmap数组下标则是0到2047，slot的范围是0到16383。

因为要判断pos这个位置的bit上是否是1，因此：

• off_t byte = pos/8：拿到在bitmap上对应的哪一个字节（Byte）上存放这个pos位置的信息。因为一个Byte有8个bit。使用pos/8可以指导需要找的Byte在哪一个。此处把bitmap当成数组处理，这里对应的便是对应下标的Byte。
• int bit = pos&7：拿到是在这个字节上对应哪一个bit表示这个pos位置的信息。&7其实就是%8。可以想象对pos每8个一组进行分组，最后一组（不满足8)的个数对应的便是在bitmap对应的Byte上对应的bit数组下标位置。
• (bitmap[byte] & (1<<bit))：判断对应的那个bit在bitmap[byte]上是否存在。

以slot为10001进行举例：

因此10001这个slot对应的是下标1250的Byte,要校验的是下标1的bit。

对应在ClusterNode->slots上的对应位置：

图示绿色的方块表示bitmap[1250]，也就是对应存放slot 10001的Byte；红框标识（bit[1]）对应的就是1<<bit 的位置。bitmap[byte] & (1<<bit)，也就是确认红框对应的位置是否是1。是的话表示bitmap上10001已经打标。

总结ClusterNodeGetSlotBit的概要逻辑是：判断当前的这个slot是否分配在当前node上。因此ClusterReplyMultiBulkSlots大概逻辑表示如下：

大概步骤如下：

• 对每一个节点进行遍历；
• 对于每一个节点，遍历所有的slots，使用ClusterNodeGetSlotBit判断遍历中的slot是否分配于当前节点；

从获取CLUSTER SLOTS指令的结果来看，可以看到，复杂度是<集群主节点个数> *<slot总个数>。其中slot的总个数是16384，固定值。

2.1.2 Redis热点排查总结

就目前来看，CLUSTER SLOTS指令时延随着Redis集群的主节点个数，线性增长。而这次我们排查的集群主节点数比较大，可以解释这次排查的现网现象中CLUSTER SLOTS指令时延为何较大。

2.2 客户端排查

了解到运维同学们存在扩容操作，扩容完成后必然涉及到一些key在访问的时候存在MOVED的错误。

当前使用的Hiredis-vip客户端代码进行简单的浏览，简要分析以下当前业务使用的Hiredis-vip客户端在遇到MOVED的时候会怎样处理。由于其他的大部分业务常用的Jedis客户端，此处也对Jedis客户端对应流程进行简单分析。

2.2.1 Hiredis-vip对MOVED处理实现原理

Hiredis-vip针对MOVED的操作：

查看Cluster_update_route的调用过程：

此处的cluster_update_route_by_addr进行了CLUSTER SLOT操作。可以看到，当获取到MOVED报错的时候，Hiredis-vip会重新更新Redis集群拓扑结构，有下面的特性：

• 因为节点通过ip:port作为key，哈希方式一样，如果集群拓扑类似，多个客户端很容易同时到同一个节点进行访问；
• 如果某个节点访问失败，会通过迭代器找下一个节点，由于上述的原因，多个客户端很容易同时到下一个节点进行访问。

2.2.2 Jedis对MOVED处理实现原理

对Jedis客户端代码进行简单浏览，发现如果存在MOVED错误，会调用renewSlotCache。

继续看renewSlotCache的调用，此处可以确认:Jedis在集群模式下在遇到MOVED的报错时候，会发送Redis命令CLUSTER SLOTS,重新拉取Redis集群的slot拓扑结构。

2.2.3 客户端实现原理小结

由于Jedis是Java的Redis客户端，Hiredis-vip是c++的Redis客户端，可以简单认为这种异常处理机制是共性操作。

对客户端集群模式下对MOVED的流程梳理大概如下：

总的来说：

1）使用客户端缓存的slot拓扑进行对key的访问；

2）Redis节点返回正常：

• 访问正常，继续后续操作

3）Redis节点返回MOVED：

• 对Redis节点进行CLUSTER SLOTS指令执行，更新拓扑；
• 使用新的拓扑对key重新访问。

2.2.3 客户端排查小结

Redis集群正在扩容，也就是必然存在一些Redis客户端在访问Redis集群遇到MOVED，执行Redis指令CLUSTER SLOTS进行拓扑结构更新。

如果迁移的key命中率高，CLUSTER SLOTS指令会更加频繁的执行。这样导致的结果是迁移过程中Redis集群会持续被客户端执行CLUSTER SLOTS指令。

2.3 排查小结

此处，结合Redis侧的CLUSTER SLOTS机制以及客户端对MOVED的处理逻辑，可以解答之前的几个个问题：

为什么会有较多的Cluster指令被执行？

• 因为发生过迁移操作，业务访问一些迁移过的key会拿到MOVED返回，客户端会对该返回重新拉取slot拓扑信息，执行CLUSTER SLOTS。

为什么Cluster指令执行的时候CPU资源比较高？

• 分析Redis源码，发现CLUSTER SLOT指令的时间复杂度和主节点个数成正比。业务当前的Redis集群主节点个数比较多，自然耗时高，占用CPU资源高。

为什么节点规模大的集群迁移slot操作容易“中招”？

• 迁移操作必然带来一些客户端访问key的时候返回MOVED；
• 客户端对于MOVED的返回会执行CLUSTER SLOTS指令；
• CLUSTER SLOTS指令随着集群主节点个数的增加，时延会上升；
• 业务的访问在slot的迁移期间会因为CLUSTER SLOTS的时延上升，在外部的感知是执行指令的时延升高。

三、优化

3.1 现状分析

根据目前的情况来看，客户端遇到MOVED进行CLUSTER SLOTS执行是正常的流程，因为需要更新集群的slot拓扑结构提高后续的集群访问效率。

此处流程除了Jedis，Hiredis-vip，其他的客户端应该也会进行类似的slot信息缓存优化。此处流程优化空间不大，是Redis的集群访问机制决定。

因此对Redis的集群信息记录进行分析。

3.1.1 Redis集群元数据分析

集群中每一个Redis节点都会有一些集群的元数据记录，记录于server.cluster，内容如下：

1
2
3
4
5
6
7
8

java复制代码typedef struct clusterState {
    ...
    dict *nodes;          /* Hash table of name -> clusterNode structures */
    /*nodes记录的是所有的节点，使用dict记录*/
    ...
    clusterNode *slots[CLUSTER_SLOTS];/*slots记录的是slot数组，内容是node的指针*/
    ...
} clusterState;

如2.1所述，原有逻辑通过遍历每个节点的slot信息获得拓扑结构。

3.1.2 Redis集群元数据分析

观察CLUSTER SLOTS的返回结果：

1
2
3
4
5
6
7
8
9
10

java复制代码/* Format: 1) 1) start slot
 *            2) end slot
 *            3) 1) master IP
 *               2) master port
 *               3) node ID
 *            4) 1) replica IP
 *               2) replica port
 *               3) node ID
 *           ... continued until done
 */

结合server.cluster中存放的集群信息，笔者认为此处可以使用server.cluster->slots进行遍历。因为server.cluster->slots已经在每一次集群的拓扑变化得到了更新，保存的是节点指针。

3.2 优化方案

简单的优化思路如下：

• 对slot进行遍历，找出slot中节点是连续的块；
• 当前遍历的slot的节点如果和之前遍历的节点一致，说明目前访问的slot和前面的是在同一个节点下，也就是是在某个节点下的“连续”的slot区域内；
• 当前遍历的slot的节点如果和之前遍历的节点不一致，说明目前访问的slot和前面的不同，前面的“连续”slot区域可以进行输出；而当前slot作为下一个新的“连续”slot区域的开始。

因此只要对server.cluster->slots进行遍历，可以满足需求。简单表示大概如下：

这样的时间复杂度降低到<slot总个数>。

3.3 实现

优化逻辑如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

java复制代码void clusterReplyMultiBulkSlots(client * c) {
    /* Format: 1) 1) start slot
     *            2) end slot
     *            3) 1) master IP
     *               2) master port
     *               3) node ID
     *            4) 1) replica IP
     *               2) replica port
     *               3) node ID
     *           ... continued until done
     */
    clusterNode *n = NULL;
    int num_masters = 0, start = -1;
    void *slot_replylen = addReplyDeferredLen(c);
 
    for (int i = 0; i <= CLUSTER_SLOTS; i++) {
        /*对所有slot进行遍历*/
        /* Find start node and slot id. */
        if (n == NULL) {
            if (i == CLUSTER_SLOTS) break;
            n = server.cluster->slots[i];
            start = i;
            continue;
        }
 
        /* Add cluster slots info when occur different node with start
         * or end of slot. */
        if (i == CLUSTER_SLOTS || n != server.cluster->slots[i]) {
            /*遍历主节点下面的备节点，添加返回客户端的信息*/
            addNodeReplyForClusterSlot(c, n, start, i-1);
            num_masters++;
            if (i == CLUSTER_SLOTS) break;
            n = server.cluster->slots[i];
            start = i;
        }
    }
    setDeferredArrayLen(c, slot_replylen, num_masters);
}

通过对server.cluster->slots进行遍历，找到某个节点下的“连续”的slot区域，一旦后续不连续，把之前的“连续”slot区域的节点信息以及其备节点信息进行输出，然后继续下一个“连续”slot区域的查找于输出。

四、优化结果对比

对两个版本的Redis的CLUSTER SLOTS指令进行横向对比。

4.1 测试环境&压测场景

操作系统：manjaro 20.2

硬件配置：

• CPU：AMD Ryzen 7 4800H
• DRAM：DDR4 3200MHz 8G*2

Redis集群信息：

1）持久化配置

• 关闭aof
• 关闭bgsave

2）集群节点信息：

• 节点个数：100
• 所有节点都是主节点

压测场景：

• 使用benchmark工具对集群单个节点持续发送CLUSTER SLOTS指令；
• 对其中一个版本压测完后，回收集群，重新部署后再进行下一轮压测。

4.2 CPU资源占用对比

perf导出火焰图。原有版本：

优化后：

可以明显看到，优化后的占比大幅度下降。基本符合预期。

4.3 耗时对比

在上进行测试，嵌入耗时测试代码：

1
2
3
4
5
6
7

java复制代码else if (!strcasecmp(c->argv[1]->ptr,"slots") && c->argc == 2) {
        /* CLUSTER SLOTS */
        long long now = ustime();
        clusterReplyMultiBulkSlots(c);
        serverLog(LL_NOTICE,
            "cluster slots cost time:%lld us", ustime() - now);
    }

输入日志进行对比；

原版的日志输出：

37351:M 06 Mar 2021 16:11:39.313 * cluster slots cost time:2061 us。

优化后版本日志输出：

35562:M 06 Mar 2021 16:11:27.862 * cluster slots cost time:168 us。

从耗时上看下降明显：从2000+us 下降到200-us；在100个主节点的集群中的耗时缩减到原来的8.2%；优化结果基本符合预期。

五、总结

这里可以简单描述下文章上述的动作从而得出的这样的一个结论：性能缺陷。

简单总结下上述的排查以及优化过程：

• Redis大集群因为CLUSTER命令导致某些节点的访问延迟明显；
• 使用perf top指令对Redis实例进行排查，发现clusterReplyMultiBulkSlots命令占用CPU资源异常；
• 对clusterReplyMultiBulkSlots进行分析，该函数存在明显的性能问题；
• 对clusterReplyMultiBulkSlots进行优化，性能提升明显。

从上述的排查以及优化过程可以得出一个结论：目前的Redis在CLUSTER SLOT指令存在性能缺陷。

因为Redis的数据分片机制，决定了Redis集群模式下的key访问方法是缓存slot的拓扑信息。优化点也只能在CLUSTER SLOTS入手。而Redis的集群节点个数一般没有这么大，问题暴露的不明显。

其实Hiredis-vip的逻辑也存在一定问题。如2.2.1所说，Hiredis-vip的slot拓扑更新方法是遍历所有的节点挨个进行CLUSTER SLOTS。如果Redis集群规模较大而且业务侧的客户端规模较多，会出现连锁反应：

1）如果Redis集群较大，CLUSTER SLOTS响应比较慢；

2）如果某个节点没有响应或者返回报错，Hiredis-vip客户端会对下一个节点继续进行请求；

3）Hiredis-vip客户端中对Redis集群节点迭代遍历的方法相同（因为集群的信息在各个客户端基本一致），此时当客户端规模较大的时候，某个Redis节点可能存在阻塞，就会导致hiredis-vip客户端遍历下一个Redis节点；

4）大量Hiredis-vip客户端挨个地对一些Redis节点进行访问，如果Redis节点无法负担这样的请求，这样会导致Redis节点在大量Hiredis-vip客户端的“遍历”下挨个请求：

结合上述第3点，可以想象一下：有1w个客户端对该Redis集群进行访问。因为某个命中率较高的key存在迁移操作，所有的客户端都需要更新slot拓扑。由于所有客户端缓存的集群节点信息相同，因此遍历各个节点的顺序是一致的。这1w个客户端都使用同样的顺序对集群各个节点进行遍历地操作CLUSTER SLOTS。由于CLUSTER SLOTS在大集群中性能较差，Redis节点很容易会被大量客户端请求导致不可访问。Redis节点会根据遍历顺序依次被大部分的客户端（例如9k+个客户端）访问，执行CLUSTER SLOTS指令，导致Redis节点挨个被阻塞。

5）最终的表现是大部分Redis节点的CPU负载暴涨，很多Hiredis-vip客户端则继续无法更新slot拓扑。

最终结果是大规模的Redis集群在进行slot迁移操作后，在大规模的Hiredis-vip客户端访问下业务侧感知是普通指令时延变高，而Redis实例CPU资源占用高涨。这个逻辑可以进行一定优化。

目前上述分节3的优化已经提交并合并到Redis 6.2.2版本中。

六、参考资料

1、Hiredis-vip: github.com

2、Jedis: https://github.com/redis/jedis

3、Redis: github.com/redis/redis

4、Perf：perf.wiki.kernel.org

作者：vivo互联网数据库团队—Yuan Jianwei

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

12306抢票,极限并发带来的思考?

每到节假日期间,一二线城市返乡、外出游玩的人们几乎都面临着一个问题：抢火车票!虽然现在大多数情况下都能订到票,但是放票瞬间即无票的场景，相信大家都深有体会。尤其是春节期间，大家不仅使用12306，还会考虑“智行”和其他的抢票软件,全国上下几亿人在这段时间都在抢票。“12306服务”承受着这个世界上任何秒杀系统都无法超越的QPS,上百万的并发再正常不过了！笔者专门研究了一下“12306”的服务端架构,学习到了其系统设计上很多亮点，在这里和大家分享一下并模拟一个例子：如何在100万人同时抢1万张火车票时，系统提供正常、稳定的服务。github代码地址

1. 大型高并发系统架构

高并发的系统架构都会采用分布式集群部署，服务上层有着层层负载均衡，并提供各种容灾手段(双火机房、节点容错、服务器灾备等)保证系统的高可用,流量也会根据不同的负载能力和配置策略均衡到不同的服务器上。下边是一个简单的示意图:

1.1 负载均衡简介

上图中描述了用户请求到服务器经历了三层的负载均衡，下边分别简单介绍一下这三种负载均衡：

• OSPF(开放式最短链路优先)是一个内部网关协议(Interior Gateway Protocol,简称IGP)。OSPF通过路由器之间通告网络接口的状态来建立链路状态数据库，生成最短路径树，OSPF会自动计算路由接口上的Cost值，但也可以通过手工指定该接口的Cost值，手工指定的优先于自动计算的值。OSPF计算的Cost，同样是和接口带宽成反比，带宽越高，Cost值越小。到达目标相同Cost值的路径，可以执行负载均衡，最多6条链路同时执行负载均衡。
• LVS (Linux VirtualServer)，它是一种集群(Cluster)技术，采用IP负载均衡技术和基于内容请求分发技术。调度器具有很好的吞吐率，将请求均衡地转移到不同的服务器上执行，且调度器自动屏蔽掉服务器的故障，从而将一组服务器构成一个高性能的、高可用的虚拟服务器。
• Nginx想必大家都很熟悉了,是一款非常高性能的http代理/反向代理服务器,服务开发中也经常使用它来做负载均衡。Nginx实现负载均衡的方式主要有三种:轮询、加权轮询、ip hash轮询，下面我们就针对Nginx的加权轮询做专门的配置和测试

1.2 Nginx加权轮询的演示

Nginx实现负载均衡通过upstream模块实现，其中加权轮询的配置是可以给相关的服务加上一个权重值，配置的时候可能根据服务器的性能、负载能力设置相应的负载。下面是一个加权轮询负载的配置，我将在本地的监听3001-3004端口,分别配置1，2，3，4的权重:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

ini复制代码#配置负载均衡
    upstream load_rule {
       server 127.0.0.1:3001 weight=1;
       server 127.0.0.1:3002 weight=2;
       server 127.0.0.1:3003 weight=3;
       server 127.0.0.1:3004 weight=4;
    }
    ...
    server {
    listen       80;
    server_name  load_balance.com www.load_balance.com;
    location / {
       proxy_pass http://load_rule;
    }
}

我在本地/etc/hosts目录下配置了 www.load\_balance.com的虚拟域名地址，接下来使用Go语言开启四个http端口监听服务，下面是监听在3001端口的Go程序,其他几个只需要修改端口即可：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

go复制代码package main

import (
 "net/http"
 "os"
 "strings"
)

func main() {
 http.HandleFunc("/buy/ticket", handleReq)
 http.ListenAndServe(":3001", nil)
}

//处理请求函数,根据请求将响应结果信息写入日志
func handleReq(w http.ResponseWriter, r *http.Request) {
 failedMsg :=  "handle in port:"
 writeLog(failedMsg, "./stat.log")
}

//写入日志
func writeLog(msg string, logPath string) {
 fd, _ := os.OpenFile(logPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
 defer fd.Close()
 content := strings.Join([]string{msg, "\r\n"}, "3001")
 buf := []byte(content)
 fd.Write(buf)
}

我将请求的端口日志信息写到了./stat.log文件当中，然后使用ab压测工具做压测:

1

arduino复制代码ab -n 1000 -c 100 http://www.load_balance.com/buy/ticket

统计日志中的结果，3001-3004端口分别得到了100、200、300、400的请求量，这和我在nginx中配置的权重占比很好的吻合在了一起，并且负载后的流量非常的均匀、随机。具体的实现大家可以参考nginx的upsteam模块实现源码，这里推荐一篇文章：Nginx 中 upstream 机制的负载均衡

2.秒杀抢购系统选型

回到我们最初提到的问题中来：火车票秒杀系统如何在高并发情况下提供正常、稳定的服务呢？

从上面的介绍我们知道用户秒杀流量通过层层的负载均衡，均匀到了不同的服务器上，即使如此，集群中的单机所承受的QPS也是非常高的。如何将单机性能优化到极致呢？要解决这个问题，我们就要想明白一件事：通常订票系统要处理生成订单、减扣库存、用户支付这三个基本的阶段，我们系统要做的事情是要保证火车票订单不超卖、不少卖，每张售卖的车票都必须支付才有效，还要保证系统承受极高的并发。这三个阶段的先后顺序改怎么分配才更加合理呢?我们来分析一下：

2.1 下单减库存

当用户并发请求到达服务端时，首先创建订单，然后扣除库存，等待用户支付。这种顺序是我们一般人首先会想到的解决方案，这种情况下也能保证订单不会超卖，因为创建订单之后就会减库存，这是一个原子操作。但是这样也会产生一些问题，第一就是在极限并发情况下，任何一个内存操作的细节都至关影响性能，尤其像创建订单这种逻辑，一般都需要存储到磁盘数据库的，对数据库的压力是可想而知的；第二是如果用户存在恶意下单的情况，只下单不支付这样库存就会变少，会少卖很多订单，虽然服务端可以限制IP和用户的购买订单数量，这也不算是一个好方法。

2.2 支付减库存

如果等待用户支付了订单在减库存，第一感觉就是不会少卖。但是这是并发架构的大忌，因为在极限并发情况下，用户可能会创建很多订单，当库存减为零的时候很多用户发现抢到的订单支付不了了，这也就是所谓的“超卖”。也不能避免并发操作数据库磁盘IO

2.3 预扣库存

从上边两种方案的考虑，我们可以得出结论：只要创建订单，就要频繁操作数据库IO。那么有没有一种不需要直接操作数据库IO的方案呢，这就是预扣库存。先扣除了库存，保证不超卖，然后异步生成用户订单，这样响应给用户的速度就会快很多；那么怎么保证不少卖呢？用户拿到了订单，不支付怎么办？我们都知道现在订单都有有效期，比如说用户五分钟内不支付，订单就失效了，订单一旦失效，就会加入新的库存，这也是现在很多网上零售企业保证商品不少卖采用的方案。订单的生成是异步的,一般都会放到MQ、kafka这样的即时消费队列中处理,订单量比较少的情况下，生成订单非常快，用户几乎不用排队。

3. 扣库存的艺术

从上面的分析可知，显然预扣库存的方案最合理。我们进一步分析扣库存的细节，这里还有很大的优化空间，库存存在哪里？怎样保证高并发下，正确的扣库存，还能快速的响应用户请求？

在单机低并发情况下，我们实现扣库存通常是这样的:

为了保证扣库存和生成订单的原子性，需要采用事务处理，然后取库存判断、减库存，最后提交事务，整个流程有很多IO，对数据库的操作又是阻塞的。这种方式根本不适合高并发的秒杀系统。

接下来我们对单机扣库存的方案做优化：本地扣库存。我们把一定的库存量分配到本地机器，直接在内存中减库存，然后按照之前的逻辑异步创建订单。改进过之后的单机系统是这样的:

这样就避免了对数据库频繁的IO操作，只在内存中做运算，极大的提高了单机抗并发的能力。但是百万的用户请求量单机是无论如何也抗不住的，虽然nginx处理网络请求使用epoll模型，c10k的问题在业界早已得到了解决。但是linux系统下，一切资源皆文件，网络请求也是这样，大量的文件描述符会使操作系统瞬间失去响应。上面我们提到了nginx的加权均衡策略，我们不妨假设将100W的用户请求量平均均衡到100台服务器上，这样单机所承受的并发量就小了很多。然后我们每台机器本地库存100张火车票，100台服务器上的总库存还是1万，这样保证了库存订单不超卖,下面是我们描述的集群架构:

问题接踵而至，在高并发情况下，现在我们还无法保证系统的高可用，假如这100台服务器上有两三台机器因为扛不住并发的流量或者其他的原因宕机了。那么这些服务器上的订单就卖不出去了，这就造成了订单的少卖。要解决这个问题，我们需要对总订单量做统一的管理，这就是接下来的容错方案。服务器不仅要在本地减库存，另外要远程统一减库存。有了远程统一减库存的操作，我们就可以根据机器负载情况，为每台机器分配一些多余的“buffer库存”用来防止机器中有机器宕机的情况。我们结合下面架构图具体分析一下:

我们采用Redis存储统一库存，因为Redis的性能非常高，号称单机QPS能抗10W的并发。在本地减库存以后，如果本地有订单，我们再去请求redis远程减库存，本地减库存和远程减库存都成功了，才返回给用户抢票成功的提示,这样也能有效的保证订单不会超卖。当机器中有机器宕机时，因为每个机器上有预留的buffer余票，所以宕机机器上的余票依然能够在其他机器上得到弥补，保证了不少卖。buffer余票设置多少合适呢，理论上buffer设置的越多，系统容忍宕机的机器数量就越多，但是buffer设置的太大也会对redis造成一定的影响。虽然redis内存数据库抗并发能力非常高，请求依然会走一次网络IO,其实抢票过程中对redis的请求次数是本地库存和buffer库存的总量，因为当本地库存不足时，系统直接返回用户“已售罄”的信息提示，就不会再走统一扣库存的逻辑，这在一定程度上也避免了巨大的网络请求量把redis压跨，所以buffer值设置多少，需要架构师对系统的负载能力做认真的考量。

4. 代码演示

Go语言原生为并发设计，我采用go语言给大家演示一下单机抢票的具体流程。

4.1 初始化工作

go包中的init函数先于main函数执行，在这个阶段主要做一些准备性工作。我们系统需要做的准备工作有：初始化本地库存、初始化远程redis存储统一库存的hash键值、初始化redis连接池；另外还需要初始化一个大小为1的int类型chan,目的是实现分布式锁的功能，也可以直接使用读写锁或者使用redis等其他的方式避免资源竞争,但使用channel更加高效，这就是go语言的哲学：不要通过共享内存来通信，而要通过通信来共享内存。redis库使用的是redigo，下面是代码实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

go复制代码...
//localSpike包结构体定义
package localSpike

type LocalSpike struct {
 LocalInStock     int64
 LocalSalesVolume int64
}
...
//remoteSpike对hash结构的定义和redis连接池
package remoteSpike
//远程订单存储健值
type RemoteSpikeKeys struct {
 SpikeOrderHashKey string //redis中秒杀订单hash结构key
 TotalInventoryKey string //hash结构中总订单库存key
 QuantityOfOrderKey string //hash结构中已有订单数量key
}

//初始化redis连接池
func NewPool() *redis.Pool {
 return &redis.Pool{
  MaxIdle:   10000,
  MaxActive: 12000, // max number of connections
  Dial: func() (redis.Conn, error) {
   c, err := redis.Dial("tcp", ":6379")
   if err != nil {
    panic(err.Error())
   }
   return c, err
  },
 }
}
...
func init() {
 localSpike = localSpike2.LocalSpike{
  LocalInStock:     150,
  LocalSalesVolume: 0,
 }
 remoteSpike = remoteSpike2.RemoteSpikeKeys{
  SpikeOrderHashKey:  "ticket_hash_key",
  TotalInventoryKey:  "ticket_total_nums",
  QuantityOfOrderKey: "ticket_sold_nums",
 }
 redisPool = remoteSpike2.NewPool()
 done = make(chan int, 1)
 done <- 1
}

4.2 本地扣库存和统一扣库存

本地扣库存逻辑非常简单，用户请求过来，添加销量，然后对比销量是否大于本地库存，返回bool值:

1
2
3
4
5
6

go复制代码package localSpike
//本地扣库存,返回bool值
func (spike *LocalSpike) LocalDeductionStock() bool{
 spike.LocalSalesVolume = spike.LocalSalesVolume + 1
 return spike.LocalSalesVolume < spike.LocalInStock
}

注意这里对共享数据LocalSalesVolume的操作是要使用锁来实现的，但是因为本地扣库存和统一扣库存是一个原子性操作，所以在最上层使用channel来实现，这块后边会讲。统一扣库存操作redis，因为redis是单线程的，而我们要实现从中取数据，写数据并计算一些列步骤，我们要配合lua脚本打包命令，保证操作的原子性:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

lua复制代码package remoteSpike
......
const LuaScript = `
        local ticket_key = KEYS[1]
        local ticket_total_key = ARGV[1]
        local ticket_sold_key = ARGV[2]
        local ticket_total_nums = tonumber(redis.call('HGET', ticket_key, ticket_total_key))
        local ticket_sold_nums = tonumber(redis.call('HGET', ticket_key, ticket_sold_key))
  -- 查看是否还有余票,增加订单数量,返回结果值
       if(ticket_total_nums >= ticket_sold_nums) then
            return redis.call('HINCRBY', ticket_key, ticket_sold_key, 1)
        end
        return 0
`
//远端统一扣库存
func (RemoteSpikeKeys *RemoteSpikeKeys) RemoteDeductionStock(conn redis.Conn) bool {
 lua := redis.NewScript(1, LuaScript)
 result, err := redis.Int(lua.Do(conn, RemoteSpikeKeys.SpikeOrderHashKey, RemoteSpikeKeys.TotalInventoryKey, RemoteSpikeKeys.QuantityOfOrderKey))
 if err != nil {
  return false
 }
 return result != 0
}

我们使用hash结构存储总库存和总销量的信息,用户请求过来时，判断总销量是否大于库存，然后返回相关的bool值。在启动服务之前，我们需要初始化redis的初始库存信息:

1

arduino复制代码 hmset ticket_hash_key "ticket_total_nums" 10000 "ticket_sold_nums" 0

4.3 响应用户信息

我们开启一个http服务，监听在一个端口上:

1
2
3
4
5
6

go复制代码package main
...
func main() {
 http.HandleFunc("/buy/ticket", handleReq)
 http.ListenAndServe(":3005", nil)
}

上面我们做完了所有的初始化工作，接下来handleReq的逻辑非常清晰，判断是否抢票成功，返回给用户信息就可以了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

go复制代码package main
//处理请求函数,根据请求将响应结果信息写入日志
func handleReq(w http.ResponseWriter, r *http.Request) {
 redisConn := redisPool.Get()
 LogMsg := ""
 <-done
 //全局读写锁
 if localSpike.LocalDeductionStock() && remoteSpike.RemoteDeductionStock(redisConn) {
  util.RespJson(w, 1,  "抢票成功", nil)
  LogMsg = LogMsg + "result:1,localSales:" + strconv.FormatInt(localSpike.LocalSalesVolume, 10)
 } else {
  util.RespJson(w, -1, "已售罄", nil)
  LogMsg = LogMsg + "result:0,localSales:" + strconv.FormatInt(localSpike.LocalSalesVolume, 10)
 }
 done <- 1

 //将抢票状态写入到log中
 writeLog(LogMsg, "./stat.log")
}

func writeLog(msg string, logPath string) {
 fd, _ := os.OpenFile(logPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
 defer fd.Close()
 content := strings.Join([]string{msg, "\r\n"}, "")
 buf := []byte(content)
 fd.Write(buf)
}

前边提到我们扣库存时要考虑竞态条件，我们这里是使用channel避免并发的读写，保证了请求的高效顺序执行。我们将接口的返回信息写入到了./stat.log文件方便做压测统计。

4.4 单机服务压测

开启服务，我们使用ab压测工具进行测试：

1

arduino复制代码ab -n 10000 -c 100 http://127.0.0.1:3005/buy/ticket

下面是我本地低配mac的压测信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

matlab复制代码This is ApacheBench, Version 2.3 <$Revision: 1826891 $>
Copyright 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licensed to The Apache Software Foundation, http://www.apache.org/

Benchmarking 127.0.0.1 (be patient)
Completed 1000 requests
Completed 2000 requests
Completed 3000 requests
Completed 4000 requests
Completed 5000 requests
Completed 6000 requests
Completed 7000 requests
Completed 8000 requests
Completed 9000 requests
Completed 10000 requests
Finished 10000 requests


Server Software:
Server Hostname:        127.0.0.1
Server Port:            3005

Document Path:          /buy/ticket
Document Length:        29 bytes

Concurrency Level:      100
Time taken for tests:   2.339 seconds
Complete requests:      10000
Failed requests:        0
Total transferred:      1370000 bytes
HTML transferred:       290000 bytes
Requests per second:    4275.96 [#/sec] (mean)
Time per request:       23.387 [ms] (mean)
Time per request:       0.234 [ms] (mean, across all concurrent requests)
Transfer rate:          572.08 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    8  14.7      6     223
Processing:     2   15  17.6     11     232
Waiting:        1   11  13.5      8     225
Total:          7   23  22.8     18     239

Percentage of the requests served within a certain time (ms)
  50%     18
  66%     24
  75%     26
  80%     28
  90%     33
  95%     39
  98%     45
  99%     54
 100%    239 (longest request)

根据指标显示，我单机每秒就能处理4000+的请求，正常服务器都是多核配置，处理1W+的请求根本没有问题。而且查看日志发现整个服务过程中，请求都很正常，流量均匀，redis也很正常：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

makefile复制代码//stat.log
...
result:1,localSales:145
result:1,localSales:146
result:1,localSales:147
result:1,localSales:148
result:1,localSales:149
result:1,localSales:150
result:0,localSales:151
result:0,localSales:152
result:0,localSales:153
result:0,localSales:154
result:0,localSales:156
...

5.总结回顾

总体来说，秒杀系统是非常复杂的。我们这里只是简单介绍模拟了一下单机如何优化到高性能，集群如何避免单点故障，保证订单不超卖、不少卖的一些策略，完整的订单系统还有订单进度的查看，每台服务器上都有一个任务，定时的从总库存同步余票和库存信息展示给用户,还有用户在订单有效期内不支付，释放订单，补充到库存等等。

我们实现了高并发抢票的核心逻辑，可以说系统设计的非常的巧妙，巧妙的避开了对DB数据库IO的操作，对Redis网络IO的高并发请求，几乎所有的计算都是在内存中完成的，而且有效的保证了不超卖、不少卖，还能够容忍部分机器的宕机。我觉得其中有两点特别值得学习总结：

• 负载均衡，分而治之。通过负载均衡，将不同的流量划分到不同的机器上，每台机器处理好自己的请求，将自己的性能发挥到极致，这样系统的整体也就能承受极高的并发了，就像工作的的一个团队，每个人都将自己的价值发挥到了极致，团队成长自然是很大的。
• 合理的使用并发和异步。自epoll网络架构模型解决了c10k问题以来，异步越来被服务端开发人员所接受，能够用异步来做的工作，就用异步来做，在功能拆解上能达到意想不到的效果，这点在nginx、node.js、redis上都能体现，他们处理网络请求使用的epoll模型，用实践告诉了我们单线程依然可以发挥强大的威力。服务器已经进入了多核时代，go语言这种天生为并发而生的语言，完美的发挥了服务器多核优势，很多可以并发处理的任务都可以使用并发来解决，比如go处理http请求时每个请求都会在一个goroutine中执行，总之:怎样合理的压榨CPU,让其发挥出应有的价值，是我们一直需要探索学习的方向。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有