发表于

博主（编码砖家）负责的项目主要采用阿里云数据库MySQL，最近频繁出现慢SQL告警，执行时**间最长的竟**然高达5分钟。导出日志后分析，主要原因竟然是没有命中索引和没有分页处理 。

其实这是非常低级的错误，我不禁后背一凉，团队成员的技术水平亟待提高啊。改造这些SQL的过程中，总结了一些经验分享给大家，如果有错误欢迎批评指正。

MySQL性能

最大数据量

抛开数据量和并发数，谈性能都是耍流氓 。MySQL没有限制单表最大记录数，它取决于操作系统对文件大小的限制。

文件系统 单文件大小限制
FAT32 最大4G
NTFS 最大64GB
NTFS5.0 最大2TB
EXT2 块大小为1024字节，文件最大容量16GB；块大小为4096字节，文件最大容量2TB
EXT3 块大小为4KB，文件最大容量为4TB
EXT4 理论可以大于16TB

《阿里巴巴Java开发手册》提出单表行数超过500万行或者单表容量超过2GB，才推荐分库分表。性能由综合因素决定，抛开业务复杂度，影响程度依次是硬件配置、MySQL配置、数据表设计、索引优化。500万这个值仅供参考，并非铁律。博主曾经操作过超过4亿行数据的单表，分页查询最新的20条记录耗时0.6秒，SQL语句大致是 select field_1,field_2 from table where id < #{prePageMinId} order by id desc limit 20，prePageMinId是上一页数据记录的最小ID。虽然当时查询速度还凑合，随着数据不断增长，有朝一日必定不堪重负。分库分表是个周期长而风险高的大活儿，应该尽可能在当前结构上优化，比如升级硬件、迁移历史数据等等，实在没辙了再分。对分库分表感兴趣的同学可以阅读分库分表的基本思想。

最大并发数

并发数是指同一时刻数据库能处理多少个请求，由max_connections和max_user_connections决定。**max_connections是指MySQL实例的最大连接数，上限值是16384，max_user_connections是指每个数据库用户的最大连接数。MySQL会为每个连接提供缓冲区，意味着消耗更多的内存。如果连接数设置太高硬件吃不消，太低又不能充分利用硬件。一般要求两者比值超过10%，计算方法如下：

1
erlang复制代码max_used_connections / max_connections * 100% = 3/100 *100% ≈ 3%

查看最大连接数与响应最大连接数：

1
sql复制代码show variables like '%max_connections%';show variables like '%max_user_connections%';

在配置文件my.cnf中修改最大连接数

1
ini复制代码[mysqld]max_connections = 100max_used_connections = 20

查询耗时0.5秒

建议将单次查询耗时控制在0.5秒以内，0.5秒是个经验值，源于用户体验的 3秒原则 。如果用户的操作3秒内没有响应，将会厌烦甚至退出。响应时间=客户端UI渲染耗时+网络请求耗时+应用程序处理耗时+查询数据库耗时，0.5秒就是留给数据库1/6的处理时间。

实施原则

相比NoSQL数据库，MySQL是个娇气脆弱的家伙。它就像体育课上的女同学，一点纠纷就和同学闹别扭(扩容难)，跑两步就气喘吁吁(容量小并发低)，常常身体不适要请假(SQL约束太多)。如今大家都会搞点分布式，应用程序扩容比数据库要容易得多，所以实施原则是 数据库少干活，应用程序多干活 。

  • 充分利用但不滥用索引，须知索引也消耗磁盘和CPU。
  • 不推荐使用数据库函数格式化数据，交给应用程序处理。
  • 不推荐使用外键约束，用应用程序保证数据准确性。
  • 写多读少的场景，不推荐使用唯一索引，用应用程序保证唯一性。
  • 适当冗余字段，尝试创建中间表，用应用程序计算中间结果，用空间换时间。
  • 不允许执行极度耗时的事务，配合应用程序拆分成更小的事务。
  • 预估重要数据表（比如订单表）的负载和数据增长态势，提前优化。

数据表设计

数据类型

数据类型的选择原则：更简单或者占用空间更小。

  • 如果长度能够满足，整型尽量使用tinyint、smallint、medium_int而非int。
  • 如果字符串长度确定，采用char类型。
  • 如果varchar能够满足，不采用text类型。
  • 精度要求较高的使用decimal类型，也可以使用BIGINT，比如精确两位小数就乘以100后保存。

尽量采用timestamp而非datetime。

类型 占据字节 描述
datetime 8字节 ‘1000-01-01 00:00:00.000000’ to ‘9999-12-31 23:59:59.999999
timestamp 4字节 ‘1970-01-01 00:00:01.000000’ to ‘2038-01-19 03:14:07.999999’

相比datetime，timestamp占用更少的空间，以UTC的格式储存自动转换时区。

避免空值

MySQL中字段为NULL时依然占用空间，会使索引、索引统计更加复杂。从NULL值更新到非NULL无法做到原地更新，容易发生索引分裂影响性能。尽可能将NULL值用有意义的值代替，也能避免SQL语句里面包含 is not null的判断。

text类型优化

由于text字段储存大量数据，表容量会很早涨上去，影响其他字段的查询性能。建议抽取出来放在子表里，用业务主键关联。

索引优化

索引分类

  1. 普通索引：最基本的索引。
  2. 组合索引：多个字段上建立的索引，能够加速复合查询条件的检索。
  3. 唯一索引：与普通索引类似，但索引列的值必须唯一，允许有空值。
  4. 组合唯一索引：列值的组合必须唯一。
  5. 主键索引：特殊的唯一索引，用于唯一标识数据表中的某一条记录，不允许有空值，一般用primary key约束。
  6. 全文索引：用于海量文本的查询，MySQL5.6之后的InnoDB和MyISAM均支持全文索引。由于查询精度以及扩展性不佳，更多的企业选择Elasticsearch。

索引优化

  1. 分页查询很重要，如果查询数据量超过30%，MYSQL不会使用索引。
  2. 单表索引数不超过5个、单个索引字段数不超过5个。
  3. 字符串可使用前缀索引，前缀长度控制在5-8个字符。
  4. 字段唯一性太低，增加索引没有意义，如：是否删除、性别。
  5. 合理使用覆盖索引，如下所示：

select login_name, nick_name from member where login_name = ?

login_name, nick_name两个字段建立组合索引，比login_name简单索引要更快

SQL优化

分批处理

博主小时候看到鱼塘挖开小口子放水，水面有各种漂浮物。浮萍和树叶总能顺利通过出水口，而树枝会挡住其他物体通过，有时还会卡住，需要人工清理。MySQL就是鱼塘，最大并发数和网络带宽就是出水口，用户SQL就是漂浮物。不带分页参数的查询或者影响大量数据的update和delete操作，都是树枝，我们要把它打散分批处理，举例说明：业务描述：更新用户所有已过期的优惠券为不可用状态。SQL语句：update status=0 FROMcoupon WHERE expire_date <= #{currentDate} and status=1;如果大量优惠券需要更新为不可用状态，执行这条SQL可能会堵死其他SQL，分批处理伪代码如下：

1
2
3
4
5
6
7
8
9
10
ini复制代码int pageNo = 1;
int PAGE_SIZE = 100;
while(true) {
    List<Integer> batchIdList = queryList('select id FROM `coupon` WHERE expire_date <= #{currentDate} and status = 1 limit #{(pageNo-1) * PAGE_SIZE},#{PAGE_SIZE}');
    if (CollectionUtils.isEmpty(batchIdList)) {
        return;
    }
    update('update status = 0 FROM `coupon` where status = 1 and id in #{batchIdList}')
    pageNo ++;
}

操作符<>优化

通常<>操作符无法使用索引，举例如下，查询金额不为100元的订单：select id from orders where amount != 100;如果金额为100的订单极少，这种数据分布严重不均的情况下，有可能使用索引。鉴于这种不确定性，采用union聚合搜索结果，改写方法如下：

1
sql复制代码(select id from orders where amount > 100) union all(select id from orders where amount < 100 and amount > 0)

OR优化

在Innodb引擎下or无法使用组合索引，比如：

1
csharp复制代码select id，product_name from orders where mobile_no = '13421800407' or user_id = 100;

OR无法命中mobile_no + user_id的组合索引，可采用union，如下所示：

1
sql复制代码(select id，product_name from orders where mobile_no = '13421800407') union(select id，product_name from orders where user_id = 100);

此时id和product_name字段都有索引，查询才最高效。

IN优化

  1. IN适合主表大子表小，EXIST适合主表小子表大。由于查询优化器的不断升级，很多场景这两者性能差不多一样了。
  2. 尝试改为join查询，举例如下：

select id from orders where user_id in (select id from user where level = ‘VIP’);

采用JOIN如下所示：

1
sql复制代码select o.id from orders o left join user u on o.user_id = u.id where u.level = 'VIP';

不做列运算

通常在查询条件列运算会导致索引失效，如下所示：查询当日订单

1
csharp复制代码select id from order where date_format(create_time，'%Y-%m-%d') = '2019-07-01';

date_format函数会导致这个查询无法使用索引，改写后：

1
sql复制代码select id from order where create_time between '2019-07-01 00:00:00' and '2019-07-01 23:59:59';

避免Select all

如果不查询表中所有的列，避免使用 SELECT *，它会进行全表扫描，不能有效利用索引。

Like优化

like用于模糊查询，举个例子（field已建立索引）：

1
sql复制代码SELECT column FROM table WHERE field like '%keyword%';

这个查询未命中索引，换成下面的写法：

1
sql复制代码SELECT column FROM table WHERE field like 'keyword%';

去除了前面的%查询将会命中索引，但是产品经理一定要前后模糊匹配呢？全文索引fulltext可以尝试一下，但Elasticsearch才是终极武器。

Join优化

join的实现是采用Nested Loop Join算法，就是通过驱动表的结果集作为基础数据，通过该结数据作为过滤条件到下一个表中循环查询数据，然后合并结果。如果有多个join，则将前面的结果集作为循环数据，再次到后一个表中查询数据。

  1. 驱动表和被驱动表尽可能增加查询条件，满足ON的条件而少用Where，用小结果集驱动大结果集。
  2. 被驱动表的join字段上加上索引，无法建立索引的时候，设置足够的Join Buffer Size。
  3. 禁止join连接三个以上的表，尝试增加冗余字段。

Limit优化

limit用于分页查询时越往后翻性能越差，解决的原则：缩小扫描范围 ，如下所示：

1
sql复制代码select * from orders order by id desc limit 100000,10 耗时0.4秒select * from orders order by id desc limit 1000000,10耗时5.2秒

先筛选出ID缩小查询范围，写法如下：

1
sql复制代码select * from orders where id > (select id from orders order by id desc  limit 1000000, 1) order by id desc limit 0,10耗时0.5秒

如果查询条件仅有主键ID，写法如下：

1
sql复制代码select id from orders where id between 1000000 and 1000010 order by id desc耗时0.3秒

如果以上方案依然很慢呢？只好用游标了，感兴趣的朋友阅读JDBC使用游标实现分页查询的方法

其他数据库

作为一名后端开发人员，务必精通作为存储核心的MySQL或SQL Server，也要积极关注NoSQL数据库，他们已经足够成熟并被广泛采用，能解决特定场景下的性能瓶颈。

分类 数据库 特性
键值型 Memcache 用于内容缓存，大量数据的高访问负载
键值型 Redis 用于内容缓存，比Memcache支持更多的数据类型，并能持久化数据
列式存储 HBase Hadoop体系的核心数据库，海量结构化数据存储，大数据必备。
文档型 MongoDb 知名文档型数据库，也可以用于缓存
文档型 CouchDB Apache的开源项目，专注于易用性，支持REST API
文档型 SequoiaDB 国内知名文档型数据库
图形 Neo4J 用于社交网络构建关系图谱，推荐系统等

本文转载自: 掘金

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

发表于

这是我参与11月更文挑战的第1天，活动详情查看：2021最后一次更文挑战

关于封面：晚饭后回自习室的路上

Easyexcel 官方文档

Easyexcel | github

本文源码

前言

最近也是在写的一个小练习中，需要用到这个。趁着这次就将写个整合的Demo给大家。

希望能够让大家有所收获。

阅读完本文，我想你对于使用Java配合Easyexcel操作Excel是完全没有问题的啦。

一、环境准备

1.1、导入相关依赖依赖

我使用Easyexcel的jar包是2021年10月的，说一句是最新版本，莫问题吧😁

easyexcel | maven

image.png

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
xml复制代码<!-- https://mvnrepository.com/artifact/com.alibaba/easyexcel -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>easyexcel</artifactId>
    <version>3.0.2</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.74</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
</dependency>
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.12</version>
</dependency>

1.2、项目结构

搭建个项目大家都会啦，这里放一下我自己的结构。

image-20211031194958702

二、读Excel操作 readExcel

2.1、前期准备

准备好一个xslx文件模板，我就是准备了我自己了。

image-20211031195724034

我们创建一个实体类，来对应xlsx中的列名。

实体类

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
java复制代码@Data
public class DemoModel {
	
    /**
     * 用名字去匹配，这里需要注意，如果名字重复，会导致只有一个字段读取到数据,所以我们常用下面这样的格式来写确定。
     */
    @ExcelProperty(value = "博客名", index = 0)
    private String name;
	
    
    @ExcelProperty(value = "社区", index = 1)
    private String communityName;

    @ExcelProperty(value = "主页", index = 2)
    private String homePageUrl;

    @ExcelProperty(value = "涉及领域", index = 3)
    private String specialty;

    @ExcelProperty(value = "联系邮箱", index = 4)
    private String email;

    /**
     * 这里用string 去接日期才能格式化。我想接收年月日格式
     */
    @DateTimeFormat("yyyy-MM-dd HH:mm:ss")
    @ExcelProperty(value = "发布的第一篇原创文章", index = 5)
    private String startDate;
}

监听器

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
java复制代码/**
 * 有个很重要的点 DocumentListener 不能被spring管理，要每次读取excel都要new,然后里面用到spring可以构造方法传进去
 * @author crush
 */
public class DemoListener extends AnalysisEventListener<DemoModel> {

    private static final Logger LOGGER = LoggerFactory.getLogger(DemoListener.class);
    

    /**
     * 每隔5条存储数据库，实际使用中可以3000条，然后清理list ，方便内存回收
     */
    private static final int BATCH_COUNT = 10;

    List<DemoModel> list = new ArrayList<DemoModel>();

    /**
     * 假设这个是一个DAO，当然有业务逻辑这个也可以是一个service。当然如果不用存储这个对象没用。
     */
    private DemoMapper demoMapper;

    public DemoListener() {
        // 这里是demo，所以随便new一个。实际使用如果到了spring,请使用下面的有参构造函数
        demoMapper = new DemoMapper();
    }

    /**
     * 如果使用了spring,请使用这个构造方法。每次创建Listener的时候需要把spring管理的类传进来
     *
     * @param demoDAO
     */
    public DemoListener(DemoMapper demoMapper) {
        this.demoMapper = demoMapper;
    }
    /**
     * 这个每一条数据解析都会来调用
     *
     * @param data
     *            one row value. Is is same as {@link AnalysisContext#readRowHolder()}
     * @param context
     */
    @Override
    public void invoke(DemoModel data, AnalysisContext context) {
        LOGGER.info("解析到一条数据:{}", JSON.toJSONString(data));
        list.add(data);
        // 达到BATCH_COUNT了，需要去存储一次数据库，防止数据几万条数据在内存，容易OOM
        if (list.size() >= BATCH_COUNT) {
            saveData();
            // 存储完成清理 list
            list.clear();
        }
    }

    /**
     * 所有数据解析完成了 都会来调用
     *
     * @param context
     */
    @Override
    public void doAfterAllAnalysed(AnalysisContext context) {
        // 这里也要保存数据，确保最后遗留的数据也存储到数据库
        saveData();
        LOGGER.info("所有数据解析完成！"+count);
    }

    /**
     * 加上存储数据库
     */
    private void saveData() {
        LOGGER.info("{}条数据，开始存储数据库！", list.size());
        //进行数据库层面操作
        demoMapper.save(list);
        LOGGER.info("存储数据库成功！");
    }
}

mapper层:此处只是模拟

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码/**
 * @Author: crush
 * @Date: 2021-10-31 11:39
 * version 1.0
 */
@Repository
public class DemoMapper {

    public void save(List<DemoModel> demoModels){
        System.out.println("Mapper:"+demoModels);
    }
}

类型转换

2.2、读单个Sheet

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
java复制代码/**
     * demo1 最简单的读
     * 我们先操作单个的Sheet,进行读操作
     * * <p>1. 创建excel对应的实体对象 参照{@link QuestionModel}
     * * <p>2. 由于默认一行行的读取excel，所以需要创建excel一行一行的回调监听器，参照{@link DocumentListener}
     * * <p>3. 直接读即可
     */
@Test
public void demo1TestRead() {
    // 有个很重要的点 DemoDataListener 不能被spring管理，要每次读取excel都要new,然后里面用到spring可以构造方法传进去
    // 写法1：
    String fileName = "E:\\project_code\\commons-utils\\springboot-excel\\src\\main\\resources\\excel\\demo.xlsx";
    // 这里 需要指定读用哪个class去读，然后读取第一个sheet 文件流会自动关闭
    EasyExcel.read(fileName, DemoModel.class, new DemoListener()).sheet().doRead();
}

这里我们无需指定sheet，因为我们就一个工作表，直接默认就完事了。

控制台输出：

1
2
3
4
5
6
cmd复制代码23:35:52.583 [main] INFO com.crush.excel.listener.DemoListener - 解析到一条数据:{"communityName":"掘金","email":"nzc_wyh@163.com","homePageUrl":"https://juejin.cn/user/2859142558267559","name":"宁在春","specialty":"Java后端开发","startDate":"2021-10-31 12:01:52"}
23:35:52.584 [main] INFO com.crush.excel.listener.DemoListener - 解析到一条数据:{"communityName":"CSDN","email":"nzc_wyh@163.com","homePageUrl":"https://blog.csdn.net/weixin_45821811?spm=1000.2115.3001.5343","name":"宁在春","specialty":"Java后端开发","startDate":"2020-05-11 12:01:52"}
23:35:52.585 [main] INFO com.crush.excel.listener.DemoListener - 2条数据，开始存储数据库！
Mapper:[DemoModel(name=宁在春, communityName=掘金, homePageUrl=https://juejin.cn/user/2859142558267559, specialty=Java后端开发, email=nzc_wyh@163.com, startDate=2021-10-31 12:01:52), DemoModel(name=宁在春, communityName=CSDN, homePageUrl=https://blog.csdn.net/weixin_45821811?spm=1000.2115.3001.5343, specialty=Java后端开发, email=nzc_wyh@163.com, startDate=2020-05-11 12:01:52)]
23:35:52.617 [main] INFO com.crush.excel.listener.DemoListener - 存储数据库成功！
23:35:52.618 [main] INFO com.crush.excel.listener.DemoListener - 所有数据解析完成！2

这是最简单的方式，也是读取Excel中，单个Sheet的操作，但我们平时中，一个excel是会有多个工作表的。

如下：

2.3、读多个Sheet

image-20211031204647817

其实本质上还是操作单个sheet工作表。

因为各个Sheet的数据不同，因而要建立多个Model和多个监听类。

这边就不再把全部model贴出了，如果不太明白，可以去文末看看源码。

image-20211031211039694

同时也要创建多个监听器。

image-20211031213105078

mapper层在这里就省略了哈。

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
java复制代码/**
     * 读多个或者全部sheet,这里注意一个sheet不能读取多次，多次读取需要重新读取文件
     * <p>
     * 1. 创建excel对应的实体对象 参照{@link QuestionModel}
     * <p>
     * 2. 由于默认一行行的读取excel，所以需要创建excel一行一行的回调监听器，参照{@link DocumentListener}
     * <p>
     * 3. 直接读即可
     */
@Test
public void repeatedRead() {
    String fileName = "E:\\project_code\\commons-utils\\springboot-excel\\src\\main\\resources\\excel\\Document_Success.xlsx";
    // 读取需要读取的Sheet
    ExcelReader excelReader = EasyExcel.read(fileName).build();
    //  自己使用功能必须不同的Listener
    ReadSheet readSheet1 =
        EasyExcel.readSheet(0).head(DocumentModel.class).registerReadListener(new DocumentListener()).build();
    ReadSheet readSheet2 =
        EasyExcel.readSheet(1).head(TemplateModel.class).registerReadListener(new TemplateListener()).build();

    ReadSheet readSheet3 =
        EasyExcel.readSheet(2).head(SectionModel.class).registerReadListener(new SectionListener()).build();

    ReadSheet readSheet4 =
        EasyExcel.readSheet(3).head(QuestionModel.class).registerReadListener(new QuestionListener()).build();

    ReadSheet readSheet5 =
        EasyExcel.readSheet(4).head(OptionModel.class).registerReadListener(new OptionListener()).build();

    ReadSheet readSheet6 =
        EasyExcel.readSheet(5).head(ConditionModel.class).registerReadListener(new ConditionListener()).build();

    ReadSheet readSheet7 =
        EasyExcel.readSheet(6).head(QuestionTooltipModel.class).registerReadListener(new QuestionTooltipListener()).build();

    ReadSheet readSheet8 =
        EasyExcel.readSheet(7).head(OptionTooltipModel.class).registerReadListener(new OptionTooltipListener()).build();

    // 这里注意 一定要把sheet1 sheet2 一起传进去，不然有个问题就是03版的excel 会读取多次，浪费性能
    excelReader.read(readSheet1, readSheet2, readSheet3, readSheet4, readSheet5, readSheet6, readSheet7, readSheet8);
    // 这里千万别忘记关闭，读的时候会创建临时文件，到时磁盘会崩的
    excelReader.finish();
}

控制台输出：

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
cmd复制代码23:42:19.733 [main] INFO com.crush.excel.listener.DocumentListener - 解析到一条数据:{"customerId":1,"name":"Document i"}

23:42:19.755 [main] INFO com.crush.excel.listener.TemplateListener - 解析到一条数据:{"documentType":1,"fontStyleId":1,"lobId":1,"name":"Template q","userGroupId":1}


23:42:19.785 [main] INFO com.crush.excel.listener.SectionListener - 解析到一条数据:{"index":1,"name":"Section 1","order":1}
23:42:19.785 [main] INFO com.crush.excel.listener.SectionListener - 解析到一条数据:{"index":2,"name":"Section 2","order":2}


23:42:19.823 [main] INFO com.crush.excel.listener.QuestionListener - 解析到一条数据:{"allowComments":"false","answerType":"1","enhancedField":"false","index":"1","order":"1","question":"This is Simple Question","required":"true","sectionIndex":"1"}
23:42:19.824 [main] INFO com.crush.excel.listener.QuestionListener - 解析到一条数据:{"allowComments":"false","answerType":"2","enhancedField":"false","index":"2","order":"2","question":"This is Simple Question","required":"false","sectionIndex":"1"}
23:42:19.825 [main] INFO com.crush.excel.listener.QuestionListener - 解析到一条数据:{"allowComments":"false","answerType":"10","enhancedField":"false","index":"3","order":"3","question":"This is Simple Question","required":"false","sectionIndex":"1"}
23:42:19.826 [main] INFO com.crush.excel.listener.QuestionListener - 解析到一条数据:{"allowComments":"false","answerType":"3","enhancedField":"false","index":"4","order":"1","question":"This is Simple Question","required":"true","sectionIndex":"2"}
23:42:19.827 [main] INFO com.crush.excel.listener.QuestionListener - 解析到一条数据:{"allowComments":"false","answerType":"5","enhancedField":"false","index":"5","order":"2","question":"This is Simple Question","required":"false","sectionIndex":"2"}
23:42:19.831 [main] INFO com.crush.excel.listener.QuestionListener - 解析到一条数据:{"allowComments":"false","answerType":"12","enhancedField":"false","index":"6","order":"3","question":"This is Simple Question","required":"true","sectionIndex":"2"}

23:42:19.847 [main] INFO com.crush.excel.listener.OptionListener - 解析到一条数据:{"index":1,"order":1,"questionIndex":3,"value":"Option 1"}
23:42:19.848 [main] INFO com.crush.excel.listener.OptionListener - 解析到一条数据:{"index":2,"order":2,"questionIndex":3,"value":"Option 2"}
23:42:19.849 [main] INFO com.crush.excel.listener.OptionListener - 解析到一条数据:{"index":3,"order":3,"questionIndex":3,"value":"Option 3"}
23:42:19.849 [main] INFO com.crush.excel.listener.OptionListener - 解析到一条数据:{"index":4,"order":4,"questionIndex":3,"value":"Option 4"}
23:42:19.850 [main] INFO com.crush.excel.listener.OptionListener - 解析到一条数据:{"index":5,"order":1,"questionIndex":6,"value":"Option 1"}
23:42:19.850 [main] INFO com.crush.excel.listener.OptionListener - 解析到一条数据:{"index":6,"order":2,"questionIndex":6,"value":"Option 2"}
23:42:19.851 [main] INFO com.crush.excel.listener.OptionListener - 解析到一条数据:{"index":7,"order":3,"questionIndex":6,"value":"Option 3"}

23:42:19.868 [main] INFO com.crush.excel.listener.ConditionListener - 解析到一条数据:{"action":1,"index":1,"questionIndex":1,"selectedAnswer":"Test","triggerQuestionIndex":1,"triggerSectionIndex":1}
23:42:19.869 [main] INFO com.crush.excel.listener.ConditionListener - 解析到一条数据:{"action":1,"index":2,"questionIndex":2,"selectedAnswer":"Test","triggerSectionIndex":2}



23:42:19.888 [main] INFO com.crush.excel.listener.QuestionTooltipListener - 解析到一条数据:{"endIndex":10,"index":1,"parentIndex":1,"startIndex":1,"text":"Sample Tooltip"}


23:42:19.908 [main] INFO com.crush.excel.listener.OptionTooltipListener - 解析到一条数据:{"endIndex":10,"index":1,"parentIndex":1,"startIndex":1,"text":"Sample Tooltip"}
23:42:19.910 [main] INFO com.crush.excel.listener.OptionTooltipListener - 解析到一条数据:{"endIndex":10,"index":2,"parentIndex":2,"startIndex":1,"text":"Sample Tooltip"}

我们可以在获取到数据的时候，存储进数据库。另外我们常常会上传excel，进行数据的导入，所以我们再来看看web中的excel的读。

2.4、web中的读

1
java复制代码@AutowiredDemoMapper demoMapper;/** * 文件上传 * <p> * 1. 创建excel对应的实体对象 参照{@link UploadData} * <p> * 2. 由于默认一行行的读取excel，所以需要创建excel一行一行的回调监听器，参照{@link UploadDataListener} * <p> * 3. 直接读即可 */@PostMapping("upload")@ResponseBodypublic String upload(MultipartFile file) throws IOException {    EasyExcel.read(file.getInputStream(), DemoModel.class, new DemoListener(demoMapper)).sheet().doRead();    return "success";}

image-20211101000304010

image-20211101000349426

我想这样的场景是我们在Web开发中最常用到的那种。

如果是内存足够，数据量不是特别大的话，像我这么写也是完全能够接受的。

如果是操作数据十万或几十万数据的大兄弟，可以前去官网看看优化方案。👉Easyexcel | github

三、写Excle操作 writeExcel

3.1、写到单个Sheet中

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复制代码/**
     * 最简单的写
     * <p>1. 创建excel对应的实体对象 参照{@link DemoModel}
     * <p>2. 直接写即可
     */
@Test
public void simpleWrite() {
    DemoModel model = new DemoModel();
    model.setName("宁在春");
    model.setCommunityName("知乎");
    model.setHomePageUrl("https://www.zhihu.com/creator/manage/creation/all");
    model.setEmail("nza_wyh@163.com");
    model.setSpecialty("SpringBoot");
    model.setStartDate("2021-10-31 12:00:00");
    DemoModel model2 = new DemoModel();
    model2.setName("宁在春2");
    model2.setCommunityName("知乎2");
    model2.setHomePageUrl("2https://www.zhihu.com/creator/manage/creation/all");
    model2.setEmail("nza_wyh@163.com2");
    model2.setSpecialty("SpringBoot2");
    model2.setStartDate("2021-10-31 12:00:00");
    List<DemoModel> models = new ArrayList<DemoModel>();
    models.add(model);
    models.add(model2);
    // 写法1
    String fileName = "E:\\project_code\\commons-utils\\springboot-excel\\src\\main\\resources\\excel\\demo2.xlsx";
    // 这里 需要指定写用哪个class去写，然后写到第一个sheet，名字为模板 然后文件流会自动关闭
    // 如果这里想使用03 则 传入excelType参数即可
    EasyExcel.write(fileName, DemoModel.class).sheet(0).doWrite(models);
    
    //        // 写法2
    //        // 这里 需要指定写用哪个class去写
    ExcelWriter excelWriter = EasyExcel.write(fileName, DemoModel.class).build();
    WriteSheet writeSheet = EasyExcel.writerSheet(0).build();
    excelWriter.write(models, writeSheet);
    // 千万别忘记finish 会帮忙关闭流
    excelWriter.finish();
}

这是操作excel文件中单个Sheet的操作，十分简单。

EasyExcel.write(fileName, DemoModel.class).sheet(0).doWrite(models);关于sheet()中的参数0，我们其实就一张Sheet工作表，填与不填其实都一样，另外，此处也可以填sheet表名字，也是一样 。

操作结果：

image-20211101004643750

注意：

注意：我发现如果我是直接向这个excel文件进行写入，默认是采用覆盖的方式进行写入，即之前有的信息都会被覆盖掉。

测试：

写之前：

image-20211101004951295

写之后：

image-20211101005021227

补充：

如果我们直接写，是采用覆盖模式的，这肯定是不符合一些业务场景的。所以肯定有解决方式的，

easyexcel中对此也是有处理的。

它有一个根据模板写入，模板如下：

image-20211101010819547

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
java复制代码/**
     * 根据模板写入
     * <p>1. 创建excel对应的实体对象 参照
     * <p>2. 使用 注解指定写入的列
     * <p>3. 使用withTemplate 写取模板
     * <p>4. 直接写即可
     */
@Test
public void templateWrite() {
    DemoModel model = new DemoModel();
    model.setName("宁在春");
    model.setCommunityName("知乎");
    model.setHomePageUrl("https://www.zhihu.com/creator/manage/creation/all");
    model.setEmail("nza_wyh@163.com");
    model.setSpecialty("SpringBoot");
    model.setStartDate("2021-10-31 12:00:00");
    DemoModel model2 = new DemoModel();
    model2.setName("宁在春2");
    model2.setCommunityName("知乎2");
    model2.setHomePageUrl("2https://www.zhihu.com/creator/manage/creation/all");
    model2.setEmail("nza_wyh@163.com2");
    model2.setSpecialty("SpringBoot2");
    model2.setStartDate("2021-10-31 12:00:00");
    List<DemoModel> models = new ArrayList<DemoModel>();
    models.add(model);
    models.add(model2);
    String templateFileName = "E:\\project_code\\commons-utils\\springboot-excel\\src\\main\\resources\\excel\\demo2.xlsx";
    String fileName = "E:\\project_code\\commons-utils\\springboot-excel\\src\\main\\resources\\excel\\templateWrite.xlsx";
    // 这里 需要指定写用哪个class去写，然后写到第一个sheet，名字为模板 然后文件流会自动关闭
    EasyExcel.write(fileName, DemoModel.class).withTemplate(templateFileName).sheet().doWrite(models);
}

最后的效果如下：

image-20211101011041698

写到注意那一小节的时候，我也有考虑这个是如何进行处理的。

我最开始想的是，是先把文件读出来，然后把数据拼接起来，再写进去。但是我一下就推翻这个想法，一旦那样做，内存消耗什么的都太大了，不太合适。

然后简单看了一下，它是直接new了一个File，加载进内存。（觉得自己好憨）

因为这个我又去测试了个不同的，如果把模板修改的不符合会咋样。

image-20211101085113831

测试的结果就是符合我的猜测，就是copy了一份原文件，在源文件的基础上进行写操作，不管原文件格式如何，都会进行保留。

image-20211101085258579

3.2、重复写入或写到多个Sheet中

重复写入：

其实就是循环了单个的操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码// 方法1 如果写到同一个sheet
    String fileName = "E:\\project_code\\commons-utils\\springboot-excel\\src\\main\\resources\\excel\\demo2.xlsx";
//        // 这里 需要指定写用哪个class去写
ExcelWriter excelWriter = EasyExcel.write(fileName, DemoModel.class).build();
WriteSheet writeSheet = EasyExcel.writerSheet(0).build();
// 去调用写入,这里我调用了五次，实际使用时根据数据库分页的总的页数来
for (int i = 0; i < 5; i++) {
    // 分页去数据库查询数据 这里可以去数据库查询每一页的数据
    List<DemoModel> data = data();
	excelWriter.write(data, writeSheet);
}
// 千万别忘记finish 会帮忙关闭流
excelWriter.finish();

data()就是生成数据的一个方法。

写到多个Sheet中：

其本质也是上面那样，只是调用不同监听类，不同的对象罢了，更简单的方式，以做到不同的处理，目前还没有找到更简单的方法。

和读多个sheet中一样的方式，在此处就不再复述了。

3.3、web中的写

既然有写就要也有读了，有始有终，这个也给大家贴出来了。

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
java复制代码/**
     * 文件下载（失败了会返回一个有部分数据的Excel）
     * <p>
     * 1. 创建excel对应的实体对象 参照
     * <p>
     * 2. 设置返回的 参数
     * <p>
     * 3. 直接写，这里注意，finish的时候会自动关闭OutputStream,当然你外面再关闭流问题不大
     */
@GetMapping("download")
public void download(HttpServletResponse response) throws IOException {
    // 这里注意 有同学反应使用swagger 会导致各种问题，请直接用浏览器或者用postman
    response.setContentType("application/vnd.ms-excel");
    response.setCharacterEncoding("utf-8");
    // 这里URLEncoder.encode可以防止中文乱码 当然和easyexcel没有关系
    String fileName = URLEncoder.encode("测试", "UTF-8");
    response.setHeader("Content-disposition", "attachment;filename=" + fileName + ".xlsx");
    //这里的sheet的参数就是生成后的工作表的名称
    EasyExcel.write(response.getOutputStream(), DemoModel.class).sheet("模板").doWrite(data());
}

/**
     * 文件下载并且失败的时候返回json（默认失败了会返回一个有部分数据的Excel）
     *
     * @since 2.1.1
     */
@GetMapping("downloadFailedUsingJson")
public void downloadFailedUsingJson(HttpServletResponse response) throws IOException {
    // 这里注意 有同学反应使用swagger 会导致各种问题，请直接用浏览器或者用postman
    try {
        response.setContentType("application/vnd.ms-excel");
        response.setCharacterEncoding("utf-8");
        // 这里URLEncoder.encode可以防止中文乱码 当然和easyexcel没有关系
        String fileName = URLEncoder.encode("测试", "UTF-8");
        response.setHeader("Content-disposition", "attachment;filename=" + fileName + ".xlsx");
        // 这里需要设置不关闭流
        EasyExcel.write(response.getOutputStream(), DemoModel.class).autoCloseStream(Boolean.FALSE).sheet("模板")
            .doWrite(data());
    } catch (Exception e) {
        // 重置response
        response.reset();
        response.setContentType("application/json");
        response.setCharacterEncoding("utf-8");
        Map<String, String> map = new HashMap<String, String>();
        map.put("status", "failure");
        map.put("message", "下载文件失败" + e.getMessage());
        response.getWriter().println(JSON.toJSONString(map));
    }
}

private List<DemoModel> data(){
    List<DemoModel> demoModels = new ArrayList<>();
    DemoModel model = new DemoModel();
    model.setName("宁在春");
    model.setCommunityName("知乎");
    model.setHomePageUrl("https://www.zhihu.com/creator/manage/creation/all");
    model.setEmail("nza_wyh@163.com");
    model.setSpecialty("SpringBoot");
    model.setStartDate("2021-10-31 12:00:00");
    DemoModel model2 = new DemoModel();
    model2.setName("宁在春2");
    model2.setCommunityName("知乎2");
    model2.setHomePageUrl("2https://www.zhihu.com/creator/manage/creation/all");
    model2.setEmail("nza_wyh@163.com2");
    model2.setSpecialty("SpringBoot2");
    model2.setStartDate("2021-10-31 12:00:00");
    demoModels.add(model);
    demoModels.add(model2);
    return demoModels;
}

结果：

image-20211101012029439

四、可能会产生的问题

  1. 在实体中已经使用了@Data注解,就不要再使用@Accessors(chain = true)，否则会产生读取为空的现象。

这一点我已经亲自踩坑了（一开始排这个错都排了会😂）

image-20211101012556789
2. 在读的时候Listener里面需要使用spring的@Autowired

给Listener创建成员变量，然后在构造方法里面传进去。必须不让spring管理Listener，每次读取都要new一个。

这两点是我认为十分常见，又比较容易踩的坑，给大家列出来了。

详细的可点击👉 Easexcel 常见问题

五、自言自语

11月更文开始，每天写写项目，做做记录，充实的每一天。

纸上得来终觉浅，绝知此事要躬行。

大家好，我是博主宁在春：主页

一名喜欢文艺却踏上编程这条道路的小青年。

希望：我们，待别日相见时，都已有所成。

一个可爱的猫咪小表情，实在太可爱了，没办法抵挡住诱惑啊。

image-20211101090600699

本文转载自: 掘金

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

发表于

这是我参与11月更文挑战的第1天，活动详情查看：2021最后一次更文挑战

最近，业务增长的很迅猛，对于我们后台这块也是一个不小的挑战，这次遇到的核心业务接口的性能瓶颈，并不是单独的一个问题导致的，而是几个问题揉在一起：我们解决一个之后，发上线，之后发现还有另一个的性能瓶颈问题。这也是我经验不足，导致没能一下子定位解决；而我又对我们后台整个团队有着固执的自尊，不想通过大量水平扩容这种方式挺过压力高峰，导致线上连续几晚都出现了不同程度的问题，肯定对于我们的业务增长是有影响的。这也是我不成熟和要反思的地方。这系列文章主要记录下我们针对这次业务增长，对于我们后台微服务系统做的通用技术优化，针对业务流程和缓存的优化由于只适用于我们的业务，这里就不再赘述了。本系列会分为如下几篇：

  1. 改进客户端负载均衡算法
  2. 开发日志输出异常堆栈的过滤插件
  3. 针对 x86 云环境改进异步日志等待策略
  4. 增加对于同步微服务的 HTTP 请求等待队列的监控以及云上部署，需要小心达到实例网络流量上限导致的请求响应缓慢
  5. 针对系统关键业务增加必要的侵入式监控

改进客户端负载均衡算法

Spring Cloud LoadBalancer 内置轮询算法以及问题

我们是用 Spring Cloud 作为我们的微服务体系，并且针对其中很多组件做了优化改造，请参考我的另一系列。之前我们的客户端负载均衡算法，是不同请求之间相互独立的轮询。由于我们实现的微服务框架会针对可以重试的请求进行重试，重试需要重试与之前不同的实例。没有重试，无法实现在线发布对于用户无感知，并且我们部署同一个微服务的不同实例是处于不同的可用区，并且微服务不会每次都全部出问题，而是某些实例出问题，有重试可以让某些实例出问题的时候，对用户无感知。当某些实例压力过大时，重试也可以让请求重试压力比较小的实例。使用 Spring Cloud LoadBalancer 的内置的负载均衡算法均无法满足我们的需求，所以我们针对其中的轮询算法进行了改进。原有的流程是：

  1. 获取服务实例列表
  2. 所有线程共用同一个原子变量 position，每次请求原子加 1
  3. position 对实例个数取余，返回对应下标的实例进行调用

这样的算法问题是：假设有微服务 A 有两个实例：实例 1 和实例 2。请求 A 到达时，RoundRobinLoadBalancer 返回实例 1，这时有请求 B 到达，RoundRobinLoadBalancer 返回实例 2。然后如果请求 A 失败重试，RoundRobinLoadBalancer 又返回了实例 1。这不是我们期望看到的。

本次优化前，我们的负载均衡算法以及问题

在本次业务突增很多的改进之前，我们第一版改进后的流程是：

  1. 获取服务实例列表，将实例列表按照 ip 端口排序，如果不排序即使 position 是下一个可能也代表的是之前已经调用过的实例
  2. 根据请求中的 traceId，从本地缓存中以 traceId 为 key 获取一个初始值为随机数的原子变量 position，这样防止所有请求都从第一个实例开始调用，之后第二个、第三个这样。
  3. position 原子加一，之后对实例个数取余，返回对应下标的实例进行调用

其中请求包含 traceId 是来自于我们使用了 spring-cloud-sleuth 链路追踪，基于这种机制我们能保证请求不会重试到之前已经调用过的实例。源码是：

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
kotlin复制代码//一定必须是实现ReactorServiceInstanceLoadBalancer
//而不是ReactorLoadBalancer<ServiceInstance>
//因为注册的时候是ReactorServiceInstanceLoadBalancer
@Log4j2
public class RoundRobinWithRequestSeparatedPositionLoadBalancer implements ReactorServiceInstanceLoadBalancer {
    private final ServiceInstanceListSupplier serviceInstanceListSupplier;
    //每次请求算上重试不会超过1分钟
    //对于超过1分钟的，这种请求肯定比较重，不应该重试
    private final LoadingCache<Long, AtomicInteger> positionCache = Caffeine.newBuilder().expireAfterWrite(1, TimeUnit.MINUTES)
            //随机初始值，防止每次都是从第一个开始调用
            .build(k -> new AtomicInteger(ThreadLocalRandom.current().nextInt(0, 1000)));
    private final String serviceId;
    private final Tracer tracer;


    public RoundRobinWithRequestSeparatedPositionLoadBalancer(ServiceInstanceListSupplier serviceInstanceListSupplier, String serviceId, Tracer tracer) {
        this.serviceInstanceListSupplier = serviceInstanceListSupplier;
        this.serviceId = serviceId;
        this.tracer = tracer;
    }
    
    //每次重试，其实都会调用这个 choose 方法重新获取一个实例
    @Override
    public Mono<Response<ServiceInstance>> choose(Request request) {
        return serviceInstanceListSupplier.get().next().map(serviceInstances -> getInstanceResponse(serviceInstances));
    }

    private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> serviceInstances) {
        if (serviceInstances.isEmpty()) {
            log.warn("No servers available for service: " + this.serviceId);
            return new EmptyResponse();
        }
        return getInstanceResponseByRoundRobin(serviceInstances);
    }

    private Response<ServiceInstance> getInstanceResponseByRoundRobin(List<ServiceInstance> serviceInstances) {
        if (serviceInstances.isEmpty()) {
            log.warn("No servers available for service: " + this.serviceId);
            return new EmptyResponse();
        }
        //为了解决原始算法不同调用并发可能导致一个请求重试相同的实例
        //从 sleuth 的 Tracer 中获取当前请求的上下文
        Span currentSpan = tracer.currentSpan();
        //如果上下文不存在，则可能不是前端用户请求，而是其他某些机制触发，我们就创建一个新的上下文
        if (currentSpan == null) {
            currentSpan = tracer.newTrace();
        }
        //从请求上下文中获取请求的 traceId，用来唯一标识一个请求
        long l = currentSpan.context().traceId();
        AtomicInteger seed = positionCache.get(l);
        int s = seed.getAndIncrement();
        int pos = s % serviceInstances.size();
        log.info("position {}, seed: {}, instances count: {}", pos, s, serviceInstances.size());
        return new DefaultResponse(serviceInstances.stream()
                //实例返回列表顺序可能不同，为了保持一致，先排序再取
                .sorted(Comparator.comparing(ServiceInstance::getInstanceId))
                .collect(Collectors.toList()).get(pos));
    }
}

但是在这次请求突增很多的时候，这种负载均衡算法还是给我们带来了问题。

首先，本次突增，我们并没有采取扩容，导致本次的性能压力对于压力的均衡分布非常敏感。举个例子是，假设微服务 A 有 9 个实例，在业务高峰点来的时候，最理想的情况是保证无论何时这 9 个负载压力都完全均衡，但是由于我们使用了初始值为随机数的原子变量 position，虽然从一天的总量上来看，负责均衡压力肯定是均衡，但是在某一小段时间内，很可能压力全都跑到了某几个实例上，导致这几个实例被压垮，熔断，然后又都跑到了另外的几个实例上，又被压垮，熔断，如此恶性循环。

然后，我们部署采用的是 k8s 部署，同一个虚拟机上面可能会跑很多微服务的 pod。在某些情况下，同一个微服务的多个 pod 可能会跑到同一个虚拟机 Node 上，这个可以从pod 的 ip 网段上看出来：例如某个微服务有如下 7 个实例：10.238.13.12:8181,10.238.13.24:8181,10.238.15.12:8181,10.238.17.12:8181,10.238.20.220:8181,10.238.21.31:8181,10.238.21.121:8181，那么 10.238.13.12:8181 与 10.238.13.24:8181 很可能在同一个 Node 上，10.238.21.31:8181 和 10.238.21.121:8181 很可能在同一个 Node 上。我们重试，需要优先重试与之前重试过的实例尽量不在同一个 Node 上的实例，因为同一个 Node 上的实例只要有一个有问题或者压力过大，其他的基本上也有问题或者压力过大。

最后，如果调用某个实例一直失败，那么这个实例的调用优先级需要排在其他正常的实例后面。这个对于减少快速刷新发布（一下子启动很多实例之后停掉多个老实例，实例个数大于重试次数配置）对于用户的影响，以及某个可用区突然发生异常导致多个实例下线对用户的影响，以及业务压力已经过去，压力变小后，需要关掉不再需要的实例，导致大量实例发生迁移的时候对用户的影响，有很大的作用。

针对以上问题的优化方案

我们针对上面三个问题，提出了一种优化后的解决方案：

  1. 针对每次请求，记录：
  2. 本次请求已经调用过哪些实例 -> 请求调用过的实例缓存
  3. 调用的实例，当前有多少请求在处理中 -> 实例运行请求数
  4. 调用的实例，最近请求错误率 -> 实例请求错误率
  5. 随机将实例列表打乱，防止在以上三个指标都相同时，总是将请求发给同一个实例。
  6. 按照 当前请求没有调用过靠前 -> 错误率越小越靠前 的顺序排序 -> 实例运行请求数越小越靠前
  7. 取排好序之后的列表第一个实例作为本次负载均衡的实例

具体实现是：以下的代码来自于：github.com/JoJoTec/spr…

我们使用了依赖：

1
2
3
4
xml复制代码<dependency>
    <groupId>io.dropwizard.metrics</groupId>
    <artifactId>metrics-core</artifactId>
</dependency>

记录实例数据的缓存类：

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
java复制代码@Log4j2
public class ServiceInstanceMetrics {
	private static final String CALLING = "-Calling";
	private static final String FAILED = "-Failed";

	private MetricRegistry metricRegistry;

	ServiceInstanceMetrics() {
	}

	public ServiceInstanceMetrics(MetricRegistry metricRegistry) {
		this.metricRegistry = metricRegistry;
	}

	/**
	 * 记录调用实例
	 * @param serviceInstance
	 */
	public void recordServiceInstanceCall(ServiceInstance serviceInstance) {
		String key = serviceInstance.getHost() + ":" + serviceInstance.getPort();
		metricRegistry.counter(key + CALLING).inc();
	}
	/**
	 * 记录调用实例结束
	 * @param serviceInstance
	 * @param isSuccess 是否成功
	 */
	public void recordServiceInstanceCalled(ServiceInstance serviceInstance, boolean isSuccess) {
		String key = serviceInstance.getHost() + ":" + serviceInstance.getPort();
		metricRegistry.counter(key + CALLING).dec();
		if (!isSuccess) {
			//不成功则记录失败
			metricRegistry.meter(key + FAILED).mark();
		}
	}

	/**
	 * 获取正在运行的调用次数
	 * @param serviceInstance
	 * @return
	 */
	public long getCalling(ServiceInstance serviceInstance) {
		String key = serviceInstance.getHost() + ":" + serviceInstance.getPort();
		long count = metricRegistry.counter(key + CALLING).getCount();
		log.debug("ServiceInstanceMetrics-getCalling: {} -> {}", key, count);
		return count;
	}

	/**
	 * 获取最近一分钟调用失败次数分钟速率，其实是滑动平均数
	 * @param serviceInstance
	 * @return
	 */
	public double getFailedInRecentOneMin(ServiceInstance serviceInstance) {
		String key = serviceInstance.getHost() + ":" + serviceInstance.getPort();
		double rate = metricRegistry.meter(key + FAILED).getOneMinuteRate();
		log.debug("ServiceInstanceMetrics-getFailedInRecentOneMin: {} -> {}", key, rate);
		return rate;
	}
}

负载均衡核心代码：

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
kotlin复制代码private final LoadingCache<Long, Set<String>> calledIpPrefixes = Caffeine.newBuilder()
        .expireAfterAccess(3, TimeUnit.MINUTES)
        .build(k -> Sets.newConcurrentHashSet());
private final String serviceId;
private final Tracer tracer;
private final ServiceInstanceMetrics serviceInstanceMetrics;

//每次重试，其实都会调用这个 choose 方法重新获取一个实例
@Override
public Mono<Response<ServiceInstance>> choose(Request request) {
    Span span = tracer.currentSpan();
    return serviceInstanceListSupplier.get().next()
            .map(serviceInstances -> {
                //保持 span 和调用 choose 的 span 一样
                try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) {
                    return getInstanceResponse(serviceInstances);
                }
            });
}


private Response<ServiceInstance> getInstanceResponse(List<ServiceInstance> serviceInstances) {
    if (serviceInstances.isEmpty()) {
        log.warn("No servers available for service: " + this.serviceId);
        return new EmptyResponse();
    }
    //读取 spring-cloud-sleuth 的对于当前请求的链路追踪上下文，获取对应的 traceId
    Span currentSpan = tracer.currentSpan();
    if (currentSpan == null) {
        currentSpan = tracer.newTrace();
    }
    long l = currentSpan.context().traceId();
    return getInstanceResponseByRoundRobin(l, serviceInstances);
}

@VisibleForTesting
public Response<ServiceInstance> getInstanceResponseByRoundRobin(long traceId, List<ServiceInstance> serviceInstances) {
    //首先随机打乱列表中实例的顺序
    Collections.shuffle(serviceInstances);
    //需要先将所有参数缓存起来，否则 comparator 会调用多次，并且可能在排序过程中参数发生改变（针对实例的请求统计数据一直在并发改变）
    Map<ServiceInstance, Integer> used = Maps.newHashMap();
    Map<ServiceInstance, Long> callings = Maps.newHashMap();
    Map<ServiceInstance, Double> failedInRecentOneMin = Maps.newHashMap();
    serviceInstances = serviceInstances.stream().sorted(
            Comparator
                    //之前已经调用过的网段，这里排后面
                    .<ServiceInstance>comparingInt(serviceInstance -> {
                        return used.computeIfAbsent(serviceInstance, k -> {
                            return calledIpPrefixes.get(traceId).stream().anyMatch(prefix -> {
                                return serviceInstance.getHost().contains(prefix);
                            }) ? 1 : 0;
                        });
                    })
                    //当前错误率最少的
                    .thenComparingDouble(serviceInstance -> {
                        return failedInRecentOneMin.computeIfAbsent(serviceInstance, k -> {
                            double value = serviceInstanceMetrics.getFailedInRecentOneMin(serviceInstance);
                            //由于使用的是移动平均值（EMA），需要忽略过小的差异（保留两位小数，不是四舍五入，而是直接舍弃）
                            return ((int) (value * 100)) / 100.0;
                        });
                    })
                    //当前负载请求最少的
                    .thenComparingLong(serviceInstance -> {
                        return callings.computeIfAbsent(serviceInstance, k ->
                                serviceInstanceMetrics.getCalling(serviceInstance)
                        );
                    })
    ).collect(Collectors.toList());
    if (serviceInstances.isEmpty()) {
        log.warn("No servers available for service: " + this.serviceId);
        return new EmptyResponse();
    }
    ServiceInstance serviceInstance = serviceInstances.get(0);
    //记录本次返回的网段
    calledIpPrefixes.get(traceId).add(serviceInstance.getHost().substring(0, serviceInstance.getHost().lastIndexOf(".")));
    //目前记录这个只为了兼容之前的单元测试（调用次数测试）
    positionCache.get(traceId).getAndIncrement();
    return new DefaultResponse(serviceInstance);
}

一些组内关于方案设计的取舍 Q&A

1. 为何没有使用所有微服务共享的缓存来保存调用数据，来让这些数据更加准确？

共享缓存的可选方案包括将这些数据记录放入 Redis，或者是 Apache Ignite 这样的内存网格中。但是有两个问题：

  1. 如果数据记录放入 Redis 这样的额外存储，如果 Redis 不可用会导致所有的负载均衡都无法执行。如果放入 Apache Ignite，如果对应的节点下线，那么对应的负载均衡也无法执行。这些都是不能接受的。
  2. 假设微服务 A 需要调用微服务 B，可能 A 的某个实例调用 B 的某个实例有问题，但是 A 的其他实例调用 B 的这个实例却没有问题，例如当某个可用区与另一个可用区网络拥塞的时候。如果用同一个缓存 Key 记录 A 所有的实例调用 B 这个实例的数据，显然是不准确的。

每个微服务使用本地缓存，记录自己调用其他实例的数据，在我们这里看来，不仅是更容易实现，也是更准确的做法。

2. 采用 EMA 的方式而不是请求窗口的方式统计最近错误率

采用请求窗口的方式统计，肯定是最准确的，例如我们统计最近一分钟的错误率，就将最近一分钟的请求缓存起来，读取的时候，将缓存起来的请求数据加在一起取平均数即可。但是这种方式在请求突增的时候，可能会占用很多很多内存来缓存这些请求。同时计算错误率的时候，随着缓存请求数的增多也会消耗更大量的 CPU 进行计算。这样做很不值得。

EMA 这种滑动平均值的计算方式，常见于各种性能监控统计场景，例如 JVM 中 TLAB 大小的动态计算，G1 GC Region 大小的伸缩以及其他很多 JVM 需要动态得出合适值的地方，都用这种计算方式。他不用将请求缓存起来，而是直接用最新值乘以一个比例之后加上老值乘以 （1 - 这个比例），这个比例一般高于 0.5，表示 EMA 和当前最新值更加相关。

但是 EMA 也带来另一个问题，我们会发现随着程序运行小数点位数会非常多，会看到类似于如下的值：0.00000000123, 0.120000001, 0.120000003, 为了忽略过于细致差异的影响（其实这些影响也来自于很久之前的错误请求），我们只保留两位小数进行排序。

微信搜索“我的编程喵”关注公众号，每日一刷，轻松提升技术，斩获各种offer

本文转载自: 掘金

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

发表于

前言

本篇将分析CyclicBarrier的源码，分析结束后，会用一个示例展示CyclicBarrier，并比较CyclicBarrier和CountDownLatch的区别。

1、CyclicBarrier的简介

CyclicBarrier允许一组线程在触发屏障之前相互等待，直到达到某一条件才继续执行；CyclicBarrier在这些线程释放后，又可以重新使用，所以也称循环栅栏。

2、分析源码

2.1、构造方法

1
2
3
4
5
6
7
8
9
10
11
java复制代码//定义在触发屏障之前必须调用的线程数
public CyclicBarrier(int parties) {
	this(parties, null);
}
public CyclicBarrier(int parties, Runnable barrierAction) {
	if (parties <= 0) throw new IllegalArgumentException();
	    this.parties = parties;
	    this.count = parties;
	    //定义最后一个线程到达屏障时执行的操作
	    this.barrierCommand = barrierAction;
}

2.2、await()方法

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
java复制代码public int await() throws InterruptedException, BrokenBarrierException {
	try {
		//调用dowait方法，不需要定义超时时间
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}

private int dowait(boolean timed, long nanos)
	throws InterruptedException, BrokenBarrierException,
		   TimeoutException {
	final ReentrantLock lock = this.lock;
	//获取“独占锁”
	lock.lock();
	try {
		//获取“当前代”
		final Generation g = generation;
		//“当前代”损坏，则抛出异常
		if (g.broken)
			throw new BrokenBarrierException();
		// 如果当前线程被中断
		if (Thread.interrupted()) {
			//终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程
			breakBarrier();
			throw new InterruptedException();
		}
		//计数
		int index = --count;
		//达到屏障（最后一个线程到达）
		//最后一个到达的线程不执行下面的for循环语句
		if (index == 0) {  // tripped
			boolean ranAction = false;
			try {
				final Runnable command = barrierCommand;
				if (command != null)
					//执行初始化传入的命令操作
					command.run();
				ranAction = true;
				//调用下一代方法
				nextGeneration();
				return 0;
			} finally {
				if (!ranAction)
					breakBarrier();
			}
		}

		// loop until tripped, broken, interrupted, or timed out
		//非最后一个到达的线程全部执行此语句，阻塞在trip.await()方法上
		for (;;) {
			try {
				//如果不是“超时等待”
				if (!timed)
					// 调用condition的await()方法
					trip.await();
				else if (nanos > 0L)
					nanos = trip.awaitNanos(nanos);
			} catch (InterruptedException ie) {
				// 如果等待过程中，线程被中断，则执行下面的函数
				if (g == generation && ! g.broken) {
					breakBarrier();
					throw ie;
				} else {
					// We're about to finish waiting even if we had not
					// been interrupted, so this interrupt is deemed to
					// "belong" to subsequent execution.
					Thread.currentThread().interrupt();
				}
			}

			if (g.broken)
				throw new BrokenBarrierException();
			// 如果“generation换代了”，则返回index。
			if (g != generation)
				return index;
			//检查超时
			if (timed && nanos <= 0L) {
				// 如果超时，则终止CyclicBarrier，并唤醒CyclicBarrier中所有等待线程
				breakBarrier();
				throw new TimeoutException();
			}
		}
	} finally {
		//释放“独占锁”，并唤醒AQS中等待的下一个线程
		lock.unlock();
	}
}

private void nextGeneration() {
    // 调用condition的signalAll()将其队列中的等待者全部转移到AQS的队列中
    trip.signalAll();
    // 重置count
    count = parties;
    // 进入下一代
    generation = new Generation();
}

3、CyclicBarrier与CountDownLatch的区别

  1. 两者都能实现阻塞一组线程，然后等待唤醒；
  2. 前者是最后一个线程到达直接唤醒，后者是调用countDown()方法；
  3. 前者是通过ReentrantLock的”独占锁”和Conditon来实现，后者是通过AQS的“共享锁”实现；
  4. 前者可以重复使用，后者只能使用一次；
  5. 前者只能实现多个线程到达后一起运行（多个条件成立才能一起运行）；
  6. 后者不仅可以实现一个线程等待多个线程（多个条件成立才能一起运行），还能实现多个线程等待一个线程（多个条件成立并且等待某个特殊信号才能一起运行）

4、示例

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复制代码public class CyclicBarrierTest {

    public static final CyclicBarrier WORK_THREAD = new CyclicBarrier(3);

    public static void main(String[] args) throws Exception{
        //主线程逻辑
        Thread.sleep(2000);
        for (int i=0; i<3; i++){
            String condition_name = "条件"+i;
            new Thread(() -> { method(); },condition_name).start();
        }
    }

    public static void method(){
        System.out.println("等待的条件是：" + Thread.currentThread().getName());
        try {
            WORK_THREAD.await();
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println("执行完的条件是：" + Thread.currentThread().getName());
    }
}

输出结果：

1
2
3
4
5
6
java复制代码等待的条件是：条件0
等待的条件是：条件1
等待的条件是：条件2
执行完的条件是：条件1
执行完的条件是：条件2
执行完的条件是：条件0

结束语

本篇介绍了CyclicBarrier，分析了源码，用了一段代码演示了使用方法，并比较了CyclicBarrier和CountDownLatch的区别。

JUC锁篇章到此就分析完了，下一篇将开启分析JUC集合。

如果你觉得本篇文章对你有帮助的话，请帮忙点个赞，再加一个关注。

本文转载自: 掘金

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

发表于

这是我参与11月更文挑战的第1天，活动详情查看：2021最后一次更文挑战。

悲观锁和乐观锁是用来解决并发问题的两种思想，在不同的平台有着各自的实现。例如在Java中，synchronized就可以认为是悲观锁的实现(不严谨，有锁升级的过程，升级到重量级锁才算)，Atomic***原子类可以认为是乐观锁的实现。

悲观锁
具有强烈的独占和排他特性，在整个处理过程中将数据处于锁定状态，一般是通过系统的互斥量来实现。当其他线程想要获取锁时会被阻塞，直到持有锁的线程释放锁。

乐观锁
对数据的修改和访问持乐观态度，假设不会发生冲突，只有当数据提交更新时才会对数据冲突与否进行检测，如果没有冲突则顺利提交更新，否则快速失败，返回一个错误给用户，让用户选择接下来该如何去做，一般来说失败后会继续重试，直到提交更新成功为止。

MySQL本身就支持锁机制，例如我们有一个「先查再写」的需求，我们希望整个流程是一个原子操作，中间不能被打断，这时候就可以通过给查询的数据行加「排他锁」来实现。只要当前事务不释放锁，其他事务要想获得排他锁，MySQL就会将其阻塞，直到当前事务释放锁。这种MySQL底层的排他锁就称作「悲观锁」。

MySQL本身不提供乐观锁的功能，需要开发者自己实现。普遍的做法是在表中加一个version列，用来标记数据行的版本，当我们需要更新数据时，必须比对version版本，version一致说明这个期间数据没有被其他事务修改过，否则说明数据已经被其他事务修改，需要自旋重试了。

实战

假设数据库有两张表：商品表和订单表。
用户下单后需要执行两个操作：

  1. 商品表减去库存。
  2. 订单表创建一条记录。

初始数据：ID为1的商品有100的库存，订单表数据为空。
客户端启动10个线程并发下单，分别在无锁、悲观锁、乐观锁的场景下有哪些表现。

如下是创建表的sql语句：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
sql复制代码-- 商品表
CREATE TABLE `goods` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `goods_name` varchar(50) NOT NULL,
  `price` decimal(10,2) NOT NULL,
  `stock` int(11) DEFAULT '0',
  `version` int(10) unsigned NOT NULL DEFAULT '0',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8

-- 订单表
CREATE TABLE `t_order` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `goods_id` bigint(20) NOT NULL,
  `order_time` datetime NOT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8

1、无锁

不做任何处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
java复制代码// 下单
private boolean order(){
    Goods goods = goodsMapper.selectById(1L);
    boolean success = false;
    if (goods.getStock() > 0) {
        goods.setStock(goods.getStock() - 1);
        // 更新库存
        goodsMapper.updateById(goods);
        // 创建订单
        orderMapper.save(goods.getId());
        success = true;
    }
    return success;
}

控制台输出结果：
在这里插入图片描述

2、悲观锁

查询商品时加FOR UPDATE，给数据行加排他锁，这样其他线程再查询时就会被阻塞，直到当前线程的事务提交并释放锁，其他线程才能继续下单。这种方式并发性能不高。

sql语句

1
2
java复制代码@Select("SELECT * FROM goods WHERE id = #{id} FOR UPDATE")
Goods selectForUpdate(Long id);

控制台输出结果：
在这里插入图片描述
注意：FOR UPDATE必须在事务中才有效，查询和更新必须在同一个事务中！！！

3、乐观锁

实现思路是：每次更新时校验版本号，如果版本号一致说明期间数据没有被其他线程改过，当前线程可以正常提交更新，否则说明数据已经被其他线程改过了，当前线程需要自旋重试，直到业务成功为止。
更新数据的同时版本号必须自增！！！

1
2
java复制代码@Update("UPDATE goods SET stock = #{stock},version = version+1 WHERE id = #{id} AND version = #{version}")
int updateByVersion(Long id, Integer stock, Integer version);

业务代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码boolean order(){
    Goods goods = goodsMapper.selectById(1L);
    boolean success = false;
    if (goods.getStock() > 0) {
        goods.setStock(goods.getStock() - 1);
        // 更新库存，带上版本号
        int result = goodsMapper.updateByVersion(goods.getId(), goods.getStock(), goods.getVersion());
        if (result <= 0) {
            // 更新失败，说明期间数据已经被其他线程修改，需要递归重试
            return order();
        }
        // 创建订单
        orderMapper.save(goods.getId());
        success = true;
    }
    return success;
}

控制台输出结果：
在这里插入图片描述

本文转载自: 掘金

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

发表于

「这是我参与11月更文挑战的第1天，活动详情查看：2021最后一次更文挑战」。

本文主要图解JAVA虚拟机的类加载机制，多图(20+)详解、抽丝剥茧。

类加载详解

Java类加载一览

程序运行的时候，class文件类通过类加载器ClassLoader把字节码加载到常量池，并进行校验、准备、解析和初始化。

jvm0.png

虚拟机黑盒操作了校验、准备、解析和初始化等过程。但加载不是虚拟机的事情，而是类加载器（ClassLoader类）的事情。

什么是类加载

类加载就是类加载器通过一个类的全限定名去获取这个类的二进制字节流到JAVA虚拟机。
如下图所示：

jvm1.png

从图上可以看出：

  • 类加载阶段的“通过一个类的全限定名去获取这个类的二进制字节流”的操作是放到虚拟机的外部去实现的
  • 类加载是让程序员自己去决定如何获取所需要的类。
  • 可帮助程序员实现这个加载动作的代码模块称为类加载器（ClassLoader）。

对于Classloader而言，加载其实要做三件事情：

  1. 通过一个类的全限定名来获取定义此类的二进制字节流。用字节流就很方便从网络获取，字节码的二进制流还可以运行时计算生成，动态代理技术Proxy就是用了ProxyGenerator来为特定接口生成形式为$Proxy的二进制字节流。
  2. 将这个字节流代表的静态存储结构转化为方法区的运行时数据结构
  3. 在内存中生成一个代表这个类的Class对象，作为方法区这个类的各种数据结构的访问入口。

类和类加载器的关系

不同的类加载器加载的类都放在虚拟机内部。如果两个不同的classloader都加载同一个类，但是他们被认为不是同一个类。这是为什么呢？

cl2.png
如上图所示，
Classloader1和classloader2都加载了a.b.A 这个class文件，

但在虚拟机里会存在两个名字为a.b.A类的Class类实例,如下图所示

cl3.png

findClass和defineClass的逻辑都是ClassLoader(程序员可以自定义自己的类加载器)里面定义的，这里生成了两个不同的Class对象。

类加载的时机

某一个类加载的时机？

JVM虚拟机并没有强制规定什么时候需要，但是对于类的初始化的时机却是有强制规定的（也意味着此时这个类需要加载了），总结为以下几点：

  • 虚拟机启动时，用户需要指定一个要执行的主类（包含main方法的类），虚拟机会先初始化这个主类

vm1.png

如上图，虚拟机启动时，
Main是执行的主类
虚拟机会先初始化Main

  • 当初始化一个类的时候，如果父类还没进行初始化，那一定先触发父类的初始化。

Object类是所有类的父类，所以在虚拟机中，Object类一定是最先开始进行初始化的。
这个从上面的图也看的出来。

  • 遇到调用点（new、getstatic、putstatic和invokestatic）这四条字节码指令，如果类还没进行过初始化，则触发其初始化，而且默认和调用点所在类的类加载器使用同样的类加载器。
  • 特殊情况： final字段有些特殊，在编译阶段就放入调用类的常量池了，所以直接引用类的静态常量字段，也不会导致类的初始化。
  • reflect 的包对类进行反射调用的时候，如果类还没有进行过初始化，则需要先进行初始化。
  • 1.7对动态语言的支持，MethodHandle实例最后的解析结果REF_getStatic的方法句柄，如果还没有进行过初始化，则需要先触发其初始化。

Class的世界

上文一直说，字节码通过类加载器加载到方法区的是Class对象。这里很需要先理解一下Class对象，如下图

jvm3.png

  • Object类:可表示万事万物
  • Class类:可表示字节码,无论是java源文件编译的或者程序生成的字节码都可以
  • Object类:没有成员
  • Class类成员:成员包含classloader， name …
    JVM加载类的时候会先加载Object类，所以启动JVM的时候大致的流程可能如下：

jvm2.png

（上面这个图为了演示ClassLoader加载类，有一个故意的bug,下文会做出更正。）

另外抛出一个问题，Object类是哪个类加载器加载的？这个问题在下文中给出解答。

认识常用的类加载器

cl7.png

  • 引导类加载器 Bootstrap ClassLoader,负责加载JDK的核心lib资源。
  • 扩展类加载器 Extension ClassLoader,负责JDK 自身扩展机制的包资源。
  • 程序员编程的类加载器,负责加载用户类路径（Classpath） 上所指定的类库。如果APP中没有定义自己的Classloader，一般来说Application Classloader就是程序中默认的类加载器。

需要理解Classpath是Application Classloader作用域（每个JVM都应该有自己的Classpath,比如当前程序运行的目录，本地maven仓库等等）

  • 在这里可以看出，java世界存在三个基本的作用域：JVM lib、JVM lib的扩展和ClassPath。

ClassLoader的parent继承机制

parent继承机制是JVM类加载器的基本思想，下面详细详细讲讲双亲委派模型。

Classloader都要定义自己的parent（如果为null，则默认是Bootstrap ClassLoader）,双亲委派模型的意思就是这些类在loadClass的时候都先委派给parent去找，如果parent找到了，那自身就不找了，否则才由自身去处理。

目前JAVA世界的ClassLoader基本如下所示：

vm3.png

这样的好处就是Object类是Bootstrap Classloader加载的，那大家读到的Object都是rt.jar定义的那个，如果不采用双亲委托模型，那用户定义一个自己的java.lang.Object，并且使用默认的Application Classloader来加载，那么Object就被替换了，这个世界也就乱套了。

双亲委派的具体实现代码：

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复制代码protected Class<?> loadClass(String name, boolean resolve) 
{
    synchronized (getClassLoadingLock(name)) {
        // 这个类如果已经存在虚拟机之中，直接返回，就不再找了
        Class<?> c = findLoadedClass(name);
        if (c == null) {
            long t0 = System.nanoTime();
            try {
          //委托给parent去找
                if (parent != null) {
                    c = parent.loadClass(name, false);
                } else {
                    c = findBootstrapClassOrNull(name);
                }
            } //...
            if (c == null) {
                // 尝试去找到这个类
                long t1 = System.nanoTime();
                c = findClass(name);
                ...
            }
       ...
        return c;
    }
}

loadClass的时候，这个类如果已经存在虚拟机之中，直接返回，就不再找了。
上面讲【Class的世界】错误的地方可以改为：

vm5.png

另外可以看出，双亲委派模型给出了类的几个作用域的不同的权限等级：

vm4.png

在C2的实现严格遵守
双亲委派模型的情况下，C2加载A类（因为已找到，不再加载）。

因为这样的加载机制，按照全限定名去加载的时候，

  • JDK lib Domain中的类具有最高优先级，
  • JDK lib ext Domain的类具有第二优先级，
  • classpath Domain的类具有第三优先级
  • 其他Domain（网络，代码生成）等产生的类需要用户自定义编写的ClassLoader去加载。

Object 类是哪个Classloader加载的？

Object 类是哪个Classloader加载的？这个问题已经很好回答了。

根据类加载的时机、类加载的委派模型，我们知道事情发生的步骤如下:

  • 虚拟机启动时，Main是执行的主类，虚拟机会先初始化Main。
  • 根据初始化原则，会先初始化父类Object。
  • Application Classloader 通过parent委派模型给Extension Classloader。
  • Extension Classloader 委派给BootStrap Classloaderloader。
  • BootStrap Classloaderloader加载Object类对象（因为Object的定义在rt.jar里）。
  • Applciation Classloader继续加载Main类，先委派给parent，但parent都没找到（往上层递归），最后Applciation Classloader自己在classpath找到这个Main并加载Main对象。
    vm2.png

Classloader隔离和可见性问题

上面讲【Class的世界】的那个图，在符合parent委派规则的场景下，那个图是不正确的。

但是如果不是遵守这个机制，还是可以存在的。只是如果JVM里面有两个名字相同的类对象怎么办呢？

vm6.png
这里做了一个小实验

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
java复制代码package com.ywz;
public class ClassLoaderTest {
    public static void main(String[] args) throws ClassNotFoundException, InstantiationException, IllegalAccessException {
        ClassLoader myloader=new ClassLoader() {
            @Override
            public Class<?> loadClass(String name) throws ClassNotFoundException {
                try {
                    String filename = name.substring(name.lastIndexOf(".") + 1) + ".class";
                    // 预备加载的类：ClassLoaderTest.class
                    InputStream stream = getClass().getResourceAsStream(filename);
                    if (stream == null) {
                        return super.loadClass(name);
                    } else {
                        byte[] bytes = new byte[stream.available()];
                        stream.read(bytes);
                        return defineClass(name, bytes, 0, bytes.length);
                    }
                } catch (IOException e) {
                    throw new ClassNotFoundException(name);
                }
            }
        };
        Class loadClass = myloader.loadClass("com.ywz.ClassLoaderTest");
        //类对象是否相等
        System.out.println("class equal ? " + loadClass.equals(com.ywz.ClassLoaderTest.class));
        //类对象的名字是否相等
        System.out.println("class name equal ? " + loadClass.getName().equals(com.ywz.ClassLoaderTest.class.getName()));
        //输出类对象的ClassLoader
        System.out.println("the classLoader = " + loadClass.getClassLoader());
        System.out.println("default classLoader = " + com.ywz.ClassLoaderTest.class.getClassLoader());
    }
}

获得的结果

1
2
3
4
java复制代码class equal ? false
class name equal ? true
the classLoader = com.ywz.ClassLoaderTest$1@d716361
default classLoader = sun.misc.Launcher$AppClassLoader@18b4aac2

也就是说Class对Classloader具有可见性问题，
Classloader只能看到自己和parent加载的类，其他的类加载器加载的类对象它看不到。

类的可见性可以总结为一句话：一个类对象只对它的类加载器和类加载器的子孙可见。

上面的示例代码其实就是下图：

VM7.png

基本意思就是：

  • App ClassLoader先加载ClassLoaderTest类
  • myClassLoader也加载ClassLoaderTest类（破坏了委托模型）

所以虽然现在系统里面有两个a.b.ClassLoaderTest Class对象了,
但是系统的默认类加载器是AppClassLoader。因为main的类是AppClassLoader加载的，
所以调用com.ywz.ClassLoaderTest获取到的是AppClassLoader加载的Class对象。

默认类加载器

JVM运行程序的默认类加载器是Application Classloader。为什么呢？
这要从JVM启动过程设置说起。

launcher.png

从下面的源码可以看到，这是Launcher初始化设置的。

1
2
3
4
5
6
7
8
9
10
java复制代码public Launcher() {
   //返回默认的classloader
   public ClassLoader getClassLoader() {
    return this.loader;
   }
    try {//AppClassLoader就是默认的loader
        this.loader = Launcher.AppClassLoader.getAppClassLoader(var1);
    } catch (IOException var9) {
        throw new InternalError("Could not create application class loader", var9);
    }

另外，当一个ClassLoader装载一个类时，除非显示地使用另一个ClassLoader，则该类所依赖及引用的类也由这个ClassLoader载入。简单点说，程序的入口使用的是什么类加载器，那么后面的类new出的对象也使用该类加载器，如下图：

invoke.png

再说说类加载的细节-数组和基本数据类型

在JAVA世界里一切都是对象，对象一定是属于某个类，
但数组和基本数据类型在JVM里面是特殊的类。

  • 非数组类的加载阶段可以使用系统提供的各层类加载器完成。也可以使用用户自定义的类加载器去完成，开发者可以通过定义自己的类加载器去控制字节流的获取方式。
  • 基本数据类型(int,char)等通过Bootstrap Classloader加载的。
  • 数组类是特殊的：数组类本身不通过类加载器创建，它是由Java虚拟机直接创建的,虚拟机创建一个数组需要考虑分配多大的内存。

cl6.png
在上图中，

  • 数据类型的元素类是引用类型B,B是由类加载器C1加载创建的。要在数组类对象标记上加载B的ClassLoader C1。故而，虚拟机创建了Class对象(vector B[10],C1)
  • 非引用类型，比如int，元素类型的加载就是和引导类加载器关联。故而，虚拟机直接创建Class对象(vector int[10],引导类加载器)

开放的类加载机制

上面已经讲完了类加载的基本原理和事项，下面会开放式的谈一谈开放的类加载机制。

灵活的类加载

JVM为了给程序员提供便利，开放了类加载器，这是java语言的创新。

我们可以定义自己的classloader做一些很有用的事情，比如

  • 加载加密的字节码
  • 热加载：开发阶段可以监听变化的字节码，再使用自定义的类加载器重新加载这些类，最后使用反射重启main函数。

vm9.png
基本流程大致是这样，想要具体了解热加载的文章，可以看我写的另一篇深入源码解读SpringBoot热加载工具DevTools-类加载机制和基本流程。

破坏parent继承模型

上面的图定义的myClassloader破坏了双亲继承，但这是在我的小世界里，无伤大雅。

双亲委派很好的解决了解决了各个类加载器的基础类的统一问题，但是如果基础类想要调用用户代码怎么办？
实际上，JVM的世界存在一些大规模破坏parent继承模型的案例，比如JNDI。

JNDI的代码由启动类加载器去加载（在JDK 1.3时放入rt.jar），但JNDI的目的是对资源进行集中管理和查找，它需要调用独立厂商实现并部署在应用程序的ClassPath下的JNDI接口提供者的代码（这原本是APP ClassLoader应该做的事情。）Bootstrap ClassLoader却不能加载ClassPath下的代码，应该怎么办？

jndi1.png

可以看出，JNDI想加载使用classpath下的代码,但有两个问题:

  • 这会破坏双亲继承
  • Bootstrap ClassLoader做不到加载Classpath下的代码

为了解决第二个问题，JVM引入Thread Context ClassLoader。

jvm4.png

线程会继承父线程的Context ClassLoader，如果在应用程序的全局范围内都没有设置过Context ClassLoader的话，
这个类加载器默认使用App ClassLoader。

这样的话，这个问题就可以这么解决了：

jndi3.png
JNDI使用contextClassLoader便可以加载到Classpath下的代码。

引入ContextClassLoader这样的一个设计，是走旁路曲线救国，但这破坏了parent继承模型。
但这也能解决问题。

参考文献

  • 深入理解JAVA虚拟机（第2版）

本文转载自: 掘金

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

发表于

这是我参与11月更文挑战的第1天，活动详情查看：2021最后一次更文挑战

JDK1.5以前，创建线程是通过继承Thread，但是继承Thread存在如下弊端：

  1. 每次new Thread的时候都需要新建一个线程，性能差
  2. 线程缺乏统一管理，可能无限制的新建线程，相互竞争，有可能占用过多系统资源导致死机或者OOM
  3. Thread类缺少更多功能，比如更多的执行、定期执行、线程中断。

JDK1.5之后，系统自带的Executor框架，为开发者们自定义线程池带来了极大的方便。线程池的好处如下：

  1. 重用存在的线程，减少对象创建、消亡的开销、性能佳
  2. 可以有效的控制最大并发线程数，提高系统资源利用率，同时可以避免过多资源竞争，避免阻塞。
  3. 提供定时执行、定期执行、单线程、并发数控制等功能。

下面先说说Executors。

Executors

Executors是一个线程池工厂类，里面有许多静态方法，供开发者调用。

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
java复制代码/* 该方法返回一个固定线程数量的线程池，该线程池池中的线程数量始终不变。
 * 当有一个新的任务提交时，线程池中若有空闲线程，则立即执行。
 * 若没有，则新的任务会被暂存在一个任务队列中，待有线程空闲时，便处理在任务队列中的任务 
 * 默认等待队列长度为Integer.MAX_VALUE
 */
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(1);

/* 该方法返回一个只有一个线程的线程池。
 * 若多余一个任务被提交到线程池，任务会被保存在一个任务队列中，等待线程空闲，按先入先出顺序执行队列中的任务
 * 默认等待队列长度为Integer.MAX_VALUE
 */
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();

/* 
 * 该方法返回一个可根据实际情况调整线程数量的线程池。
 * 线程池的线程数量不确定，但若有空闲线程可以复用，则会优先使用可复用的线程。
 * 若所有线程均在工作，又有新任务的提交，则会创建新的线程处理任务。
 * 所有线程在当前任务执行完毕后，将返回线程池进行复用
 */
ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();

/* 该方法返回一个ScheduledExecutorService对象，线程池大小为1。
 * ScheduledExecutorService接口在ExecutorService接口之上扩展了在给定时间内执行某任务的功能，
 * 如在某个固定的延时之后执行，或者周期性执行某个任务
 */
ExecutorService newSingleThreadScheduledExecutor = Executors.newSingleThreadScheduledExecutor();

/*
 * 该方法也返回一个ScheduledExecutorService对象，但该线程池可以指定线程数量
 */
ExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(1);

以上的工具类的具体实现都是基于ThreadPoolExecutor类，处理策略都是AbortPolicy（直接抛出异常，阻止系统正常工作）

Executors 返回的线程池对象的弊端如下：

  1. FixedThreadPool、SingleThreadPool: 允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致OOM。
  2. CachedThreadPool、 ScheduledThreadPool、SingleThreadScheduledExecutor: 允许的创建线程数量为 Integer.MAX_VALUE， 可能会创建大量的线程，从而导致 OOM。

通常情况下，线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。

ThreadPoolExecutor说明

1
2
3
4
5
6
7
java复制代码public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

参数说明：

  • corePoolSize： 当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads方法，线程池会提前创建并启动所有基本线程。\
  • maximumPoolSize： 线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是如果使用了无界的任务队列这个参数就没什么效果。\
  • keepAliveTime： 当线程池线程数量超过corePoolSize时，多余的空余线程存活时间，即超过corePoolSize的空闲线程，在多长时间内被销毁。如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率。\
  • unit： keepAliveTime的单位。\
  • workQueue： 任务队列，被提交但尚未被执行任务的任务。有以下几种队列模式：
  1. ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序。
  2. LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。
  3. SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。
  4. PriorityBlockingQueue：一个具有优先级得无限阻塞队列。
  • threadFactory： 线程工厂，用于创建线程，一般用默认的即可。也可以通过线程工厂给每个创建出来的线程设置更有意义的名字，Debug和定位问题时非常有帮助。\
  • handler： 拒绝策略，当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。系统内置的策略模式如下：
  1. AbortPolicy：直接抛出异常，阻止系统正常工作。
  2. CallerRunsPolicy：只要线程未关闭，该策略直接在调用者线程中，运行当前被丢弃的任务。（即如果线程池的线程数量达到上限，该策略会把任务队列中的任务放在调用者线程当中运行）
  3. DiscardOldestPolicy：该策略将丢弃最老的一个请求，也就是即将被执行的那个任务，并尝试再次提交当前任务。
  4. DiscardPolicy：该策略会默默丢弃无法处理的任务，不予任何处理。当然使用此策略，业务场景中需允许任务的丢失。
  5. 当然也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略。

ThreadPoolExecutor 的执行流程

  1. 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（注意：执行这一步骤需要获取全局锁）。
  2. 如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue。
  3. 如果无法将任务加入BlockingQueue（队列已满），则创建新的线程来处理任务（注意，执行这一步骤需要获取全局锁）。
  4. 如果创建新线程将使当前运行的线程超出maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

ThreadPoolExecutor 样例

1
2
3
java复制代码ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20, 1,TimeUnit.SECONDS, 
new ArrayBlockingQueue<Runnable>(15),
new ThreadPoolExecutor.DiscardOldestPolicy());

上面的语句说明如下：

  • 10是表示允许10个线程，
  • 20是最多可以有20个线程在进行或等待，
  • 1是线程池维护线程所允许的空闲时间为1秒，
  • TimeUnt.SECONDS 线程池维护线程所允许的空闲时间的单位以秒为单位，
  • 下一个参数是指线程池所使用的缓冲队列为有界队列，
  • 最后一个参数是线程池对拒绝任务的处理策略抛弃旧的任务。

本文转载自: 掘金

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

发表于

一、核心组成

1.1、核心组成部分

image-20211028095933103

Server：又称Broker ,接受客户端的连接，实现AMQP实体服务。 安装rabbitmq-server

Connection：连接，应用程序与Broker的网络连接 TCP/IP/ 三次握手和四次挥手

Channel：网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道，客户端可以建立对各Channel，每个Channel代表一个会话任务。

Message :消息：服务与应用程序之间传送的数据，由Properties和body组成，Properties可是对消息进行修饰，比如消息的优先级，延迟等高级特性，Body则就是消息体的内容。

Virtual Host 虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个虚拟主机理由可以有若干个Exhange和Queueu，同一个虚拟主机里面不能有相同名字的Exchange

Exchange：交换机，接受消息，根据路由键发送消息到绑定的队列。(==不具备消息存储的能力==)

Bindings：Exchange和Queue之间的虚拟连接，binding中可以保护多个routing key.

Routing key：是一个路由规则，虚拟机可以用它来确定如何路由一个特定消息。

Queue：队列：也成为Message Queue,消息队列，保存消息并将它们转发给消费者。

1.2、RabbitMQ 整体架构

img

1.3、运行流程

  • 1、生产者产生消息数据
  • 2、经过序列化只会指定交换机和路由信息
  • 3、将消息发送到Broker 中
  • 4、消费者根据路由来消费指定队列消息

img

二、消息模式入门案例

RabbitMQ 它提供了六种消息模型，但是第 6 种其实是 RPC，并非 MQ，不用专门学习，剩下的5种，3、4、5这三种都是订阅模型，只不过进行路由的方式不同。第一种的简单模式我们已经在初始篇中学习过了，在这小节中，我们来学习其他四种消息的使用入门。

image-20211028101843898

2.1 Work Queues

工作队列或者竞争消费者模式。

与简单模式不同的是存在多个队列来消费这些消息。也就是存在多个消费者，同时一条消息只能被一个工作队列消费。这还是很好理解的。

img

那么 RabbitMQ 如何确保消息发送到哪个消费者呢？这就有两种发送策略：

  • 轮询发送：一个消费者一条，按均分配；
  • 公平发送：根据消费者的消费能力进行公平分发，按劳分配；
2.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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
java复制代码/**
 * 工作队列---轮询
 * @Author xiaolei
 * @Date 2021/10/28 11:21
 **/
public class ProducerRobinTest {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("192.168.81.102");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("111111");
​
        //3、从连接工厂中获取连接
        Connection connection = connectionFactory.newConnection("producer");
        //4、从连接中获取通道 channel
        Channel channel = connection.createChannel();
        //5、申明 queue 存储消息
        /**
         * 如果队列不存在，则会创建,不允许相同topic 存在
         * @params1： queue 队列的名称
         * @params2： durable 队列是否持久化
         * @params3： exclusive 是否排他，即是否私有的，如果为true,会对当前队列加锁，
         * @params4： autoDelete 是否自动删除
         */
        channel.queueDeclare("queue1", false, false, false, null);
​
        // 6、发送消息
        for (int i = 1; i <=10; i++) {
            // @params1: 交换机exchange
            // @params2: 队列名称/routing
            // @params3: 属性配置
            // @params4: 发送消息的内容
            channel.basicPublish("","queue1",null,("潇雷挺帅，说第"+i+"遍。").getBytes());
            Thread.sleep(1000);
        }
​
        // 7、关闭连接
        channel.close();
    }
}

消费者代码：两个消费者一致

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
java复制代码public class RobinWork1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("192.168.81.102");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("111111");
​
        //3、从连接工厂中获取连接
        Connection connection = connectionFactory.newConnection("consumer");
        //4、从连接中获取通道 channel
        Channel channel = connection.createChannel();
​
        DeliverCallback deliverCallback =(String a, Delivery b)->{
            String message = new String(b.getBody());
            System.out.println("work1"+message);
        };
        CancelCallback cancelCallback =(String a)->{
            System.out.println("消息消费被中断");
        };
        /**
         * 消费者消费消息
         * @params1: 消费哪个队列
         * @params2：消费成功之后是否要自动应答 true代表自动应答 ，flase代表手动应答。
         * @params3： 消费者消费成功的回调
         * @params4： 消费者消费失败的回调
         */
        channel.basicConsume("queue1",true,deliverCallback,cancelCallback);
    }
}

打印结果是按均分配的，

2.1.2 公平发送

该策略是根据消费者处理消息的能力不同，存在处理慢的问题时候，就采用能者多劳的模式处理。 “厉害的多干点” 。

要开启该策略，就需要消费者开启手动应答，关闭自动应答。而

关闭自动应答的代码：

1
arduino复制代码channel.basicConsume("queue1",false,deliverCallback,cancelCallback);

开启手动应答的代码：

1
less复制代码channel.basicAck(b.getEnvelope().getDeliveryTag(),false);

原理是消费者要告知MQ，在未处理完当前消息前，不要发新的消息过来，每消费完一次就应答一次。

消费者代码都一样，不同的是给其中一个消费者sleep 的时间久一点就好。

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 FairWork2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("192.168.81.102");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("111111");
​
        //3、从连接工厂中获取连接
        Connection connection = connectionFactory.newConnection("consumer2");
        //4、从连接中获取通道 channel
        Channel channel = connection.createChannel();
        //声明在一条消息被确认消费前，不会再发给该消费者另外的消息
        channel.basicQos(1);
        DeliverCallback deliverCallback =(String a, Delivery b)->{
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String message = new String(b.getBody());
            System.out.println("work2"+message);
            // 手动应答消息 false 代表单个应答  true：代表批量应答，是以通道未单位的
            channel.basicAck(b.getEnvelope().getDeliveryTag(),false);
        };
        CancelCallback cancelCallback =(String a)->{
            System.out.println("消息消费被中断");
        };
        /**
         * 消费者消费消息
         * @params1: 消费哪个队列
         * @params2：消费成功之后是否要自动应答 true代表自动应答 ，flase代表手动应答。
         * @params3： 消费者消费成功的回调
         * @params4： 消费者消费失败的回调
         */
        channel.basicConsume("queue2",false,deliverCallback,cancelCallback);
    }
}

在写测试类的时候，如果channel没有很好的关闭连接，可以看到这边有多余的消费者存在，将这些线程关闭即可。保证工作线程只有自己。

image-20211028144723255

2.2 发布订阅模式（fanout）

发布订阅模式的结构如图：

img

简单模式和工作队列模式本质是一样的，只有一个 queue。message 发送给 exchange 后，exchange 判断为 direct 模式后，就会把message 转发到绑定的 queue。而且一条消息只能发送给 1个consumer，发布订阅模式存在多个队列来共同消费数据。

该模式下：

  • 1、发布者只创建 type 为 fanout 的命名 exchange；
  • 2、consumer 根据需要创建自己的 queue，并且连接到 publisher 创建的 exchange 中，实施exchange与 queue 的绑定；如果 queue 不需要持久化，则可用临时 queue。
  • 3、发布者发送 message，message 到达 exchange 后，exchange 判断为 fanout 模式，就直接把 message 发送给与自己绑定的 queue 中
  • 4、每个queue将消息分发给自己连接的 consumer
2.2.1 交换机四种类型：

发送消息 rabbitmq 它一定有个 交换机，默认不指定的话为下面这个。

image-20211028151837562

  • direct ：处理路由键，需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。

img

  • topic： 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“ ”匹配不多不少一个词。因此“abc.#”能够匹配到“abc.def.ghi”，但是“abc. ” 只会匹配到“abc.def”。

img

  • headers：

不处理路由键。而是根据发送的消息内容中的headers属性进行匹配。在绑定Queue与Exchange时指定一组键值对；当消息发送到RabbitMQ时会取到该消息的headers与Exchange绑定时指定的键值对进行匹配；如果完全匹配则消息会路由到该队列，否则不会路由到该队列。headers属性是一个键值对，可以是Hashtable，键值对的值可以是任何类型。而fanout，direct，topic 的路由键都需要要字符串形式的。

  • fanout ： 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。
2.2.2 发布订阅模式测试

生产者

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
java复制代码public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
                // 1、创建连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            // 2、设置连接属性
            connectionFactory.setHost("192.168.81.102");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("test");
            connectionFactory.setUsername("admin");
            connectionFactory.setPassword("111111");
​
            //3、从连接工厂中获取连接
            Connection connection = connectionFactory.newConnection("producer");
            //4、从连接中获取通道 channel
            Channel channel = connection.createChannel();
            /**
             * 5、制定交换机的模式 fanout、名称  exchange1
             */
            channel.exchangeDeclare("exchange1","fanout");
            // 6、发送消息
            for (int i = 1; i <=20; i++) {
                // @params1: 交换机exchange
                // @params2: 队列名称/routing
                // @params3: 属性配置
                // @params4: 发送消息的内容
                channel.basicPublish("exchange1","",null,("潇雷挺帅，说第"+i+"遍。").getBytes());
                System.out.println("发送第"+i);
        }
    }
}

消费者：

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
java复制代码public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("192.168.81.102");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("test");
        connectionFactory.setPassword("test");
        connectionFactory.setVirtualHost("test");
​
        //3、从连接工厂中获取连接
        Connection connection = connectionFactory.newConnection("consumer2");
        //4、从连接中获取通道 channel
        Channel channel = connection.createChannel();
        //声明在一条消息被确认消费前，不会再发给该消费者另外的消息
        channel.basicQos(1);
        DeliverCallback deliverCallback =(String a, Delivery b)->{
            String message = new String(b.getBody());
            System.out.println("work2"+message);
        };
        CancelCallback cancelCallback =(String a)->{
            System.out.println("消息消费被中断");
        };
        channel.queueDeclare("queue2",false,false,true,null);
        channel.queueBind("queue2","exchange1","");
        /**
         * 消费者消费消息
         * @params1: 消费哪个队列
         * @params2：消费成功之后是否要自动应答 true代表自动应答 ，flase代表手动应答。
         * @params3： 消费者消费成功的回调
         * @params4： 消费者消费失败的回调
         */
        channel.basicConsume("queue2",true,deliverCallback,cancelCallback);
    }
}

消费者2 队列名字不一样，其他都一样，此时能看到 exchange1 下面bind 的两个队列。

image-20211028152357472

2.3 路由模式（direct）

Direct 模式是 fanout模式的一种，新增加了路由key 的选择。

结构如下：

img

设置路由模式，然后绑定对应的 key 之后：

  • routing-key 为 error 的 message 只会发送到 Q1；
  • routing-key 为 info、error、warning 的消息会发送到 Q2，而其他消息都会忽略丢弃。
  • 其中的 error 被所有消费者都关注了，那么它其实就相当于发布订阅模式，它的消息都会发送到consumer1和consumer2.

生产者指定路由：

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 Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("192.168.81.102");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("test");
        connectionFactory.setUsername("test");
        connectionFactory.setPassword("test");
​
        //3、从连接工厂中获取连接
        Connection connection = connectionFactory.newConnection("producer");
        //4、从连接中获取通道 channel
        Channel channel = connection.createChannel();
        /**
         * 5、制定交换机的模式 fanout、名称  exchange1
         */
        channel.exchangeDeclare("exchange2","direct");
        // 6、发送消息
        for (int i = 1; i <=20; i++) {
            // @params1: 交换机exchange
            // @params2: 队列名称/routing
            // @params3: 属性配置
            // @params4: 发送消息的内容
            if(i%2==0){
                channel.basicPublish("exchange2","error",null,("潇雷挺帅，说第"+i+"遍。").getBytes());
            }else{
                channel.basicPublish("exchange2","info",null,("潇雷挺帅，说第"+i+"遍。").getBytes());
            }
            System.out.println("发送第"+i);
        }
    }
}
​

消费者指定路由：

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
java复制代码public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("192.168.81.102");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("test");
        connectionFactory.setPassword("test");
        connectionFactory.setVirtualHost("test");
​
        //3、从连接工厂中获取连接
        Connection connection = connectionFactory.newConnection("consumer2");
        //4、从连接中获取通道 channel
        Channel channel = connection.createChannel();
        //声明在一条消息被确认消费前，不会再发给该消费者另外的消息
        channel.basicQos(1);
        DeliverCallback deliverCallback =(String a, Delivery b)->{
            String message = new String(b.getBody());
            System.out.println("work2"+message);
        };
        CancelCallback cancelCallback =(String a)->{
            System.out.println("消息消费被中断");
        };
        channel.queueDeclare("queue2",false,false,true,null);
        channel.queueBind("queue2","exchange2","error");
        /**
         * 消费者消费消息
         * @params1: 消费哪个队列
         * @params2：消费成功之后是否要自动应答 true代表自动应答 ，flase代表手动应答。
         * @params3： 消费者消费成功的回调
         * @params4： 消费者消费失败的回调
         */
        channel.basicConsume("queue2",true,deliverCallback,cancelCallback);
    }
}

2.4 主题模式（Topic）

img

主题模式可以看作路由模式的拓展，它增加了路由key 的模式。就像我们的模糊匹配。

发送到 topic exchange 的 routing-key 必须是有规则的字符串，例如以 “ . “ 分割，并且每个分隔符内的字符串带有特征，例如 “xiaolei.orange”,它的最大字符长度最大为 255 字节。

topic 其实是与 direct 模式很相似，对于 bind keys ，topic 模式多了两个特征：

  • “ * ” 星号可以替代任意一个字符
  • “ # ” 可以替代 0 到多个字符。

理解了这个，我们来做个案例，例如：一批影片里面有 爱国片、动作片和喜剧片。爱国片、动作片里面有吴京；爱国片、喜剧片里面有沈腾。

  • 那么有批粉丝是吴京的粉丝，那他就找出所有吴京的影片，路由规则为 “*. 吴京”或”#.吴京”；
  • 有粉丝只看爱国片里的吴京，对动作片不感兴趣，路由规则为 “爱国.吴京”；

生产者：

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
java复制代码public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("192.168.81.102");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("test");
        connectionFactory.setUsername("test");
        connectionFactory.setPassword("test");
​
        //3、从连接工厂中获取连接
        Connection connection = connectionFactory.newConnection("producer");
        //4、从连接中获取通道 channel
        Channel channel = connection.createChannel();
        /**
         * 5、制定交换机的模式 fanout、名称  exchange1
         */
        channel.exchangeDeclare("exchange3","topic");
        // 6、发送消息
        for (int i = 1; i <=40; i++) {
            // @params1: 交换机exchange
            // @params2: 队列名称/routing
            // @params3: 属性配置
            // @params4: 发送消息的内容
            if(i%4==0){
                channel.basicPublish("exchange3","爱国.吴京",null,("爱国.吴京，说第"+i+"遍。").getBytes());
            }else if(i%4 ==1){
                channel.basicPublish("exchange3","爱国.沈腾",null,("爱国.沈腾，说第"+i+"遍。").getBytes());
            }else if(i%4 ==2){
                System.out.println("ggg");
                channel.basicPublish("exchange3","动作.吴京",null,("动作.吴京，说第"+i+"遍。").getBytes());
            }else if(i%4 ==3){
                channel.basicPublish("exchange3","喜剧.沈腾",null,("喜剧.沈腾，说第"+i+"遍。").getBytes());
            }
            System.out.println("发送第"+i);
        }
    }
}

消费者：

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
java复制代码public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2、设置连接属性
        connectionFactory.setHost("192.168.81.102");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("test");
        connectionFactory.setPassword("test");
        connectionFactory.setVirtualHost("test");
​
        //3、从连接工厂中获取连接
        Connection connection = connectionFactory.newConnection("consumer2");
        //4、从连接中获取通道 channel
        Channel channel = connection.createChannel();
        //声明在一条消息被确认消费前，不会再发给该消费者另外的消息
        channel.basicQos(1);
        DeliverCallback deliverCallback =(String a, Delivery b)->{
            String message = new String(b.getBody());
            System.out.println("work2"+message);
        };
        CancelCallback cancelCallback =(String a)->{
            System.out.println("消息消费被中断");
        };
        channel.queueDeclare("queue2",false,false,true,null);
        channel.queueBind("queue2","exchange3","#.吴京");
        /**
         * 消费者消费消息
         * @params1: 消费哪个队列
         * @params2：消费成功之后是否要自动应答 true代表自动应答 ，flase代表手动应答。
         * @params3： 消费者消费成功的回调
         * @params4： 消费者消费失败的回调
         */
        channel.basicConsume("queue2",true,deliverCallback,cancelCallback);
    }
}

三、小结

工作模式总结：

  • 1、简单模式 ：一个生产者，一个消费组，不需要设置交换机（使用默认的交换机）；
  • 2、工作队列模式：Work Queue ：一个生产者，多个消费者，不需要设置交换机（使用默认的交换机）；
  • 3、发布订阅模式：需要设置类型为 fanout 的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列；
  • 4、路由模式：需要设置类型为 direct 的交换机，交换机和队列进行绑定，并且指定 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列；
  • 5、通配符模式 Topic：需要设置为 topic 的交换机，交换机和队列进行绑定，并且指定通配符方式的 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列。

本文主要介绍了 RabbitMQ 的队列结构和它几大核心的路由模式的使用。

在下一的章节中，将会介绍Rabbbitmq 与 SpringBoot 项目的集成使用。

本文转载自: 掘金

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

发表于

「这是我参与11月更文挑战的第1天，活动详情查看：2021最后一次更文挑战」

  • 备战2022春招或暑期实习，祝大家每天进步亿点点！Day5
  • 本篇总结的是 《Java 枚举的正确使用方式》，后续会每日更新~
  • 关于《Redis入门到精通》、《并发编程》等知识点可以参考我的往期博客
  • 相信自己，越活越坚强，活着就该逢山开路，遇水架桥！生活，你给我压力，我还你奇迹！

image.png

1、简介

不知道大家有没有在自己项目中看到过类似下面这样的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
arduino复制代码public static void fruitsHandle(String fruits) {

    switch (fruits) {
        case "Apple":
            // TODO
            break;
        case "Banana":
            // TODO
            break;
        case "Orange":
            // TODO
            break;
        default:
            throw new IllegalStateException("Unexpected value: " + fruits);
    }

}

出现上面这种情况是非常少的，小萌新一般也不会直接在方法中重复定义字符串进行比较，而会将其定义为常量，或者统一抽取为常量类。所以一般会看到这种代码（小捌经常在项目中看到类似这样的代码，但是小捌不敢吭声😄😄）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
arduino复制代码private static final String APPLE = "Apple";
private static final String BANANA = "Banana";
private static final String ORANGE = "Orange";

public static void fruitsHandle(String fruits) {

    switch (fruits) {
        case APPLE:
            // TODO
            break;
        case BANANA:
            // TODO
            break;
        case ORANGE:
            // TODO
            break;
        default:
            throw new IllegalStateException("Unexpected value: " + fruits);
    }

}

上面这种情况我们在代码中出现的频率非常高；它需要程序员提供一组固定常量，并且这一组固定常量在开发时或者说编译时就知道了具体的成员，这个时候我们就应该使用枚举。

枚举类型（enum type）是指由一组固定常量组成合法值的类型。

2、优势

使用枚举类型，相比直接定义常量能够带来非常多的好处。

2.1 类型安全

分别定义一个简单的肉类枚举和水果枚举

1
2
3
4
5
6
7
8
arduino复制代码// 肉类枚举
public enum MeetEnums {

    BEEF,
    PORK,
    FISH;

}
1
2
3
4
5
6
7
8
arduino复制代码// 水果枚举
public enum FruitsEnums {

    APPLE,
    BANANA,
    ORANGE;

}

我们改造上面的代码，修改入参类型即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
arduino复制代码public static void fruitsHandle(FruitsEnums fruits) {

    switch (fruits) {
        case APPLE:
            // TODO
            break;
        case BANANA:
            // TODO
            break;
        case ORANGE:
            // TODO
            break;
        default:
            throw new IllegalStateException("Unexpected value: " + fruits);
    }

}

可以看到定义枚举类型带来函数类型安全性，如果定义的是常量则无法代理这种效果

2.2 枚举能够提供更多信息

枚举在本质上还是一个类，它能够定义属性和方法，我们可以在枚举类中定义想要的方法、或者通过属性扩展枚举提供的基础信息。

比如我们做web开发时最常见的HttpStatus，在springframework框架中就被定义成了枚举类，它不仅包含了Http响应码，还能包含描述状态。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
arduino复制代码public enum HttpStatus {

	OK(200, "OK"),
    NOT_FOUND(404, "Not Found"),
    INTERNAL_SERVER_ERROR(500, "Internal Server Error");
    
    private final int value;
    private final String reasonPhrase;

    private HttpStatus(int value, String reasonPhrase) {
        this.value = value;
        this.reasonPhrase = reasonPhrase;
    }
    
}

2.3 通过函数提供更多服务

此外HttpStatus它内部还嵌套了Series枚举类，这个类可以协助HttpStatus枚举类，通过statusCode / 100的模判断当前的枚举状态是is1xxInformational、is2xxSuccessful、is3xxRedirection、is4xxClientError、is5xxServerError等等。

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
java复制代码public static enum Series {
        INFORMATIONAL(1),
        SUCCESSFUL(2),
        REDIRECTION(3),
        CLIENT_ERROR(4),
        SERVER_ERROR(5);

        private final int value;

        private Series(int value) {
            this.value = value;
        }

        public int value() {
            return this.value;
        }

        public static HttpStatus.Series valueOf(HttpStatus status) {
            return valueOf(status.value);
        }

        public static HttpStatus.Series valueOf(int statusCode) {
            HttpStatus.Series series = resolve(statusCode);
            if (series == null) {
                throw new IllegalArgumentException("No matching constant for [" + statusCode + "]");
            } else {
                return series;
            }
        }

        @Nullable
        public static HttpStatus.Series resolve(int statusCode) {
            int seriesCode = statusCode / 100;
            HttpStatus.Series[] var2 = values();
            int var3 = var2.length;

            for(int var4 = 0; var4 < var3; ++var4) {
                HttpStatus.Series series = var2[var4];
                if (series.value == seriesCode) {
                    return series;
                }
            }

            return null;
        }
    }

2.4 获取所有定义的类型

所有的枚举类会自动产生一个values()方法，它能返回当前定义枚举类的数组集，因此可以很方便的遍历怎么枚举类定义的所有枚举。比如我们简单改造一下MeetEnums枚举类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
scss复制代码public enum MeetEnums {

    BEEF("牛肉"),
    PORK("猪肉"),
    FISH("鱼肉");

    String name;

    public String getName() {
        return name;
    }

    MeetEnums(String name) {
        this.name = name;
    }

    public static MeetEnums getMeetEnumsByName(String name) {
        MeetEnums[] values = values();
        Optional<MeetEnums> optional = Stream.of(values).filter(v -> v.getName().equals(name)).findAny();
        return optional.isPresent() ? optional.get() : null;
    }

}

总之枚举类相比常量来说有太多的优点，它能使得代码更加整洁美观、安全性强、功能强大。虽然大部分情况下，枚举类的选择是由于常量定义的，但是也并不是任何时候都一定要把常量定义成枚举；具体情况大家就可以自己去斟酌啦！

本文转载自: 掘金

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

发表于

作者：小傅哥

博客：bugstack.cn

沉淀、分享、成长，让自己和他人都能有所收获！😄

一、我，多能折腾！

🚌 我跟你说，折腾蓄蓝，还能赚钱！

  • 2013年，毕业以学Java的身份进入传统企业写C#和PLC，每天5:30下班，总有伙伴找到我帮忙做个网站。PS：要用PHP写，省服务器费用！
  • 2014年，租住门脸，下班回来顺便，卖电话、内存卡、充电线、手机壳、充话费
  • 2015年，搭建 itstack.org 域名博客，编写 Netty 5.0 案例，后因 org 域名不能备案，博客也挂了
  • 2016年，从传统行业进入互联网已经半年，开始疯狂折腾 Java 以前不熟悉的技术
  • 2017年，好奇所接触的一切新鲜框架技术，羡慕技术大佬的能力，书童一样学习高并发场景实现，这么大的调用体量怎么没有挂
  • 2018年，不只是看了，开始对源码动手，把 RPC 实现了一遍，做了一套监控系统，写了一堆的中间件，申请了好几个专利，差点被淘宝中间件团队挖去。
  • 2019年，从《用Java实现一个JVM》开始，捡回我的热情，申请并备案域名、搭建 jekyll 博客、开通公众号，年底成为 CSDN 博客专家并已积累公众号粉丝1k
  • 2020年，2月编写第一个基于 Netty 实现的仿微信 IM，放到腾讯技术社区参赛，获得第一名(运营小姐姐告诉我，她把仿微信IM推给了小马哥，哈哈哈)。同年字节码编程、设计模式、面经手册等PDF相继出炉，火了 他们都问我是不是字节的！
  • 2021年，赶在图书节，我的第一本技术书籍《重学Java设计模式》出版了，出书并不是终点，只是想走走没走过的路、爬爬没爬过的山，看看风雨过后的彩虹是否那么绚烂。

结束了吗？ 不，折腾依旧在！

2021年的1024，小傅哥一路趟坑，把300篇文章的bugstack虫洞栈博客从 jekyll 升级到 vuepress！

这些年的折腾在域名申请到备案、服务器选择到上线、PHP开发扩展功能，迁移兼容博客等，经验和技能储备都没问题，也会点 html、css、js、jquery 做自定义的开发处理。但在面对没接触过的 vue 时，依旧有点不知道从哪下手，而为了vuepress博客从能用到好用，一顿操作差点干拉胯！

技术拓宽、实力增强、薪资拉升，来吧，一起折腾！

二、坑，差点肝拉？

可能你没自己搞过 vuepress 的博客，你不知道。它这个官网说啥？说：像数 1, 2, 3 一样容 tui！糟老头子坏的很，没有一点前端知识根本跑不起来，更别说是生产使用，只能算是玩具级别。当然，不要误会，人家这个也是想着你有vue的知识背景的，不过我不是没有吗！ 为此在我折腾完把坑趟平以后，知道怎么正确操作后，还专门录了个搭建 vuepress 博客教程的视频，如下：

嘿，好嘛！有了这个案例后，我就一直在琢嘛，怎么把我的博客迁移过来呢，毕竟这套架构可以更好的承载我的文章，也能很大程度的提升阅读体验和对博客的操作空间。

日日思念惦记呀，就跟你处对象一样，朝为日，暮为月，卿为朝朝暮暮啊！咋整，办它吧，想过会遇到坑，也提前铺平垫稳了，但无奈还是会有很多因为没折腾过，vue 知识储备不足而遇到的问题。

1. 300篇文章迁移，链接问题

  • 背景：旧版博客到新版博客的迁移，不只是想把界面结构更换了，还希望解决文章分类的问题，所以在有些文章会重新摆放它所属的文件夹。
  • 问题：那么就会出现文章路径与原来文章路径不一致的问题。这其中还包括原有文章中用到的图片路径问题。
  • 方案：分两阶段解决，如果目前把旧版文章全部下线会导致很大一部分老用户，尤其是保留了旧版链接的用户，在访问文章时就会出现 404 找不到了文章了，体验非常不好。所以这里上新版博客的同时也保留旧版博客在同一个服务器中，另外把服务器的 404 配置页面配置到 bugstack.cn 以免将来旧版链接几乎已经没有什么流量删除时，再次出现 404 所以为了安全起见先把 404 页面配置成首页，这样也能满足用户找到现有的文章路径。

2. 浏览器限制cookie，解锁问题

  • 背景：小傅哥的博客有一个阅读解锁功能，这个解锁是基于公众号做的开发，浏览器生成通过友盟cookie获取6位的唯一码，当做用户的个人ID，再通过这个ID到公众号输入后绑定到公众号个人身份ID上，就解锁了。
  • 问题：但有些用户限制了浏览器cookie，或者安装了一些拦截广告的插件，那么 cookie 就不能正确拿到了，一点阅读全文这个解锁码就不唯一了，卡卡的闪。我头都大了！
  • 方案：咋办，总不能告诉每个用户，你把插件给我删了 cookie 给我打开吧，那他们肯定觉得傅哥技术拉胯呀！后来我就想肯定会有一个技术手段能在浏览器计算出唯一ID，否则我不登录时候浏览器xxxx，怎么就给我推荐了呢。经过搜索还真有对应的东西，就是浏览器指纹，fingerprintjs 浏览器指纹库，这个库用于查询浏览器属性并从中计算散列的访问者标识符。与 cookie 和本地存储不同，指纹在匿名/私有模式下保持不变，甚至在清除浏览器数据时也保持不变。妥，有了这个技术就妥嘞，在 cookie 限制的情况下，通过浏览器指纹获取唯一ID截取成文章解锁码。

3. 构建文件哈希命名，版本问题

  • 背景：看到这的时候我人都傻了，vue 变异出来的文件竟然 数字+哈希值.js 想想也对，人家这样可以更好的保证版本升级，不至于出现缓存问题。
  • 问题：有啥问题呢，你看左侧本地和右侧服务器，发现没。只要每次部署一个版本，服务器端就多出一个新的js文件，好家伙这用不了多久，不把我这小服务器干满了吗！！！删除？删错了呢，那多吓人！
  • 方案：其实有时候遇到这样问题不知道咋查你懂吧，搜了半天不如问前端一下，原来这个是可以通过webpack把打包文件名称处理的，也就是你可以让每次的构建出来的 js、css、img 文件名，都是一个固定的名字。但还有点小问题，vue 里是这么搞，vuepress 怎么搞呢，找了不少资料终于找到 chainWebpack 可以配置文件名 config.output.filename('assets/js/cg-[name].js?v=' + dateTime).end();

模块打包配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
java复制代码chainWebpack: config => {
    if (process.env.NODE_ENV === 'production') {
        const dateTime = new Date().getTime();
        
        // 清除js版本号
        config.output.filename('assets/js/cg-[name].js?v=' + dateTime).end();
        config.output.chunkFilename('assets/js/cg-[name].js?v=' + dateTime).end();
        
        // 清除css版本号
        config.plugin('mini-css-extract-plugin').use(require('mini-css-extract-plugin'), [{
            filename: 'assets/css/[name].css?v=' + dateTime,
            chunkFilename: 'assets/css/[name].css?v=' + dateTime
        }]).end();
    }
}

  • 通过这样配置后，再打包出来的文件就都是 cg-名称.js 的文件了，每次 push 到我的站点服务器是上就可以只有一份同名的 js 了，不至于把我服务器搞爆。

4. Java VS PHP 公众号开发，服务器资源问题

流程图

DDD + PHP = 爷青回

  • 背景：因为要开发博客到公众号的引流，所以需要对公众号做一些开发处理。
  • 问题：最初我是使用 Java 语言写了一版，但目前随着用户体量的上升，调用我这个跑着 Java 程序的小服务器，已经有点小马拉大车了。现象就是服务崩溃，需要重启！！！
  • 方案：咋办，花钱升级服务器？我有技术我花那钱，闹呢？上PHP呀，这玩意多升服务器资源，况且我在13年就折腾会了，把DDD思想给我架到PHP上，让它爷青回！

三、那，还能干啥？

你觉得视野范围重要吗，如果我不曾如此折腾，可能与我相遇的美好，也都将错过！

哪怕是工作，我也不希望自己是一个工具人。趁年轻、趁着有资源、趁着有时间，多给自己补充一些未来的可能，你现在所学到的所有知识、所结交的所有人脉、所开拓的所有技能，都将会在未来使用上，这些一点一点的点，会聚合成一条条知识线，在撑起你整个人生的页面。

尤其是现在在大厂，可能很多时候自己接触的工作范围是非常局限的，只是知道自己把代码写好了就可以了，关于它是怎么跑起来的，部署到哪里了，域名、VIP、Nginx、CDN、SSL，也并不清楚都是什么。这会有什么问题吗，短时间不会有，甚至还会做的很开心，但时间长了需要你承担更大的职责了呢，又或者需要你自己出去干点啥了呢，怎么在冷启动阶段让自己把这些事拿捏呢？千万别把补充自己的知识当做卷的借口，否则坑的是自己！

其实有一台服务器可以做很多事情，这不像你自己开虚拟机，一个模拟环境和真实环境，还是有很大区别的！

你可以用一台服务器部署你在工作中遇到的知识，也可以是自己想学习搭建的系统，就像小傅哥折腾的这些内容：

  1. 搭博客：mp.weixin.qq.com/s/ZoQ0xAphJ…
  2. 搞论坛：phpwind、Discuz、wordpress(有博客和论坛等模板)
  3. 弄网盘：mp.weixin.qq.com/s/gzUrFexHc…
  4. 聊天室：mp.weixin.qq.com/s/OmXCY4fTf…
  5. 其他的：练习下自己的项目、搞个集群、玩玩ES、弄弄实战、留着接私活给别人部署演示

而这些内容的练习，都能让你把一整套从研发到运维的内容玩透，彻底的了解域名、备案、ssl、宝塔、Linux常用命令等等。

四、卷，咱搞起来！

1. 先neng个服务器

首先，无论你是否有服务器，你都可以跟小傅哥一起学习关于服务器的使用，我们建了个群专门学习服务器，添加我的微信：fustack，备注：服务器学习加群。

如果你还是一个新用户小白，那么可以跟着我的流程一起来，先neng一个便宜的服务器，学习使用即可。这里小傅哥给新人弄了个活动，79元即可买一台一年有效期的服务器，还是比我以前买的便宜多了！

  • 购买链接：www.aliyun.com/minisite/go…
  • 优惠价格：¥1,234.20 ¥69
  • 参与条件：¥79 仅适用于新注册或注册未购买的用户，如果你是老用户可以找下对象或媳妇的帐号注册。

  • 当你购买服务器的时候会看到，地域、实例、操作系统等，地域北京、上海、杭州的网速比较好，张家口的便宜但是网速会比较慢。操作系统默认即可，停机后可以更换。

2. 服务器介绍

在购买完服务器后，等待云平台数分钟初始化服务，完事就可以直接使用配置。如下：

  • 重置密码：点击你的实例，蓝色的这个字母，进入后再右侧有一个，重置实例密码，操作。
  • 远程链接：点击远程链接即可链接到你服务上，它是一个在线的操作。你可以通过本地的软件 xshell 链接到服务上去。
  • 更换系统：如果你对自己默认选择的系统不是很满意或者有其他需求，都可以点击停止系统，之后开始操作系统更换。

3. 系统更换成宝塔镜像

对于服务器系统来说你可以使用Linux命令安装各项服务组件，例如k8s、docker、jdk、tomcat、mysql或者php需要的内容等，但对于实际使用的运维来说，我们更希望运维成本越低越好，所以这里我们选择了宝塔，这样一个服务器运维面板来管理我们的服务器。

在各类的云平台上，包括：百度云、华为云、阿里云、腾讯云，都可以安装宝塔的，有的云平台还会有自己的已经准备好的宝塔镜像，这里我们以阿里云服务为例，把系统停机更好为宝塔。

停机

  • 位置：点击云服务的实例，就可以进入到这个页面
  • 操作：更换系统之前我们需要先进行停机操作，停机后就可以点击更换操作系统了

换系统

选镜像

  • 更换完系统进行确认订单，接下来会跳转到管理后台，这时稍等会，服务器会进行启动。

4. 配置并登录宝塔

远程登录

  • 这一步我们直接在网页上登录了，你也可以使用 xshell 登录公网IP

初始化宝塔

命令：[root@CodeGuide ~]# bt default

配置安全组

  • 宝塔的访问要配置 8888 端口，否则是不能访问到的，这个在服务器的安全组中开放即可。
  • 这里我们为了方便就直接开启全部的了，如果你是实际使用，可不能这样操作，否则很不安全！

登录宝塔

  • 地址：39.96.73.xxx:8888/ - 换成你的地址
  • 账号：用户名和密码已经在控制台打印，你可以复制自己的，登录宝塔后可以修改这个默认的密码

5. 安装阿帕奇和FTP

接下来我们在宝塔中安装一个阿帕奇服务器和FTP，这样就可以部署和访问我们的静态博客了，也就是一个html，如下:

安装 Apache

  • 安装过程中会自动的执行一些命令，这个你不用管，只要默默看着就行了。

安装 FTP

  • 安装 FTP 主要是为了通过本地可以把文件传送到服务器上，比如你的一个静态博客是 html，就可以通过 FTP 传到服务器上。

6. 网站配置

安装了阿帕奇和FTP我们就可以简单的配置一个站点了，有了这个站点就可以访问到我们自己的博客！

创建站点

  • 创建站点的适合如果你还没有申请域名，或者域名还没有备案呢，那么就可以直接把公网IP填写进来。

访问站点

7. 网站内容

  • 在宝塔的文件里，你可以选择第6步骤中添加的站点，在里面找到你的文件，做一些修改动作。这个时候在访问网站，就会发现内容已经是你新的内容了。

8. 域名配置

如果你有域名并已经备案好了，那么在创建站点的时候就可以直接把域名配置上，在访问你的网站的时候就可以通过域名访问了。

  • 添加域名：这个里面小傅哥配置的是已经申请好并备案了的域名，你配置成你的就可以。记得配置好域名后，需要在你的域名服务里，通过A记录把服务器IP映射配置上去。
  • FTP 配置：为了更加方便的上传你的文件，你可以把FTP打开，这样就可以通过FTP传输配置了。
  • 访问地址：blog.itedus.cn - 由于域名不是在阿里云，可能http会监测为未备案，拒绝访问

9. SSL 配置

关于 SSL 的申请可以有很多免费网站提供，也可以在宝塔中申请，如果你是用阿里云服务，可以免费申请20个 SSL 证书，另外如果你的域名和服务都是在阿里云，那么在申请 SSL 可以直接走 DNS 认证，否则你需要把 DNS 信息手动配置到你自己的域名上去。放心这个在申请的时候都有提示，按照说明配置即可

下载证书

因为我们需要把 ssl 配置到宝塔上，所以这里需要把 SSL 下载下来，选择 Apache 格式下载。

配置证书

  • 配置后点击保存即可，另外需要强制开启 HTTPS，否则你的网站访问 http 也能继续访问，就没有意义了。
  • 现在你就可以通过 https，访问自己的博客或者网站了，是不看上去高大上了不少！

10. 其他说明

可能你还希望配置 jdk、tomcat，没关系，在宝塔里你都可以安装，也可以安装 mysql，有了这些入门的内容，剩下的就可以搜索一些通用配置的内容，也可以在阿里云中搜索。

五、嗯，总结一下

  • 讲道理，我希望你遇到我，因为总有一些热情于技术的男人，会带着你搞点啥，搞着搞着你的薪资就上去了！
  • 做技术号主的两年多，我把自己定位成37开，70%开源贡献社区，沉淀、分享，帮助有需要的技术同号。30%做付费内容用于支撑起可以继续完成70%的事情！
  • 我见过自己的粉丝伙伴应届生拿到总包50的，也见过清北出国的，也有高考失利又不断学习考证拿下研究生学位进入大厂管培职位的。但无一例外，这些人从不会给自己设限，甚至是非常自律，很难看到这样的伙伴会去胡扯、瞎聊、当别人的分母，而是不断的“折腾”，折腾学习、折腾感兴趣的事情、折腾自己的爱好，而高薪职位也只不过是过路的风景。
  • 趁着年轻：承遇朝霞、年少正恰，整装戎马、刻印风华！

本文转载自: 掘金

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

0%