使用TiDB把自己写分库分表方案推翻了背景目前痛点目的

发表于 2020-04-05

背景

在日益数据量增长的情况下，影响数据库的读写性能，我们一般会有分库分表的方案和使用newSql方案，newSql如TIDB。那么为什么需要使用TiDB呢？有什么情况下才用TiDB呢？解决传统分库分表的什么问题呢？还会解释一些关键点和踩坑点。下面我会用比较白话的形式解读，当做对TiDB进行推广。

点赞再看，关注公众号：【地藏思维】给大家分享互联网场景设计与架构设计方案
掘金：地藏Kelvin juejin.cn/user/104639…

目前痛点

目前分库表无论使用原生JDBC+ThreadLocal方案，还是使用中间件proxy、还是SDK嵌入代码的形式，即使用sharding-jdbc、zdal、mycat都存在着以下问题。

分库分表算法方案的选型
分库分表后带来的后续维护工作，每次增加节点，都需要申请磁盘、机器
新增节点需要进行停机、然后迁移数据，停机迁移对线上用户造成实时的读写影响。迁移失败还有代码回滚。迁移前还要等mysql没有binlog产生后才能迁移。
分库分表后，跨库一致性问题，都是使用最终一致性，代码维护繁琐。
数据存储压力、数据存放量偏移于某个节点
数据索引查询效率：即使是分库了，要走索引查询，其实还是需要查询多个库后得出结果后汇聚的。
不支持跨库left join其他表
唯一索引在跨库跨表不能保证唯一，场景如：支付流水号。现在分库分表的唯一key都很靠应用层代码控制。
表加字段麻烦。每个库每个表都要加
接入elastic-search需要查多个库才能入elastic-search节点
不能做分页查询

目的

分析TiDB如何解决痛点

TiDB整体架构

TiDB是一种分布式数据库。其实形式上来讲比较像Hadoop的做法，把数据分布在不同的机器上，并且有副本，有负责计算的机器、也由负责存储的机器。

入口层为tidb-server，图中TiDB，是客户端接入的入口，负责处理请求接口，这一层对存储要求不高，用于计算所以对CPU要求高，还有记录每个region的负责的范围。
第二层是PD，负责调度，如zookeeper的形式，负责数据迁移的调度、选举的调度。
第三层是tikv，也叫store，负责真实存储数据的一层。其中tikv由1个或者多个region组成，Region为最小的存储单元，就如JVM G1算法的Region的意思。每个Region将会打散分布在各个tikv下。

数据存储模型

行数据（元数据）
一个表将会由一个或者多个Region存储。不同的表将会在不同的Region，而不是如传统分库那样每个库里的表都是相同。
那么一个表下，每一行数据存储在哪个Region下是如何确定呢？
首先，Region里面是一个Map， key 由 table_id表id、rowid主键组成。如：

t[table_id]_r[row_id]

map的value为表中每行数据的真实数据。

索引数据
索引数据将会在另外一个Region存储，每建一个索引，就会有那个索引对应的Region。它的Map的 key 由 table_id、index_id 以及索引列的值编码组成。如：

t[table_id]_i[index_id][index_value]

value为rowid，这样就能用rowid来找到上面的表数据的位置。
就如mysql按索引查询，会先去找索引记录，再去找到主键聚簇索引来获取真实数据一个逻辑。

数据切分

定位在哪个Region，就是靠Key来算出落在哪个Region里面。和分库分表的根据某个字段来一致性hash算法方案不同。
TiDB的负责行真实数据的Region是使用主键范围来划分的。
有索引情况下，负责索引的Region会根据索引字段范围来划分。
基于Key通过计算，将会得出一个数字，然后按范围划分多个区间，每个区间由一个Region管理。

如：一个表数据主键rowid落在3个Region。
[0,10000)
[10001,20000)
[20001,30000)

这个范围需要数据入表前确定这个规则。

因为Region将会分布在所有TiKV上，也就有多个服务器去存数据，所以利用多机器CPU和磁盘，解决了痛点5存储压力，也解决了痛点1分库分表用哪个算法方案，只需要确定主键范围即可。

后续会说如何扩容。

提高索引效率

现有问题：
传统分库分表的索引都是在每个mysql实例里，跟着表走的。分库分表规则，一般都是根据表中的userid用户字段或组合性较高的字段来做切分库或者表的键，相同的userid将会落在相同的库或者表。

但是上述情况下，表中的索引字段假设为code，则code=”aaa”的可能会因为不同的userid落在不同的库中，需要查询全量的库和表后，再重新聚合，这样就会增加CPU查询的消耗、还有TCP连接握手的消耗。

TiDB解决：
然而TiKV的有专门用于存储索引的Region，它数据结构的Key是由表id+索引id+索引值id来决定的，value是rowid数据行主键，并且一个Region管理一个范围的Key，所以同一个索引同一个值都会在一个Region里面，这样就比较好快速定位相同的索引值的Region，得出对应的rowid，再根据rowid去存储表数据的Region中更快速找到表真实数据。就不需要走全量库的索引查找，因为mysql索引查找机制是先找到索引值，然后再找聚簇的主键后返回整行数据，从而提高性能。

这种做法有点像elastic-search的倒排索引，先根据value值再定位数据原来位置。这里解决痛点6减少索引查询压力。

TIDB特性

提供乐观事务模型和悲观事务模型

在3.0.8之前只有乐观事务模型，都是通过2PC两次提交的方式来进行事务提交。如果开启悲观事务模型，会比较像sharding-jdbc的柔性事务，有重试的功能，但是依然重试过多次（256次）失败仍然会丢失。

1.1 优缺点分析

TiDB 事务有如下优点：

实现原理简单，易于理解。
基于单实例事务实现了跨节点事务。
锁管理实现了去中心化。
但 TiDB 事务也存在以下缺点：
两阶段提交使网络交互增多。
需要一个中心化的版本管理服务。
事务数据量过大时易导致内存暴涨。

1.2 事务的重试

使用乐观事务模型时，在高冲突率的场景中，事务很容易提交失败。而 MySQL 内部使用的是悲观事务模型，在执行 SQL 语句的过程中进行冲突检测，所以提交时很难出现异常。为了兼容 MySQL 的悲观事务行为，TiDB 提供了重试机制。
这种加重试就是悲观事务。

上述解决痛点4，不用再去自己维护跨库处理的事务最终一致性的代码，如A用户转账到B用户，也如商家和买家的情况，商家比较多收入时的交易情况。
虽然重试多次仍然会失败，但是这部分由TiDB处理。如果跨库事务以前的系统有框架处理，那现在就不需要如sharding-jdbc的sdk方式需要靠程序运行时才能重试，不然如果我们程序down机重试就没了。

自动扩容

2.1 Region分裂

Region为最小的存储单元，当数据进入一个Region后达到一定数量，就会开始分裂（默认是超过现有Region负责范围的1/16）。
注意：数据表 Key 的 Range 范围划分，需要提前设置好，TiKV 是根据 Region 的大小动态分裂的。

这里是解决痛点2的每次都需要申请资源，不再运维来做上线前迁移数据，痛点3迁移时又要停机影响生成用户。

因为TiDB作为中间件，不带任何业务属性，所以就不能使用userid等字段来做分片规则的键和自定义算法，使用主键是最通用的选择。（其实我觉得如果TiDB能做到就最好了）

2.2 新增存储节点

新增节点或者分裂Region，都有可能会触发迁移Region，由TiDB自动完成。不再需要入侵代码、或者使用中间件做分库分表逻辑和数据迁移、上线演练，全程交给运维（手动甩锅）。
并且不需要代码服务停机，不需要等没有新sql执行后才能迁移，这个是运行过程中实时迁移数据的。

这里就解决了痛点3停机迁移数据、痛点5存储压力。

副本容灾

每个 Region 负责维护集群的一段连续数据（默认配置下平均约 96 MiB），每份数据会在不同的 Store 存储多个副本（默认配置是 3 副本），每个副本称为 Peer。同一个 Region 的多个 Peer 通过 raft 协议进行数据同步，所以 Peer 也用来指代 raft 实例中的成员。

所以如果有1亿数据，将会由3亿数据落在磁盘中，虽然消耗磁盘，但是提高了可靠性。

TIDB成本

官方推荐至少部署 3 个 TiKV， 3 个 PD，2 个 TiDB。
TiDB需要能使用线程数多的，PD需要CPU比较好的，TiKV需要SSD和CPU比较好的。
在论坛看到大家用的内存都是100G的，磁盘都是2T 的SSD。因为每行数据都总共有3个副本，消耗磁盘多。所以一个系统使用一套TiDB需要不少的成本。
然而这只是一个系统所需，一个项目中有多个系统组成情况下，就消耗更多资源了。并且随着数据日益增多将会越来越多资源。

使用场景

数据量达到一定量级，需要减少查询压力或者连接池不够等等因素后才需要进行。因为官方建议需要有2个tidb-server、至少两个PD、三个tikv，而且tikv需要都是SSD固态硬盘。所以在这种成本下，不一定所有项目都会使用，公司不一定愿意花成本去使用。而在一些数据量小的情况，建议还是使用mysql，等到数据量上来后，再做数据同步到TiDB。
已经分库分表后，希望改为使用TiDB，也能进行合并，需要使用TiDB Data Migration。
先在公司的架构组的项目使用，再到不是核心业务的项目使用，最后铺开给核心项目使用。
入门成本高，实验起来需要成本，因为官方推荐的部署方式需要多台好的机器。

注意事项与坑点

建议使用3.0.4、3.0.8或者4.0.0 （现在是2020年4月2日），不建议使用2.0版本，不然会出现升级不兼容的问题，需要去解决。
在增加节点扩容时，或者Region分裂时，同时有SQL执行insert或者更新数据，如果命中到对应的Region正在迁移，就可能会出现insert或者update出错说“not leader”没有找到对应的位置的意思。但是会有重试的形式，把数据最终提交。
提前设置路由Region的分片范围规则，不然导入数据时会都落在一个节点上。如果你以前的主键数据时雪花算法得出的，那就需要求出最大最小值自己算范围手动设置好范围规则。
TiDB 不支持 SELECT LOCK IN SHARE MODE。使用这个语句执行的时候，效果和没有加锁是一样的，不会阻塞其他事务的读写。
不要再使用Syncer同步数据或迁移到TiDB，因为如果在已经分库分表的情况下，使用Syncer同步，在某个帖子看的说会出问题。建议使用TiDB Data Migration
TiDB无法修改字段类型

总结

其实有了上述功能，就可以减少分库分表的开发、运维维护成本，主要是平常分库分表到一定量要迁移，经常需要监控是否到迁移的量了，迁移时需要演练，迁移时要更新代码或者配置并且停业务是影响最大的。
虽然完成分库分表好像解决了一些问题，但是带来的后续还是很多的，TiDB就给我们解决了上面的问题，这样就可以更加专注的做业务了。

欢迎关注，文章更快一步

我的公众号：地藏思维

地藏思维

掘金：地藏Kelvin

简书：地藏Kelvin

我的Gitee：地藏Kelvin gitee.com/kelvin-cai

本文转载自: 掘金

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

看完这篇 Session、Cookie、Token，和面试官

发表于 2020-04-05

HTTP 协议是一种无状态协议，即每次服务端接收到客户端的请求时，都是一个全新的请求，服务器并不知道客户端的历史请求记录；Session 和 Cookie 的主要目的就是为了弥补 HTTP 的无状态特性。

Session 是什么

客户端请求服务端，服务端会为这次请求开辟一块内存空间，这个对象便是 Session 对象，存储结构为 ConcurrentHashMap。Session 弥补了 HTTP 无状态特性，服务器可以利用 Session 存储客户端在同一个会话期间的一些操作记录。

Session 如何判断是否是同一会话

服务器第一次接收到请求时，开辟了一块 Session 空间（创建了Session对象），同时生成一个 sessionId ，并通过响应头的 **Set-Cookie：JSESSIONID=XXXXXXX **命令，向客户端发送要求设置 Cookie 的响应；客户端收到响应后，在本机客户端设置了一个 **JSESSIONID=XXXXXXX **的 Cookie 信息，该 Cookie 的过期时间为浏览器会话结束；

接下来客户端每次向同一个网站发送请求时，请求头都会带上该 Cookie信息（包含 sessionId ），然后，服务器通过读取请求头中的 Cookie 信息，获取名称为 JSESSIONID 的值，得到此次请求的 sessionId。

Session 的缺点

Session 机制有个缺点，比如 A 服务器存储了 Session，就是做了负载均衡后，假如一段时间内 A 的访问量激增，会转发到 B 进行访问，但是 B 服务器并没有存储 A 的 Session，会导致 Session 的失效。

Cookies 是什么

HTTP 协议中的 Cookie 包括 Web Cookie 和浏览器 Cookie，它是服务器发送到 Web 浏览器的一小块数据。服务器发送到浏览器的 Cookie，浏览器会进行存储，并与下一个请求一起发送到服务器。通常，它用于判断两个请求是否来自于同一个浏览器，例如用户保持登录状态。

HTTP Cookie 机制是 HTTP 协议无状态的一种补充和改良

Cookie 主要用于下面三个目的

会话管理

登陆、购物车、游戏得分或者服务器应该记住的其他内容

个性化

用户偏好、主题或者其他设置

追踪

记录和分析用户行为

Cookie 曾经用于一般的客户端存储。虽然这是合法的，因为它们是在客户端上存储数据的唯一方法，但如今建议使用现代存储 API。Cookie 随每个请求一起发送，因此它们可能会降低性能（尤其是对于移动数据连接而言）。

当接收到客户端发出的 HTTP 请求时，服务器可以发送带有响应的 Set-Cookie 标头，Cookie 通常由浏览器存储，然后将 Cookie 与 HTTP 标头一同向服务器发出请求。

Set-Cookie HTTP 响应标头将 cookie 从服务器发送到用户代理。下面是一个发送 Cookie 的例子

此标头告诉客户端存储 Cookie

现在，随着对服务器的每个新请求，浏览器将使用 Cookie 头将所有以前存储的 Cookie 发送回服务器。

有两种类型的 Cookies，一种是 Session Cookies，一种是 Persistent Cookies，如果 Cookie 不包含到期日期，则将其视为会话 Cookie。会话 Cookie 存储在内存中，永远不会写入磁盘，当浏览器关闭时，此后 Cookie 将永久丢失。如果 Cookie 包含有效期 ，则将其视为持久性 Cookie。在到期指定的日期，Cookie 将从磁盘中删除。

还有一种是 Cookie的 Secure 和 HttpOnly 标记，下面依次来介绍一下

会话 Cookies

上面的示例创建的是会话 Cookie ，会话 Cookie 有个特征，客户端关闭时 Cookie 会删除，因为它没有指定Expires或 Max-Age 指令。

但是，Web 浏览器可能会使用会话还原，这会使大多数会话 Cookie 保持永久状态，就像从未关闭过浏览器一样。

永久性 Cookies

永久性 Cookie 不会在客户端关闭时过期，而是在特定日期（Expires）或特定时间长度（Max-Age）外过期。例如

1	复制代码Set-Cookie: id=a3fWa; Expires=Wed, 21 Oct 2015 07:28:00 GMT;

安全的 Cookie 需要经过 HTTPS 协议通过加密的方式发送到服务器。即使是安全的，也不应该将敏感信息存储在cookie 中，因为它们本质上是不安全的，并且此标志不能提供真正的保护。

HttpOnly 的作用

会话 Cookie 中缺少 HttpOnly 属性会导致攻击者可以通过程序(JS脚本、Applet等)获取到用户的 Cookie 信息，造成用户 Cookie 信息泄露，增加攻击者的跨站脚本攻击威胁。
HttpOnly 是微软对 Cookie 做的扩展，该值指定 Cookie 是否可通过客户端脚本访问。
如果在 Cookie 中没有设置 HttpOnly 属性为 true，可能导致 Cookie 被窃取。窃取的 Cookie 可以包含标识站点用户的敏感信息，如 ASP.NET 会话 ID 或 Forms 身份验证票证，攻击者可以重播窃取的 Cookie，以便伪装成用户或获取敏感信息，进行跨站脚本攻击等。

Domain 和 Path 标识定义了 Cookie 的作用域：即 Cookie 应该发送给哪些 URL。

Domain 标识指定了哪些主机可以接受 Cookie。如果不指定，默认为当前主机(不包含子域名）。如果指定了Domain，则一般包含子域名。

例如，如果设置 Domain=mozilla.org，则 Cookie 也包含在子域名中（如developer.mozilla.org）。

例如，设置 Path=/docs，则以下地址都会匹配：

/docs
/docs/Web/
/docs/Web/HTTP

JSON Web Token 和 Session Cookies 的对比

JSON Web Token ，简称 JWT，它和 Session都可以为网站提供用户的身份认证，但是它们不是一回事。

下面是 JWT 和 Session 不同之处的研究

JWT 和 Session Cookies 的相同之处

在探讨 JWT 和 Session Cookies 之前，有必要需要先去理解一下它们的相同之处。

它们既可以对用户进行身份验证，也可以用来在用户单击进入不同页面时以及登陆网站或应用程序后进行身份验证。

如果没有这两者，那你可能需要在每个页面切换时都需要进行登录了。因为 HTTP 是一个无状态的协议。这也就意味着当你访问某个网页，然后单击同一站点上的另一个页面时，服务器的内存中将不会记住你之前的操作。

因此，如果你登录并访问了你有权访问的另一个页面，由于 HTTP 不会记录你刚刚登录的信息，因此你将再次登录。

JWT 和 Session Cookies 就是用来处理在不同页面之间切换，保存用户登录信息的机制。

也就是说，这两种技术都是用来保存你的登录状态，能够让你在浏览任意受密码保护的网站。通过在每次产生新的请求时对用户数据进行身份验证来解决此问题。

所以 JWT 和 Session Cookies 的相同之处是什么？那就是它们能够支持你在发送不同请求之间，记录并验证你的登录状态的一种机制。

什么是 Session Cookies

Session Cookies 也称为会话 Cookies，在 Session Cookies 中，用户的登录状态会保存在服务器的内存中。当用户登录时，Session 就被服务端安全的创建。

在每次请求时，服务器都会从会话 Cookie 中读取 SessionId，如果服务端的数据和读取的 SessionId 相同，那么服务器就会发送响应给浏览器，允许用户登录。

什么是 Json Web Tokens

Json Web Token 的简称就是 JWT，通常可以称为 Json 令牌。它是RFC 7519 中定义的用于安全的将信息作为 Json 对象进行传输的一种形式。JWT 中存储的信息是经过数字签名的，因此可以被信任和理解。可以使用 HMAC 算法或使用 RSA/ECDSA 的公用/专用密钥对 JWT 进行签名。

使用 JWT 主要用来下面两点

认证(Authorization)：这是使用 JWT 最常见的一种情况，一旦用户登录，后面每个请求都会包含 JWT，从而允许用户访问该令牌所允许的路由、服务和资源。单点登录是当今广泛使用 JWT 的一项功能，因为它的开销很小。
信息交换(Information Exchange)：JWT 是能够安全传输信息的一种方式。通过使用公钥/私钥对 JWT 进行签名认证。此外，由于签名是使用 head 和 payload 计算的，因此你还可以验证内容是否遭到篡改。

JWT 的格式

下面，我们会探讨一下 JWT 的组成和格式是什么

JWT 主要由三部分组成，每个部分用 . 进行分割，各个部分分别是

Header
Payload
Signature

因此，一个非常简单的 JWT 组成会是下面这样

然后我们分别对不同的部分进行探讨。

Header

Header 是 JWT 的标头，它通常由两部分组成：令牌的类型(即 JWT)和使用的 签名算法，例如 HMAC SHA256 或 RSA。

例如

复制代码{
  "alg": "HS256",
  "typ": "JWT"
}

指定类型和签名算法后，Json 块被 Base64Url 编码形成 JWT 的第一部分。

Payload

Token 的第二部分是 Payload，Payload 中包含一个声明。声明是有关实体（通常是用户）和其他数据的声明。共有三种类型的声明：registered, public 和 private 声明。

registered 声明：包含一组建议使用的预定义声明，主要包括

ISS	签发人
iss (issuer)	签发人
exp (expiration time)	过期时间
sub (subject)	主题
aud (audience)	受众
nbf (Not Before)	生效时间
iat (Issued At)	签发时间
jti (JWT ID)	编号

public 声明：公共的声明，可以添加任何的信息，一般添加用户的相关信息或其他业务需要的必要信息，但不建议添加敏感信息，因为该部分在客户端可解密。
private 声明：自定义声明，旨在在同意使用它们的各方之间共享信息，既不是注册声明也不是公共声明。

例如

复制代码{
  "sub": "1234567890",
  "name": "John Doe",
  "admin": true
}

然后 payload Json 块会被Base64Url 编码形成 JWT 的第二部分。

signature

JWT 的第三部分是一个签证信息，这个签证信息由三部分组成

header (base64后的)
payload (base64后的)
secret

比如我们需要 HMAC SHA256 算法进行签名

复制代码HMACSHA256(
  base64UrlEncode(header) + "." +
  base64UrlEncode(payload),
  secret)

签名用于验证消息在此过程中没有更改，并且对于使用私钥进行签名的令牌，它还可以验证 JWT 的发送者的真实身份

拼凑在一起

现在我们把上面的三个由点分隔的 Base64-URL 字符串部分组成在一起，这个字符串可以在 HTML 和 HTTP 环境中轻松传递这些字符串。

下面是一个完整的 JWT 示例，它对 header 和 payload 进行编码，然后使用 signature 进行签名

1	复制代码eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiYWRtaW4iOnRydWV9.TJVA95OrM7E2cBab30RMHrHDcEfxjoYZgeFONFh7HgQ

如果想自己测试编写的话，可以访问 JWT 官网 jwt.io/#debugger-i…

JWT 和 Session Cookies 的不同

JWT 和 Session Cookies 都提供安全的用户身份验证，但是它们有以下几点不同

密码签名

JWT 具有加密签名，而 Session Cookies 则没有。

JSON 是无状态的

JWT 是无状态的，因为声明被存储在客户端，而不是服务端内存中。

身份验证可以在本地进行，而不是在请求必须通过服务器数据库或类似位置中进行。这意味着可以对用户进行多次身份验证，而无需与站点或应用程序的数据库进行通信，也无需在此过程中消耗大量资源。

可扩展性

Session Cookies 是存储在服务器内存中，这就意味着如果网站或者应用很大的情况下会耗费大量的资源。由于 JWT 是无状态的，在许多情况下，它们可以节省服务器资源。因此 JWT 要比 Session Cookies 具有更强的可扩展性。

JWT 支持跨域认证

Session Cookies 只能用在单个节点的域或者它的子域中有效。如果它们尝试通过第三个节点访问，就会被禁止。如果你希望自己的网站和其他站点建立安全连接时，这是一个问题。

使用 JWT 可以解决这个问题，使用 JWT 能够通过多个节点进行用户认证，也就是我们常说的跨域认证。

JWT 和 Session Cookies 的选型

我们上面探讨了 JWT 和 Cookies 的不同点，相信你也会对选型有了更深的认识，大致来说

对于只需要登录用户并访问存储在站点数据库中的一些信息的中小型网站来说，Session Cookies 通常就能满足。

如果你有企业级站点，应用程序或附近的站点，并且需要处理大量的请求，尤其是第三方或很多第三方（包括位于不同域的API），则 JWT 显然更适合。

后记

前两天面试的时候问到了这个题，所以写篇文章总结一下，还问到了一个面试题，禁用 Cookies，如何使用 Session ？网上百度了一下，发现这是 PHP 的面试题……

但还是选择了解了一下，如何禁用 Cookies 后，使用 Session

如果禁用了 Cookies，服务器仍会将 sessionId 以 cookie 的方式发送给浏览器，但是，浏览器不再保存这个cookie (即sessionId) 了。
如果想要继续使用 session，需要采用 URL 重写 的方式来实现，可以参考 www.cnblogs.com/Renyi-Fan/p…

我去，你竟然还在用 try–catch-finally

发表于 2020-04-04

二哥，你之前那篇我去 switch 的文章也特么太有趣了，读完后意犹未尽啊，要不要再写一篇啊？虽然用的是 Java 13 的语法，对旧版本不太友好。但谁能保证 Java 不会再来一次重大更新呢，就像 Java 8 那样，活生生地把 Java 6 拍死在了沙滩上。Java 8 是香，但早晚要升级，我挺你，二哥，别在乎那些反对的声音。

这是读者 Alice 上周特意给我发来的信息，真令我动容。的确，上次的“我去”阅读量杠杠的，几个大号都转载了，包括 CSDN，次条当天都 1.5 万阅读。但比如“还以为你有什么新特技，没想到用的是 Java 13”这类批评的声音也不在少数。

不过我的心一直很大。从我写第一篇文章至今，被喷的次数就好像头顶上茂密的发量一样，数也数不清。所以我决定再接再厉，带来新的一篇“我去”。

这次不用远程 review 了，因为我们公司也复工了。这次 review 的代码仍然是小王的，他编写的大部分代码都很漂亮，严谨的同时注释也很到位，这令我非常满意。但当我看到他没用 try-with-resources 时，还是忍不住破口大骂：“我擦，小王，你丫的竟然还在用 try–catch-finally！”

来看看小王写的代码吧。

复制代码public class Trycatchfinally {
    public static void main(String[] args) {
        BufferedReader br = null;
        try {
            br = new BufferedReader(new FileReader("/牛逼.txt"));
            String str = null;
            while ((str =br.readLine()) != null) {
                System.out.println(str);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

咦，感觉这段代码很完美无缺啊，try–catch-finally 用得中规中矩，尤其是文件名 牛逼.txt 很亮。不用写注释都能明白这段代码是干嘛的：在 try 块中读取文件中的内容，并一行一行地打印到控制台。如果文件找不到或者出现 IO 读写错误，就在 catch 中捕获并打印错误的堆栈信息。最后，在 finally 中关闭缓冲字符读取器对象 BufferedReader，有效杜绝了资源未被关闭的情况下造成的严重性能后果。

在 Java 7 之前，try–catch-finally 的确是确保资源会被及时关闭的最佳方法，无论程序是否会抛出异常。

但是呢，有经验的读者会从上面这段代码中发现 2 个严重的问题：

1）文件名“牛逼.txt”包含了中文，需要通过 java.net.URLDecoder 类的 decode() 方法对其转义，否则这段代码在运行时铁定要抛出文件找不到的异常。

2）如果直接通过 new FileReader("牛逼.txt") 创建 FileReader 对象，“牛逼.txt”需要和项目的 src 在同一级目录下，否则同样会抛出文件找不到的异常。但大多数情况下，（配置）文件会放在 resources 目录下，便于编译后文件出现在 classes 目录下，见下图。

为了解决以上 2 个问题，我们需要对代码进行优化：

复制代码public class TrycatchfinallyDecoder {
    public static void main(String[] args) {
        BufferedReader br = null;
        try {
            String path = TrycatchfinallyDecoder.class.getResource("/牛逼.txt").getFile();
            String decodePath = URLDecoder.decode(path,"utf-8");
            br = new BufferedReader(new FileReader(decodePath));

            String str = null;
            while ((str =br.readLine()) != null) {
                System.out.println(str);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

运行这段代码，程序就可以将文件中的内容正确输出到控制台。但如果你对“整洁”这个词心生向往的话，会感觉这段代码非常臃肿，尤其是 finally 中的代码，就好像一个灌了 12 瓶雪花啤酒的大肚腩。

网上看到一幅 Python 程序员调侃 Java 程序员的神图，直接 copy 过来（侵删），逗你一乐：

况且，try–catch-finally 至始至终存在一个严重的隐患：try 中的 br.readLine() 有可能会抛出 IOException，finally 中的 br.close() 也有可能会抛出 IOException。假如两处都不幸地抛出了 IOException，那程序的调试任务就变得复杂了起来，到底是哪一处出了错误，就需要花一番功夫，这是我们不愿意看到的结果。

为了模拟上述情况，我们来自定义一个类 MyfinallyReadLineThrow，它有两个方法，分别是 readLine() 和 close()，方法体都是主动抛出异常。

复制代码class MyfinallyReadLineThrow {
    public void close() throws Exception {
        throw new Exception("close");
    }

    public void readLine() throws Exception {
        throw new Exception("readLine");
    }
}

然后我们在 main() 方法中使用 try-finally 的方式调用 MyfinallyReadLineThrow 的 readLine() 和 close() 方法：

复制代码public class TryfinallyCustomReadLineThrow {
    public static void main(String[] args) throws Exception {
        MyfinallyReadLineThrow myThrow = null;
        try {
            myThrow = new MyfinallyReadLineThrow();
            myThrow.readLine();
        } finally {
            myThrow.close();
        }
    }
}

运行上述代码后，错误堆栈如下所示：

1
2
3

复制代码Exception in thread "main" java.lang.Exception: close
	at com.cmower.dzone.trycatchfinally.MyfinallyOutThrow.close(TryfinallyCustomOutThrow.java:17)
	at com.cmower.dzone.trycatchfinally.TryfinallyCustomOutThrow.main(TryfinallyCustomOutThrow.java:10)

readLine() 方法的异常信息竟然被 close() 方法的堆栈信息吃了，这必然会让我们误以为要调查的目标是 close() 方法而不是 readLine()——尽管它也是应该怀疑的对象。

但自从有了 try-with-resources，这些问题就迎刃而解了，只要需要释放的资源（比如 BufferedReader）实现了 AutoCloseable 接口。有了解决方案之后，我们来对之前的 finally 代码块进行瘦身。

复制代码try (BufferedReader br = new BufferedReader(new FileReader(decodePath));) {
    String str = null;
    while ((str =br.readLine()) != null) {
        System.out.println(str);
    }
} catch (IOException e) {
    e.printStackTrace();
}

你瞧，finally 代码块消失了，取而代之的是把要释放的资源写在 try 后的 () 中。如果有多个资源（BufferedReader 和 PrintWriter）需要释放的话，可以直接在 () 中添加。

复制代码try (BufferedReader br = new BufferedReader(new FileReader(decodePath));
     PrintWriter writer = new PrintWriter(new File(writePath))) {
    String str = null;
    while ((str =br.readLine()) != null) {
        writer.print(str);
    }
} catch (IOException e) {
    e.printStackTrace();
}

如果你想释放自定义资源的话，只要让它实现 AutoCloseable 接口，并提供 close() 方法即可。

复制代码public class TrywithresourcesCustom {
    public static void main(String[] args) {
        try (MyResource resource = new MyResource();) {
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

class MyResource implements AutoCloseable {
    @Override
    public void close() throws Exception {
        System.out.println("关闭自定义资源");
    }
}

代码运行后输出的结果如下所示：

1	复制代码关闭自定义资源

是不是很神奇？我们在 try () 中只是 new 了一个 MyResource 的对象，其他什么也没干，但偏偏 close() 方法中的输出语句执行了。想要知道为什么吗？来看看反编译后的字节码吧。

复制代码class MyResource implements AutoCloseable {
    MyResource() {
    }

    public void close() throws Exception {
        System.out.println("关闭自定义资源");
    }
}

public class TrywithresourcesCustom {
    public TrywithresourcesCustom() {
    }

    public static void main(String[] args) {
        try {
            MyResource resource = new MyResource();
            resource.close();
        } catch (Exception var2) {
            var2.printStackTrace();
        }

    }
}

咦，编译器竟然主动为 try-with-resources 进行了变身，在 try 中调用了 close() 方法。

接下来，我们在自定义类中再添加一个 out() 方法，

复制代码class MyResourceOut implements AutoCloseable {
    @Override
    public void close() throws Exception {
        System.out.println("关闭自定义资源");
    }

    public void out() throws Exception{
        System.out.println("沉默王二，一枚有趣的程序员");
    }
}

这次，我们在 try 中调用一下 out() 方法：

复制代码public class TrywithresourcesCustomOut {
    public static void main(String[] args) {
        try (MyResourceOut resource = new MyResourceOut();) {
            resource.out();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

再来看一下反编译的字节码：

复制代码public class TrywithresourcesCustomOut {
    public TrywithresourcesCustomOut() {
    }

    public static void main(String[] args) {
        try {
            MyResourceOut resource = new MyResourceOut();

            try {
                resource.out();
            } catch (Throwable var5) {
                try {
                    resource.close();
                } catch (Throwable var4) {
                    var5.addSuppressed(var4);
                }

                throw var5;
            }

            resource.close();
        } catch (Exception var6) {
            var6.printStackTrace();
        }

    }
}

这次，catch 块中主动调用了 resource.close()，并且有一段很关键的代码 var5.addSuppressed(var4)。它有什么用处呢？当一个异常被抛出的时候，可能有其他异常因为该异常而被抑制住，从而无法正常抛出。这时可以通过 addSuppressed() 方法把这些被抑制的方法记录下来。被抑制的异常会出现在抛出的异常的堆栈信息中，也可以通过 getSuppressed() 方法来获取这些异常。这样做的好处是不会丢失任何异常，方便我们开发人员进行调试。

哇，有没有想到我们之前的那个例子——在 try-finally 中，readLine() 方法的异常信息竟然被 close() 方法的堆栈信息吃了。现在有了 try-with-resources，再来看看作用和 readLine() 方法一致的 out() 方法会不会被 close() 吃掉。

在 close() 和 out() 方法中直接抛出异常：

复制代码class MyResourceOutThrow implements AutoCloseable {
    @Override
    public void close() throws Exception {
        throw  new Exception("close()");
    }

    public void out() throws Exception{
        throw new Exception("out()");
    }
}

调用这 2 个方法：

复制代码public class TrywithresourcesCustomOutThrow {
    public static void main(String[] args) {
        try (MyResourceOutThrow resource = new MyResourceOutThrow();) {
            resource.out();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

程序输出的结果如下所示：

复制代码java.lang.Exception: out()
	at com.cmower.dzone.trycatchfinally.MyResourceOutThrow.out(TrywithresourcesCustomOutThrow.java:20)
	at com.cmower.dzone.trycatchfinally.TrywithresourcesCustomOutThrow.main(TrywithresourcesCustomOutThrow.java:6)
	Suppressed: java.lang.Exception: close()
		at com.cmower.dzone.trycatchfinally.MyResourceOutThrow.close(TrywithresourcesCustomOutThrow.java:16)
		at com.cmower.dzone.trycatchfinally.TrywithresourcesCustomOutThrow.main(TrywithresourcesCustomOutThrow.java:5)

瞧，这次不会了，out() 的异常堆栈信息打印出来了，并且 close() 方法的堆栈信息上加了一个关键字 Suppressed。一目了然，不错不错，我喜欢。

总结一下，在处理必须关闭的资源时，始终有限考虑使用 try-with-resources，而不是 try–catch-finally。前者产生的代码更加简洁、清晰，产生的异常信息也更靠谱。答应我好不好？别再用 try–catch-finally 了。

鸣谢

好了，我亲爱的读者朋友，以上就是本文的全部内容了，是不是感觉又学到了新知识？我是沉默王二，一枚有趣的程序员。原创不易，莫要白票，请你为本文点赞个吧，这将是我写作更多优质文章的最强动力。

如果觉得文章对你有点帮助，请微信搜索「 沉默王二 」第一时间阅读，回复【666】更有我为你精心准备的 500G 高清教学视频（已分门别类）。本文 GitHub 已经收录，有大厂面试完整考点，欢迎 Star。

本文转载自: 掘金

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

如何保证分布式系统中 ID 的唯一性前言正文 By th

发表于 2020-04-03

前言

前面我在公众号提了一个问题，如何保证分布式系统中 ID 的唯一性，今天我就来给大家解答一下，如果有什么补充或者疑问的可以到我的公众号【6曦轩】留言，看到的话会尽快回复。

系统唯一 ID 是我们在设计一个系统的时候常常会遇见的问题，也常常为这个问题而纠结。生成 ID 的方法有很多，适应不同的场景、需求以及性能要求。所以有些比较复杂的系统会有多个 ID 生成的策略。下面就介绍一些常见的 ID 生成策略。

正文

数据库自增长序列或字段

最常见的方式。利用数据库，全数据库唯一。

优点：

1）简单，代码方便，性能可以接受;
2）数字ID天然排序，对分页或者需要排序的结果很有帮助。

缺点：

1）不同数据库语法和实现不同，数据库迁移的时候或多数据库版本支持的时候需要处理;
2）在单个数据库或读写分离或一主多从的情况下，只有一个主库可以生成。有单点故障的风险;
3）在性能达不到要求的情况下，比较难于扩展;
4）如果遇见多个系统需要合并或者涉及到数据迁移会相当痛苦;
5）分表分库的时候会有麻烦。

优化方案：

1）针对主库单点，如果有多个Master库，则每个Master库设置的起始数字不一样，步长一样，可以是Master的个数。比如：Master1 生成的是 1，4，7，10，Master2生成的是2,5,8,11 Master3生成的是 3,6,9,12。这样就可以有效生成集群中的唯一ID，也可以大大降低ID生成数据库操作的负载。

UUID

常见的方式。可以利用数据库也可以利用程序生成，一般来说全球唯一。

优点：

1）简单，代码方便。

2）生成ID性能非常好，基本不会有性能问题。

3）全球唯一，在遇见数据迁移，系统数据合并，或者数据库变更等情况下，可以从容应对。

缺点：

1）没有排序，无法保证趋势递增;

2）UUID往往是使用字符串存储，查询的效率比较低;

3）存储空间比较大，如果是海量数据库，就需要考虑存储量的问题;

4）传输数据量大;

5）不可读。

UUID的变种

1）为了解决UUID不可读，可以使用UUID to Int64的方法。

复制代码/// <summary>
/// 根据GUID获取唯一数字序列
/// </summary>
public static long GuidToInt64()
{
    byte[] bytes = Guid.NewGuid().ToByteArray();
    return BitConverter.ToInt64(bytes, 0);
}

2）为了解决UUID无序的问题，NHibernate在其主键生成方式中提供了Comb算法（combined guid/timestamp）。保留GUID的10个字节，用另6个字节表示GUID生成的时间（DateTime）。

复制代码/// <summary> 
/// Generate a new <see cref="Guid"/> using the comb algorithm. 
/// </summary> 
private Guid GenerateComb()
{
    byte[] guidArray = Guid.NewGuid().ToByteArray();
 
    DateTime baseDate = new DateTime(1900, 1, 1);
    DateTime now = DateTime.Now;
 
    // Get the days and milliseconds which will be used to build    
    //the byte string    
    TimeSpan days = new TimeSpan(now.Ticks - baseDate.Ticks);
    TimeSpan msecs = now.TimeOfDay;
 
    // Convert to a byte array        
    // Note that SQL Server is accurate to 1/300th of a    
    // millisecond so we divide by 3.333333    
    byte[] daysArray = BitConverter.GetBytes(days.Days);
    byte[] msecsArray = BitConverter.GetBytes((long)
      (msecs.TotalMilliseconds / 3.333333));
 
    // Reverse the bytes to match SQL Servers ordering    
    Array.Reverse(daysArray);
    Array.Reverse(msecsArray);
 
    // Copy the bytes into the guid    
    Array.Copy(daysArray, daysArray.Length - 2, guidArray,
      guidArray.Length - 6, 2);
    Array.Copy(msecsArray, msecsArray.Length - 4, guidArray,
      guidArray.Length - 4, 4);
 
    return new Guid(guidArray);
}

用上面的算法测试一下，得到如下的结果：作为比较，前面3个是使用 COMB 算法得出的结果，最后12个字符串是时间序（统一毫秒生成的3个 UUID ），过段时间如果再次生成，则12个字符串会比图示的要大。后面3个是直接生成的 GUID。

如果想把时间序放在前面，可以生成后改变 12 个字符串的位置，也可以修改算法类的最后两个 Array.Copy。

Redis生成ID

当使用数据库来生成 ID 性能不够要求的时候，我们可以尝试使用 Redis 来生成 ID。这主要依赖于 Redis 是单线程的，所以也可以用生成全局唯一的 ID。可以用 Redis 的原子操作 INCR 和 INCRBY 来实现。

可以使用Redis集群来获取更高的吞吐量。假如一个集群中有 5 台 Redis。可以初始化每台 Redis 的值分别是1, 2, 3, 4, 5，然后步长都是 5。各个 Redis 生成的 ID 为：

A：1,6,11,16,21

B：2,7,12,17,22

C：3,8,13,18,23

D：4,9,14,19,24

E：5,10,15,20,25

这个，随便负载到哪个机确定好，未来很难做修改。但是 3-5 台服务器基本能够满足上，都可以获得不同的 ID。但是步长和初始值一定是事先需要了, 使用 Redis 集群也可以防止单点故障的问题。

另外，比较适合使用 Redis 来生成每天从0开始的流水号。比如订单号=日期+当日自增长号。可以每天在 Redis 中生成一个 Key，使用 INCR 进行累加。

优点：

1）不依赖于数据库，灵活方便，且性能优于数据库。

2）数字ID天然排序，对分页或者需要排序的结果很有帮助。

缺点：

1）如果系统中没有Redis，还需要引入新的组件，增加系统复杂度。

2）需要编码和配置的工作量比较大。

Twitter的snowflake算法

snowflake 是 Twitter 开源的分布式 ID 生成算法，结果是一个 long 型的 ID。其核心思想是：使用41bit作为毫秒数，10 bit 作为机器的 ID（5个 bit 是数据中心，5个 bit的机器ID），12 bit作为毫秒内的流水号（意味着每个节点在每毫秒可以产生 4096 个 ID），最后还有一个符号位，永远是0。具体实现的代码可以参看https://github.com/twitter/snowflake。

C#代码如下：

复制代码/// <summary>
    /// From: https://github.com/twitter/snowflake
    /// An object that generates IDs.
    /// This is broken into a separate class in case
    /// we ever want to support multiple worker threads
    /// per process
    /// </summary>
    public class IdWorker
    {
        private long workerId;
        private long datacenterId;
        private long sequence = 0L;

        private static long twepoch = 1288834974657L;

        private static long workerIdBits = 5L;
        private static long datacenterIdBits = 5L;
        private static long maxWorkerId = -1L ^ (-1L << (int)workerIdBits);
        private static long maxDatacenterId = -1L ^ (-1L << (int)datacenterIdBits);
        private static long sequenceBits = 12L;

        private long workerIdShift = sequenceBits;
        private long datacenterIdShift = sequenceBits + workerIdBits;
        private long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;
        private long sequenceMask = -1L ^ (-1L << (int)sequenceBits);

        private long lastTimestamp = -1L;
        private static object syncRoot = new object();

        public IdWorker(long workerId, long datacenterId)
        {

            // sanity check for workerId
            if (workerId > maxWorkerId || workerId < 0)
            {
                throw new ArgumentException(string.Format("worker Id can't be greater than %d or less than 0", maxWorkerId));
            }
            if (datacenterId > maxDatacenterId || datacenterId < 0)
            {
                throw new ArgumentException(string.Format("datacenter Id can't be greater than %d or less than 0", maxDatacenterId));
            }
            this.workerId = workerId;
            this.datacenterId = datacenterId;
        }

        public long nextId()
        {
            lock (syncRoot)
            {
                long timestamp = timeGen();

                if (timestamp < lastTimestamp)
                {
                    throw new ApplicationException(string.Format("Clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
                }

                if (lastTimestamp == timestamp)
                {
                    sequence = (sequence + 1) & sequenceMask;
                    if (sequence == 0)
                    {
                        timestamp = tilNextMillis(lastTimestamp);
                    }
                }
                else
                {
                    sequence = 0L;
                }

                lastTimestamp = timestamp;

                return ((timestamp - twepoch) << (int)timestampLeftShift) | (datacenterId << (int)datacenterIdShift) | (workerId << (int)workerIdShift) | sequence;
            }
        }

        protected long tilNextMillis(long lastTimestamp)
        {
            long timestamp = timeGen();
            while (timestamp <= lastTimestamp)
            {
                timestamp = timeGen();
            }
            return timestamp;
        }

        protected long timeGen()
        {
            return (long)(DateTime.UtcNow - new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalMilliseconds;
        }
    }

测试代码如下：

复制代码private static void TestIdWorker()
        {
            HashSet<long> set = new HashSet<long>();
            IdWorker idWorker1 = new IdWorker(0, 0);
            IdWorker idWorker2 = new IdWorker(1, 0);
            Thread t1 = new Thread(() => DoTestIdWoker(idWorker1, set));
            Thread t2 = new Thread(() => DoTestIdWoker(idWorker2, set));
            t1.IsBackground = true;
            t2.IsBackground = true;

            t1.Start();
            t2.Start();
            try
            {
                Thread.Sleep(30000);
                t1.Abort();
                t2.Abort();
            }
            catch (Exception e)
            {
            }

            Console.WriteLine("done");
        }

        private static void DoTestIdWoker(IdWorker idWorker, HashSet<long> set)
        {
            while (true)
            {
                long id = idWorker.nextId();
                if (!set.Add(id))
                {
                    Console.WriteLine("duplicate:" + id);
                }

                Thread.Sleep(1);
            }
        }

snowflake算法可以根据自身项目的需要进行一定的修改。比如估算未来的数据中心个数，每个数据中心的机器数以及统一毫秒可以能的并发数来调整在算法中所需要的bit数。

优点：

1）不依赖于数据库，灵活方便，且性能优于数据库。

2）ID按照时间在单机上是递增的。

缺点：

1）在单机上是递增的，但是由于涉及到分布式环境，每台机器上的时钟不可能完全同步，也许有时候也会出现不是全局递增的情况。

利用zookeeper生成唯一ID

zookeeper 主要通过其 znode 数据版本来生成序列号，可以生成 32 位和 64 位的数据版本号，客户端可以使用这个版本号来作为唯一的序列号。
很少会使用 zookeeper 来生成唯一 ID。主要是由于需要依赖 zookeeper，并且是多步调用API，如果在竞争较大的情况下，需要考虑使用分布式锁。因此，性能在高并发的分布式环境下，也不甚理想。

MongoDB的ObjectId

MongoDB 的 ObjectId 和 snowflake 算法类似。它设计成轻量型的，不同的机器都能用全局唯一的同种方法方便地生成它。MongoDB 从一开始就设计用来作为分布式数据库，处理多个节点是一个核心要求。使其在分片环境中要容易生成得多。

其格式如下：

前4 个字节是从标准纪元开始的时间戳，单位为秒。时间戳，与随后的5 个字节组合起来，提供了秒级别的唯一性。由于时间戳在前，这意味着 ObjectId 大致会按照插入的顺序排列。这对于某些方面很有用，如将其作为索引提高效率。这4 个字节也隐含了文档创建的时间。绝大多数客户端类库都会公开一个方法从ObjectId 获取这个信息。
接下来的3 字节是所在主机的唯一标识符。通常是机器主机名的散列值。这样就可以确保不同主机生成不同的ObjectId，不产生冲突。
为了确保在同一台机器上并发的多个进程产生的 ObjectId 是唯一的，接下来的两字节来自产生 ObjectId 的进程标识符（PID）。
前9 字节保证了同一秒钟不同机器不同进程产生的 ObjectId 是唯一的。后3 字节就是一个自动增加的计数器，确保相同进程同一秒产生的 ObjectId 也是不一样的。同一秒钟最多允许每个进程拥有2563（16 777 216）个不同的 ObjectId。

实现的源码可以到 MongoDB 官方网站下载。

By the way

有问题？可以给我留言或私聊
有收获？那就顺手点个赞呗~

当然，也可以到我的公众号下「6曦轩」，

回复“学习”，即可领取一份
【Java工程师进阶架构师的视频教程】~

回复“面试”，可以获得：
【本人呕心沥血整理的 Java 面试题】

回复“MySQL脑图”，可以获得
【MySQL 知识点梳理高清脑图】

曦轩我是科班出身的程序员，php，Android以及硬件方面都做过，不过最后还是选择专注于做 Java，所以有啥问题可以到公众号提问讨论（技术情感倾诉都可以哈哈哈），看到的话会尽快回复，希望可以跟大家共同学习进步，关于服务端架构，Java 核心知识解析，职业生涯，面试总结等文章会不定期坚持推送输出，欢迎大家关注~

在这里插入图片描述

本文转载自: 掘金

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

🔥JVM从入门到入土之实战JVM调优(二) 前言

发表于 2020-04-03

前言

文本已收录至我的GitHub仓库，欢迎Star：github.com/bin39232820…

「种一棵树最好的时间是十年前，其次是现在」鼓励大家在技术的路上写博客

絮叨

前面的章节

昨天我们做了一个基本的调优，但是你发现没有我们仅仅是说明了新生代的调优，今天我们就来看看老年代应该怎么来搞

前文回顾

上一篇文章我们已经给大家介绍了一个每日百万活跃以及亿圾请求的案例背景，同时采用电商大促期间高峰的下单场景，作为我们JVM优化分析的一个场景，推测出在大促销峰期，每秒每台机器会有300个下单

进而推测每秒会使用60MB的内存，然后根据这个背景推算出来我们一台4核8G的机器上，应该如何合理的分配内存，

进而保证可以每隔20秒一次新生代GC后的100MB左右存活对象，会进入200MB的Survivor区域内，一般不会因为Survivor赛不下或者动态年龄判断进入老年代。

同时还根据Minor GC 的频率，合理降低了大龄对象进入老年代的年龄，尽快让一些长期存活的对象赶紧进入老年代，不要停留在新生代，如下图所示

此时的JVM参数如下所示

在案例背景下什么时候进入老年代？

接下来我们分析，在目前优化好的背景下，一般什么情况下会让一些对象进入老年代呢？

首先第一种情况，就是-XX:Max TenuringThreshold=5这个参数会让在一两分钟内连续躲过5次GC的对象，直接进入老年代

这种对象一般是@Service @Controller之类的注解标志的那种业务逻辑组件，

这种对象一般一个系统最多就是几十MB

所以此时长期存活对象就会进入老年代中，如下图所示

此外，按照我们JVM参数，如果分配一个超过1MB的大对象，比如你说你创建一个大的数组或者一个大的List,那么这些对象会直接进入老年代。

但是这种情况假设我们的场景没有，所以忽略不计

此外还有就是Minor GC 过后可能存活的对象超过200MB 放不下进入老年区的，或者一下子占到Surviovr的50%，此时会有一些对象进入老年代

但是我们之前对新生代的优化，就是避免这种情况，但是也有可能刚好这么多，就进入了老年代了

我们可以做一个假设，大概就是这个订单系统促销期间，每隔5分钟会在GC之后有一小批对象进入老年代，大概200M左右，此时的JVM内存如下所示

大促销期间多久会触发一次Full GC

接着我们来研究一下多久会触发Full GC ,

首先我们看Full GC触发的条件

没有打开 -XX:HandlePromotionFailure选型，结果老年代可用内存最多也就1G，新生代你对象总大小最多是1.8G，那么会导致每次Mino GC前一检查，都发现老年代的可用内存<新生代总对象大小，这样会导致每次Minor GC前都触发Full GC
当然在JDK 1.6之后废弃了这个参数，其实只要看下面2个条件就好了
每次Minor GC之前，都检查一下老年代可用内存空间< 历代Minor GC后升入老年代的平均大小，其实按照我们的设定背景，要很多次的Minor GC 才有可能碰巧有200M的对象进入老年代，所以历代进入老年代的平均对象的大小，基本上是很小的
可能某次Minor GC后要升入老年代的对象有几百M，但是老年代的可用空间不足了
设置了 -XX:CMSInitiatingOccupancyFaction参数，比如设置了92%，那么此前几个条件可能没有满足，但是刚好总内存超过了92%，也会进行Full GC。

其实在真正系统运行期间，可能会慢慢的有对象进入老年代，但是因为新生代我们优化过了内存分配，所以对象进入老年代的速度很慢的

所以很可能是在系统运行半小时到1小时之后，才会有接近1G的对象进入老年代

此时只要满足上述 2 3 4其中一个，就好触发Full GC

大家可以思考一下，我们假设再大促销期间，订单系统运行一小时之后，才触发Full GC,这种问题应该是不大的

因为这个是高峰期，如果是平时的话几小时才有可能一次Full GC 所以说这样的设计是可以的

老年代GC的时候会发生 Concurrent Mode Failure吗？

经过前面的推算，我们基本可知道，假设就是订单系统运行1小时后，老年代就有900MB的对象了，剩下的空间只有100MB了，此时就会触发一次Full GC,如下图

但是有一个问题，就是CMS再垃圾回收的时候，尤其是并发清理期间，系统程序是可以并发进行的，所以老年代的空闲空间只剩下100M了

然后此时系统还在不停的创建对象，万一此时触发了一个条件，有200M要进入老年代，此时会怎么样呢？

此时就会触发Concurrent Mode Failure 问题，因为老年代没有足够的内存来存放这200M对象，此时就会导致进入Stop the World,然后切换CMS为Serial Old ,直接禁止程序运行，然后单线程去进行老年代的垃圾回收，会收掉之后，再让系统继续运行，如下图

但是这种概率是很低的的，我们就不用去理会了

CMS垃圾回收之后进行内存碎片的整理的频率应该多高？

在CMS完成Full GC之后，一般需要执行内存的碎片整理，可以设置多少次Full GC 之后执行一次对内存碎片的整理，但是我们能有必要去修改这些参数吗

其实没有必要，因为通过上面的分析，在大促销的高峰期，Full GC可能也就是1小时，然后高峰期过去，可能几小时才有一次Full GC

所以保持默认的设置就可以了，每次Full GC之后整理一次内存碎片就好了，目前的JVM的参数如下

总结
–

其实对很多的普通的Java系统来说，只要对系统运行期间的内存的使用模型做好估计，然后分配好内存，尽量让Minor GC之后的存活对象在Survior不要去老年代，然后其余的参数不必要做过多的优化，系统基本上就不会太差

结尾

大家还是要自己多去尝试，自己去做过才有经验。

因为博主也是一个开发萌新我也是一边学一边写我有个目标就是一周二到三篇希望能坚持个一年吧希望各位大佬多提意见，让我多学习，一起进步。

日常求赞

好了各位，以上就是这篇文章的全部内容了，能看到这里的人呀，都是「真粉」。

创作不易，各位的支持和认可，就是我创作的最大动力，我们下篇文章见

六脉神剑 | 文【原创】如果本篇博客有任何错误，请批评指教，不胜感激！

本文转载自: 掘金

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

【SpringBoot WEB 系列】SSE 服务器发送事件

发表于 2020-04-02

【SpringBoot WEB系列】SSE 服务器发送事件详解

SSE 全称Server Sent Event，直译一下就是服务器发送事件，一般的项目开发中，用到的机会不多，可能很多小伙伴不太清楚这个东西，到底是干啥的，有啥用

本文主要知识点如下：

SSE 扫盲，应用场景分析
借助异步请求实现 sse 功能，加深概念理解
使用SseEmitter实现一个简单的推送示例

I. SSE 扫盲

对于 sse 基础概念比较清楚的可以跳过本节

1. 概念介绍

sse(Server Sent Event)，直译为服务器发送事件，顾名思义，也就是客户端可以获取到服务器发送的事件

我们常见的 http 交互方式是客户端发起请求，服务端响应，然后一次请求完毕；但是在 sse 的场景下，客户端发起请求，连接一直保持，服务端有数据就可以返回数据给客户端，这个返回可以是多次间隔的方式

2. 特点分析

SSE 最大的特点，可以简单规划为两个

长连接
服务端可以向客户端推送信息

了解 websocket 的小伙伴，可能也知道它也是长连接，可以推送信息，但是它们有一个明显的区别

sse 是单通道，只能服务端向客户端发消息；而 webscoket 是双通道

那么为什么有了 webscoket 还要搞出一个 sse 呢？既然存在，必然有着它的优越之处

sse	websocket
http 协议	独立的 websocket 协议
轻量，使用简单	相对复杂
默认支持断线重连	需要自己实现断线重连
文本传输	二进制传输
支持自定义发送的消息类型	-

3. 应用场景

从 sse 的特点出发，我们可以大致的判断出它的应用场景，需要轮询获取服务端最新数据的 case 下，多半是可以用它的

比如显示当前网站在线的实时人数，法币汇率显示当前实时汇率，电商大促的实时成交额等等…

II. 手动实现 sse 功能

sse 本身是有自己的一套玩法的，后面会进行说明，这一小节，则主要针对 sse 的两个特点长连接 + 后端推送数据，如果让我们自己来实现这样的一个接口，可以怎么做？

1. 项目创建

借助 SpringBoot 2.2.1.RELEASE来创建一个用于演示的工程项目，核心的 xml 依赖如下

xml复制代码<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.1.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <java.version>1.8</java.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

<build>
    <pluginManagement>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </pluginManagement>
</build>
<repositories>
    <repository>
        <id>spring-snapshots</id>
        <name>Spring Snapshots</name>
        <url>https://repo.spring.io/libs-snapshot-local</url>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </repository>
    <repository>
        <id>spring-milestones</id>
        <name>Spring Milestones</name>
        <url>https://repo.spring.io/libs-milestone-local</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
    <repository>
        <id>spring-releases</id>
        <name>Spring Releases</name>
        <url>https://repo.spring.io/libs-release-local</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
</repositories>

2. 功能实现

在 Http1.1 支持了长连接，请求头添加一个Connection: keep-alive即可

在这里我们借助异步请求来实现 sse 功能，至于什么是异步请求，推荐查看博文: 【WEB 系列】异步请求知识点与使用姿势小结

因为后端可以不定时返回数据，所以我们需要注意的就是需要保持连接，不要返回一次数据之后就断开了；其次就是需要设置请求头Content-Type: text/event-stream;charset=UTF-8 （如果不是流的话会怎样？）

java复制代码// 新建一个容器，保存连接，用于输出返回
private Map<String, PrintWriter> responseMap = new ConcurrentHashMap<>();

// 发送数据给客户端
private void writeData(String id, String msg, boolean over) throws IOException {
    PrintWriter writer = responseMap.get(id);
    if (writer == null) {
        return;
    }

    writer.println(msg);
    writer.flush();
    if (over) {
        responseMap.remove(id);
    }
}

// 推送
@ResponseBody
@GetMapping(path = "subscribe")
public WebAsyncTask<Void> subscribe(String id, HttpServletResponse response) {

    Callable<Void> callable = () -> {
        response.setHeader("Content-Type", "text/event-stream;charset=UTF-8");
        responseMap.put(id, response.getWriter());
        writeData(id, "订阅成功", false);
        while (true) {
            Thread.sleep(1000);
            if (!responseMap.containsKey(id)) {
                break;
            }
        }
        return null;
    };

    // 采用WebAsyncTask 返回 这样可以处理超时和错误 同时也可以指定使用的Excutor名称
    WebAsyncTask<Void> webAsyncTask = new WebAsyncTask<>(30000, callable);
    // 注意：onCompletion表示完成，不管你是否超时、是否抛出异常，这个函数都会执行的
    webAsyncTask.onCompletion(() -> System.out.println("程序[正常执行]完成的回调"));

    // 这两个返回的内容，最终都会放进response里面去===========
    webAsyncTask.onTimeout(() -> {
        responseMap.remove(id);
        System.out.println("超时了!!!");
        return null;
    });
    // 备注：这个是Spring5新增的
    webAsyncTask.onError(() -> {
        System.out.println("出现异常!!!");
        return null;
    });


    return webAsyncTask;
}

看一下上面的实现，基本上还是异步请求的那一套逻辑，请仔细看一下callable中的逻辑，有一个 while 循环，来保证长连接不中断

接下来我们新增两个接口，用来模拟后端给客户端发送消息，关闭连接的场景

java复制代码@ResponseBody
@GetMapping(path = "push")
public String pushData(String id, String content) throws IOException {
    writeData(id, content, false);
    return "over!";
}

@ResponseBody
@GetMapping(path = "over")
public String over(String id) throws IOException {
    writeData(id, "over", true);
    return "over!";
}

我们简单的来演示下操作过程

III. SseEmitter

上面只是简单实现了 sse 的长连接 + 后端推送消息，但是与标准的 SSE 还是有区别的，sse 有自己的规范，而我们上面的实现，实际上并没有管这个，导致的问题是前端按照 sse 的玩法来请求数据，可能并不能正常工作

1. sse 规范

在 html5 的定义中，服务端 sse，一般需要遵循以下要求

请求头

开启长连接 + 流方式传递

1
2
3

yaml复制代码Content-Type: text/event-stream;charset=UTF-8
Cache-Control: no-cache
Connection: keep-alive

数据格式

服务端发送的消息，由 message 组成，其格式如下:

1	makefile复制代码field:value\n\n

其中 field 有五种可能

空: 即以:开头，表示注释，可以理解为服务端向客户端发送的心跳，确保连接不中断
data：数据
event: 事件，默认值
id: 数据标识符用 id 字段表示，相当于每一条数据的编号
retry: 重连时间

2. 实现

SpringBoot 利用 SseEmitter 来支持 sse，可以说非常简单了，直接返回SseEmitter对象即可；重写一下上面的逻辑

java复制代码@RestController
@RequestMapping(path = "sse")
public class SseRest {
    private static Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
    @GetMapping(path = "subscribe")
    public SseEmitter push(String id) {
        // 超时时间设置为1小时
        SseEmitter sseEmitter = new SseEmitter(3600_000L);
        sseCache.put(id, sseEmitter);
        sseEmitter.onTimeout(() -> sseCache.remove(id));
        sseEmitter.onCompletion(() -> System.out.println("完成！！！"));
        return sseEmitter;
    }

    @GetMapping(path = "push")
    public String push(String id, String content) throws IOException {
        SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter != null) {
            sseEmitter.send(content);
        }
        return "over";
    }

    @GetMapping(path = "over")
    public String over(String id) {
        SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter != null) {
            sseEmitter.complete();
            sseCache.remove(id);
        }
        return "over";
    }
}

上面的实现，用到了 SseEmitter 的几个方法，解释如下

send(): 发送数据，如果传入的是一个非SseEventBuilder对象，那么传递参数会被封装到 data 中
complete(): 表示执行完毕，会断开连接
onTimeout(): 超时回调触发
onCompletion(): 结束之后的回调触发

同样演示一下访问请求

上图总的效果和前面的效果差不多，而且输出还待上了前缀，接下来我们写一个简单的 html 消费端，用来演示一下完整的 sse 的更多特性

html复制代码<!doctype html>
<html lang="en">
<head>
    <title>Sse测试文档</title>
</head>
<body>
<div>sse测试</div>
<div id="result"></div>
</body>
</html>
<script>
    var source = new EventSource('http://localhost:8080/sse/subscribe?id=yihuihui');
    source.onmessage = function (event) {
        text = document.getElementById('result').innerText;
        text += '\n' + event.data;
        document.getElementById('result').innerText = text;
    };
    <!-- 添加一个开启回调 -->
    source.onopen = function (event) {
        text = document.getElementById('result').innerText;
        text += '\n 开启: ';
        console.log(event);
        document.getElementById('result').innerText = text;
    };
</script>

将上面的 html 文件放在项目的resources/static目录下；然后修改一下前面的SseRest

java复制代码@Controller
@RequestMapping(path = "sse")
public class SseRest {
    @GetMapping(path = "")
    public String index() {
        return "index.html";
    }

    @ResponseBody
    @GetMapping(path = "subscribe", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})
    public SseEmitter push(String id) {
        // 超时时间设置为3s，用于演示客户端自动重连
        SseEmitter sseEmitter = new SseEmitter(1_000L);
        // 设置前端的重试时间为1s
        sseEmitter.send(SseEmitter.event().reconnectTime(1000).data("连接成功"));
        sseCache.put(id, sseEmitter);
        System.out.println("add " + id);
        sseEmitter.onTimeout(() -> {
            System.out.println(id + "超时");
            sseCache.remove(id);
        });
        sseEmitter.onCompletion(() -> System.out.println("完成！！！"));
        return sseEmitter;
    }
}

我们上面超时时间设置的比较短，用来测试下客户端的自动重连，如下，开启的日志不断增加

其次将 SseEmitter 的超时时间设长一点，再试一下数据推送功能

请注意上面的演示，当后端结束了长连接之后，客户端会自动重新再次连接，不用写额外的重试逻辑了，就这么神奇

3. 小结

本篇文章介绍了 SSE 的相关知识点，并对比 websocket 给出了 sse 的优点（至于啥优点请往上翻）

请注意，本文虽然介绍了两种 sse 的方式，第一种借助异步请求来实现，如果需要完成 sse 的规范要求，需要自己做一些适配，如果需要了解 sse 底层实现原理的话，可以参考一下；在实际的业务开发中，推荐使用SseEmitter

IV. 其他

0. 项目

系列博文

源码

工程：github.com/liuyueyi/sp…
项目源码: github.com/liuyueyi/sp…

1. 一灰灰 Blog

尽信书则不如，以上内容，纯属一家之言，因个人能力有限，难免有疏漏和错误之处，如发现 bug 或者有更好的建议，欢迎批评指正，不吝感激

下面一灰灰的个人博客，记录所有学习和工作中的博文，欢迎大家前去逛逛

一灰灰 Blog 个人博客 blog.hhui.top
一灰灰 Blog-Spring 专题博客 spring.hhui.top

一灰灰blog

本文转载自: 掘金

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

从零开始 Docker 搭建 Redis 集群从零开始 D

发表于 2020-04-01

从零开始 Docker 搭建 Redis 集群

Docker 图片

Docker 安装

Docker 官方安装说明地址

初识 Docker

轻量,简单的建模方式.
为云计算而生.
多平台可以移植,易于构建,易于协作.
基础设施即代码:通过 Docker 的镜像,我们可以将创建过程变为自动和且可重复,而且可以做版本管理.
不可变基础设施:对无状态服务升级,部署会更为容易和简单,我们无需再修改配置,只需要销毁重建即可.
配上 SOA 或微服务架构,Docker 会更香.

Docker 的结构

Docker Archetecture

解释一下上图的几个关键词:

Client:这就是你所看到的 docker 控制终端.
DOCKER_HOST:这是 docker 运行的整体运行时上下文
Docker daemon:这是 Docker 的守护程序,由他来和 Client 进行通信,获取 Client 的命令,并对 Image Container 等组件进行统一管理.
Images:这是一切容器运行的模板或者是基础,容器都是通过 image 来创建的.
Containers:由 Image 所创建的运行实例,程序都运行在容器中,一个 Image 可以扩展出 N 个 Container
Docker Registry: Docker Image 的存放库,一般情况下都是使用官方的 DockerHub.com.不过,你也可以搭建自己的 Docker Registry 将 Image 管理私有化.

Docker Container

Docker 安装成功后,请你在命令行中使用docker info来确认 Docker 是否正确的安装在本地环境中.

启动你的第一个 Container

docker run -t -i ubuntu /bin/bash

如果,启动成功的话,你会看到你的命令行变成

root@c8fabbcb9f8a:/#

@符号后面的这串字符可能和我的不太一样,他们都是随机生成的容器 ID

你可以尝试在 Container 终端中使用 ps aux,hostname,ip a等命令,来体会容器的便利性.

你还可以在 Container 中安装你所需要的工具,你可以使用 apt-get update && apt-get install vim来给他安装 vim.

请你记住,只要不是在 Image 中所包含的,当 Container 被 Remove 后,这一切都将不复存在.

接着,我们来一个个解释命令是什么意思:

docker run :让 docker 启动一个容器的命令.
-t:创建一个虚拟的 TTY 终端.
-i:代表我们创建的 Container 是会被捕获 STDIN 的.
ubuntu:这是我们启动 Container 所对应的 Image,如果需要指定版本的话可以通过 ubuntu:18.04 这种方式.
/bin/bash:这是当 Container 启动完成后,就会执行这个命令.

最后,输入exit接上回车来停止这个终端.这时候你会退回到你自己的操作系统终端中,那我们怎么去看刚刚创建的容器去哪里了呢.

输入docker ps -a

你会看到下面这样的内容

1
2
3

复制代码
CONTAINER ID        IMAGE                  COMMAND                  CREATED             STATUS                     PORTS                    NAMES
c8fabbcb9f8a        ubuntu                 "/bin/bash"              7 seconds ago       Exited (0) 4 seconds ago                            elated_banzai

我们可以通过这些发现很多有用的信息:

CONTAINER ID:我们之前 tty 里面看到的容器 ID.
IMAGE: Container 所使用的容器.
COMMAND: Container 启动后所使用的命令.
CREATED: Container 创建的时间.
STATUS: Container 目前的状态.
PORTS: Container 所暴露出内部的端口,不过我们这个实例没有,后面的实例中我们会看到的具体的操作和概念.
NAMES:Container 的名字,我们可以在 container run 的时候通过--name 指定名字,如果没有指定,那么 docker 会自动生成一个名字.

查看 Container

docker ps,docker container ls这 2 个命令有同样的结果,就是查看当前正在运行的 Containers.如果,需要查看所有的容器,需要在命令后面加上-a

重启已经停止的 Container

docker start elated_banzai或者docker start c8fabbcb9f8a

在 docker start 命令后面跟上 container 的 name 或者 id 都可以重新启动它.

但是,我们会发现这次启动以后,我们没有进入终端,但是查看最新的状态,他却是 UP 的状态.这是为什么呢

因为,我们这次只是重启了 Container,但是并没有通过 tty 连接到这个容器上去.

那我们还想连接回去怎么办呢,下面我们会讲到

通过 tty 连接运行中的 Container

docker exec -i -t elated_banzai /bin/bash

当然,上面使用 Container name 的地方也可以互换为 Container id.在 docker 命令中,所有指定 Container 的地方,都是 name 和 id 可以互换的.

这个命令比较简单,与 docker run基本上是类似的,这里就不做展开讲解了.

Container 输出的查看

首先,我们下面的命令,来创建一个叫 log_container 的 Container

docker run -d --name log_container ubuntu /bin/sh -c "while true;do echo hello world; sleep 3; done"

这时,可能你会发现,这条命令里面多了一个 -d的配置,他是用来指定使用 daemon 的模式来运行这个 Container.

我们在最后写了一个死循环的 shell 脚本,让他每 3 秒输出一个 hello world.

回车以后,他只是输出了这个 container 的 id,我们可以通过下面的命令结合 id或者 name来看输出的内容.

docker logs log_container

添加-f来持续查看他的输出,这个和 linux 中的 tail 命令是类似的.

我们还可以添加-t来查看输出的时间戳.

查看 Container 内部的进程

docker top name/id

查看所有 Containers 或者指定的 Container 的统计信息

docker stats,docker stats ContainerName1 ContainerName2...

停止 Container

docker stop name/id

删除 Container

docker rm name/id

这个命令如果不添加-f,那我们只能删除已经停止的容器.

深入查看 Container 信息

docker inspect name/id

这个命令显示的内容比较多,不过大部分都能通过keyname看懂,如果有不清楚的部分请自行查阅相关文档.

Image

查看 Images

docker images,docker image ls

删除 Image

docker images

Dockerfile

Dockerfile 是一个使用基本的 Docker DSL 语法的指令来构建 Docker Image 的文件.
通过,Dockerfile来构建 Image ,更具备重复性,透明性和幂等性.

基础理论基本上介绍的差不多了,那么我们直接上干货.

用 Docker 搭建一个 redis cluster

创建 replica Base Image

首先,我们创建 Redis 的主镜像

1
2
3

复制代码$ mkdir redis-replica
$ cd redis-replica
$ touch Dockerfile

Dockerfile:

复制代码#使用的母镜像
FROM redis
#维护者信息,这里填的是我自己的邮箱
MAINTAINER crowhyc@163.com
#环境变量,添加以后可以在容器和接下来的命令里面都可以使用
#这个环境变量是用于定义这个镜像的版本和日期
ENV REFESHED_AT 2020-04-01
#用来向基于Image创建的 Container 添加卷
#一个卷是可以存在于一个或多个 Container 内特定的目录,这个目录可以共享数据或对数据进行持久化功能
#下面这个 VOLUME 会在 Container 里面创建这 2 个目录
VOLUME ["/var/lib/redis","/var/log/redis/"]
#基于此 Image 创建的 Container 会对外暴露 6379 接口
#可以通过 docker ports name/id 来查询 Container 暴露接口与本地接口所映射的关系
#也可以在 Container docker run 的时候通过-p 6379:6379 把他绑定到指定的 port 上
EXPOSE 6379
#用于指定一个 Container 启动时要运行的命令,类似于后面会遇到的 RUN 命令
#只是 RUN 指令是指定 Image 被构建时需要运行的命令
#docker run 命令可以覆盖 CMD 命令,这里留空的目的是让 docker run 的时候可以自定义需要执行的命令
#后面介绍到的 ENTRYPOINT 命令则不可以被覆盖
CMD []

保存文件后,我们运行下面的这条命令

docker build -t crowhyc/redis-replica:1.0 .

我们来解释一下这条命令

docker build是用来创建镜像的.

-t和crowhyc/redis-replica:1.0是用来表明后面我们会用这种格式的方式来指定 Image 的名称和版本.

最后,还有一个不起眼的.这个是用来指定 Dockerfile 的地址..是说明 Dockerfile 就在当前目录下

我们不会 run 这个 Image.它是接下来我们 redis-primary 和 redis-slave 的母镜像

创建 redis primary Image

1
2
3

复制代码$ mkdir redis-primary
$ cd redis-primary
$ touch Dockerfile

复制代码FROM crowhyc/redis-primary:1.0
MAINTAINER crowhyc "crowhyc@163.com"
ENV REFRESHED_AT 2020-04-01
#是 Container 启动以后执行的命令,与 CMD 不同,Container 启动的命令不能覆盖它
ENTRYPOINT ["redis-server","--logfile /var/log/redis/redis-server.log"]

紧接着,我们创建 redis 的主镜像

docker build -t crowhyc/redis-primary:1.0 .

创建 redis slave Image

1
2
3

复制代码$ mkdir redis-slave
$ cd redis-slave
$ touch Dockerfile

复制代码FROM crowhyc/redis-replica:1.0
MAINTAINER crowhyc "crowhyc@163.com"
ENV REFRESHED_AT 2020-04-01
#这里有一点需要注意的是,我们第三个参数使用的是 redis-primary 当我们启动 primary Container 的时候需要与这个名字相同
ENTRYPOINT ["redis-server","--logfile /var/log/redis/redis-server.log","--slaveof redis-primary 6379"]

启动 redis cluster 并查看状态

创建 Network

docker network create redis-cluster

首先,我们使用上面的命令创建一个新的 network.

docker network是 docker Container 之间用来通信的网络,用户可以自己创建网络,用来组件自己的集群网络.

docker network ls

同样,我们也可以用上面的命令来查看所有的网络

启动 redis-primary

接着,我们使用下面的docker run来启动我们的 primary redis server.

docker run -d -h redis-primary --net redis-cluster --name reids-primary crowhyc/redis-primary:1.0

-h是一个之前没有出现过的新标志,他用来设置 Container 的 hostname,这会覆盖默认的行为即将 Container 主机名设置为 ContainerId.

使用这个标志可以保证redis-primary被作为 Container 的 hostname,并被本地的 DNS 服务正确解析.

--net是用来设定这个 Container 所使用的 Network

查看 redis-primary 日志

我们输入 docker logs 命令去查看日志

docker logs -f redis-primary

稍作等待,你会发现–根本什么都没有…那这是为什么呢?

这是因为我们在做 redis-primary 的时候,让日志记录到了/var/log/redis/redis-server.log 这个文件里面.

接着,我们用更加巧妙的方式去观察类似这种情况下的日志输出.

docker run -it --rm --volumes-from redis-primary ubuntu cat /var/log/redis/redis-server.log

这下我们就能看到对应的 redis-primary 的日志了,如果出现了Ready to accept connections说明 redis-primary已经启动成功.

接着我们来为他启动 2 个 redis-slave 吧

启动多个 redis-slave

docker run -d -h redis-slave01 --net redis-cluster crowhyc/redis-slave:1.0
docker run -d -h redis-slave02 --net redis-cluster crowhyc/redis-slave:1.0

接着我们用上面刚刚学的 docker run 来观察 slave 是否和 primary 建立了联系
docker run -it --rm --volumes-from redis-slave01 ubuntu cat /var/log/redis/redis-server.log
docker run -it --rm --volumes-from redis-slave02 ubuntu cat /var/log/redis/redis-server.log

如果,我们能看到类似于MASTER <-> REPLICA sync: Finished with success这样的日志,说明我们 slave 与 primary 已经连接成功了
接下来,我们用 docker 的方式来测试 redis 吧

简要 redis 集群测试

docker run -it --rm --net redis-cluster redis /bin/bash
紧接着我们连接到 redis-primary

redis-cli -h redis-primary

然后,我们 SET 一个值

redis-primary:6379> set "test" 1234

接着,我们再进入 redis-slave01

docker exec -it redis-slave01 /bin/bash

redis-cli -h localhost

get "test"

如果,我们在执行完get命令后看到了”1234”说明我们的 redis 集群已经完成搭建了.

本文转载自: 掘金

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

工具介绍 ASAN和HWASAN原理解析

发表于 2020-03-31

由于虚拟机的存在，Android应用开发者们通常不用考虑内存访问相关的错误。而一旦我们深入到Native世界中，原本面容和善的内存便开始凶恶起来。这时，由于程序员写法不规范、逻辑疏漏而导致的内存错误会统统跳到我们面前，对我们嘲讽一番。

这些错误既影响了程序的稳定性，也影响了程序的安全性，因为好多恶意代码就通过内存错误来完成入侵。不过麻烦的是，Native世界中的内存错误很难排查，因为很多时候导致问题的地方和发生问题的地方相隔甚远。为了更好地解决这些问题，各路大神纷纷祭出自己手中的神器，相互PK，相互补充。

ASAN(Address Sanitizer)和HWASAN(Hardware-assisted Address Sanitizer)就是这些工具中的佼佼者。

在ASAN出来之前，市面上的内存调试工具要么慢，要么只能检测部分内存错误，要么这两个缺点都有。总之，不够优秀。

HWASAN则是ASAN的升级版，它利用了64位机器上忽略高位地址的特性，将这些被忽略的高位地址重新利用起来，从而大大降低了工具对于CPU和内存带来的额外负载。

ASAN

ASAN工具包含两大块：

插桩模块(Instrumentation module)
一个运行时库(Runtime library)

插桩模块主要会做两件事：

对所有的memory access都去检查该内存所对应的shadow memory的状态。这是静态插桩，因此需要重新编译。
为所有栈上对象和全局对象创建前后的保护区(Poisoned redzone)，为检测溢出做准备。

运行时库也同样会做两件事：

替换默认路径的malloc/free等函数。为所有堆对象创建前后的保护区，将free掉的堆区域隔离(quarantine)一段时间，避免它立即被分配给其他人使用。
对错误情况进行输出，包括堆栈信息。

1.1 Shadow Memory

如果想要了解ASAN的实现原理，那么shadow memory将是第一个需要了解的概念。

Shadow memory有一些元数据的思维在里面。它虽然也是内存中的一块区域，但是其中的数据仅仅反应其他正常内存的状态信息。所以可以理解为正常内存的元数据，而正常内存中存储的才是程序真正需要的数据。

Malloc函数返回的地址通常是8字节对齐的，因此任意一个由（对齐的）8字节所组成的内存区域必然落在以下9种状态之中：最前面的k（0≤k≤8）字节是可寻址的，而剩下的8-k字节是不可寻址的。这9种状态便可以用shadow memory中的一个字节来进行编码。

实际上，一个byte可以编码的状态总共有256（2^8）种，因此用在这里绰绰有余。

Shadow memory和normal memory的映射关系如上图所示。一个byte的shadow memory反映8个byte normal memory的状态。那如何根据normal memory的地址找到它对应的shadow memory呢？

对于64位机器上的Android而言，二者的转换公式如下：

Shadow memory address = (Normal memory address >> 3) + 0x1000000000 (9个0)

右移三位的目的是为了完成 8➡1的映射，而加一个offset是为了和Normal memory区分开来。最终内存空间种会存在如下的映射关系：

Bad代表的是shadow memory的shadow memory，因此其中数据没有意义，该内存区域不可使用。

上文中提到，8字节组成的memory region共有9中状态：

1~~7个字节可寻址（共七种），shadow memory的值为1~~7。
8个字节都可寻址，shadow memory的值为0。
0个字节可寻址，shadow memory的值为负数。

为什么0个字节可寻址的情况shadow memory不为0，而是负数呢？是因为0个字节可寻址其实可以继续分为多种情况，譬如：

这块区域是heap redzones
这块区域是stack redzones
这块区域是global redzones
这块区域是freed memory

对所有0个字节可寻址的normal memory region的访问都是非法的，ASAN将会报错。而根据其shadow memory的值便可以具体判断是哪一种错。

yaml复制代码Shadow byte legend (one shadow byte represents 8 application bytes):
  Addressable:           00
  Partially addressable: 01 02 03 04 05 06 07 
  Heap left redzone:     fa (实际上Heap right redzone也是fa)
  Freed Heap region:     fd
  Stack left redzone:    f1
  Stack mid redzone:     f2
  Stack right redzone:   f3
  Stack after return:    f5
  Stack use after scope: f8
  Global redzone:        f9
  Global init order:     f6
  Poisoned by user:      f7
  Container overflow:    fc
  Array cookie:          ac
  Intra object redzone:  bb
  ASan internal:         fe
  Left alloca redzone:   ca
  Right alloca redzone:  cb
  Shadow gap:            cc

1.2 检测算法

ini复制代码ShadowAddr = (Addr >> 3) + Offset;
k = *ShadowAddr;
if (k != 0 && ((Addr & 7) + AccessSize > k))
	ReportAndCrash(Addr);

在每次内存访问时，都会执行如上的伪代码，以判断此次内存访问是否合规。

首先根据normal memory的地址找到对应shadow memory的地址，然后取出其中存取的byte值：k。

k!=0，说明Normal memory region中的8个字节并不是都可以被寻址的。
Addr & 7，将得知此次内存访问是从memory region的第几个byte开始的。
AccessSize是此次内存访问需要访问的字节长度。
(Addr&7)+AccessSize > k，则说明此次内存访问将会访问到不可寻址的字节。（具体可分为k大于0和小于0两种情况来分析）

当此次内存访问可能会访问到不可寻址的字节时，ASAN会报错并结合shadow memory中具体的值明确错误类型。

1.3 典型错误

1.3.1 Use-After-Free

想要检测UseAfterFree的错误，需要有两点保证：

已经free掉的内存区域需要被标记成特殊的状态。在ASAN的实现里，free掉的normal memory对应的shadow memory值为0xfd（猜测有freed的意思）。
已经free掉的内存区域需要放入隔离区一段时间，防止发生错误时该区域已经通过malloc重新分配给其他人使用。一旦分配给其他人使用，则可能漏掉UseAfterFree的错误。

测试代码：

c++复制代码// RUN: clang -O -g -fsanitize=address %t && ./a.out
int main(int argc, char **argv) {
  int *array = new int[100];
  delete [] array;
  return array[argc];  // BOOM
}

ASAN输出的错误信息：

yaml复制代码=================================================================
==6254== ERROR: AddressSanitizer: heap-use-after-free on address 0x603e0001fc64 at pc 0x417f6a bp 0x7fff626b3250 sp 0x7fff626b3248
READ of size 4 at 0x603e0001fc64 thread T0
    #0 0x417f69 in main example_UseAfterFree.cc:5
    #1 0x7fae62b5076c (/lib/x86_64-linux-gnu/libc.so.6+0x2176c)
    #2 0x417e54 (a.out+0x417e54)
0x603e0001fc64 is located 4 bytes inside of 400-byte region [0x603e0001fc60,0x603e0001fdf0)
freed by thread T0 here:
    #0 0x40d4d2 in operator delete[](void*) /home/kcc/llvm/projects/compiler-rt/lib/asan/asan_new_delete.cc:61
    #1 0x417f2e in main example_UseAfterFree.cc:4
previously allocated by thread T0 here:
    #0 0x40d312 in operator new[](unsigned long) /home/kcc/llvm/projects/compiler-rt/lib/asan/asan_new_delete.cc:46
    #1 0x417f1e in main example_UseAfterFree.cc:3
Shadow bytes around the buggy address:
  0x1c07c0003f30: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x1c07c0003f40: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x1c07c0003f50: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x1c07c0003f60: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x1c07c0003f70: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
=>0x1c07c0003f80: fa fa fa fa fa fa fa fa fa fa fa fa[fd]fd fd fd
  0x1c07c0003f90: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
  0x1c07c0003fa0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
  0x1c07c0003fb0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fa fa
  0x1c07c0003fc0: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x1c07c0003fd0: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa

可以看到，=>指向的那行有一个byte数值用中括号给圈出来了：[fd]。它表示的是此次出错的内存地址对应的shadow memory的值。而其之前的fa表示Heap left redzone，它是之前该区域有效时的遗留产物。连续的fd总共有50个，每一个shadow memory的byte和8个normal memory byte对应，所以可以知道此次free的内存总共是50×8=400bytes。这一点在上面的log中也得到了验证，截取出来展示如下：

1	csharp复制代码0x603e0001fc64 is located 4 bytes inside of 400-byte region [0x603e0001fc60,0x603e0001fdf0)

此外，ASAN的log中不仅有出错时的堆栈信息，还有该内存区域之前free时的堆栈信息。因此我们可以清楚地知道该区域是如何被释放的，从而快速定位问题，解决问题。

1.3.2 Heap-Buffer-Overflow

想要检测HeapBufferOverflow的问题，只需要保证一点：

正常的Heap前后需要插入一定长度的安全区，而且此安全区对应的shadow memory需要被标记为特殊的状态。在ASAN的实现里，安全区被标记为0xfa。

测试代码：

ASAN输出的错误信息：

yaml复制代码=================================================================
==1405==ERROR: AddressSanitizer: heap-buffer-overflow on address 0x0060bef84165 at pc 0x0058714bfb24 bp 0x007fdff09590 sp 0x007fdff09588
WRITE of size 1 at 0x0060bef84165 thread T0
    #0 0x58714bfb20  (/system/bin/bootanimation+0x8b20)
    #1 0x7b434cd994  (/apex/com.android.runtime/lib64/bionic/libc.so+0x7e994)

0x0060bef84165 is located 1 bytes to the right of 100-byte region [0x0060bef84100,0x0060bef84164)
allocated by thread T0 here:
    #0 0x7b4250a1a4  (/system/lib64/libclang_rt.asan-aarch64-android.so+0xc31a4)
    #1 0x58714bfac8  (/system/bin/bootanimation+0x8ac8)
    #2 0x7b434cd994  (/apex/com.android.runtime/lib64/bionic/libc.so+0x7e994)
    #3 0x58714bb04c  (/system/bin/bootanimation+0x404c)
    #4 0x7b45361b04  (/system/bin/bootanimation+0x54b04)

SUMMARY: AddressSanitizer: heap-buffer-overflow (/system/bin/bootanimation+0x8b20) 
Shadow bytes around the buggy address:
  0x001c17df07d0: fa fa fa fa fa fa fa fa fd fd fd fd fd fd fd fd
  0x001c17df07e0: fd fd fd fd fd fa fa fa fa fa fa fa fa fa fa fa
  0x001c17df07f0: fd fd fd fd fd fd fd fd fd fd fd fd fd fa fa fa
  0x001c17df0800: fa fa fa fa fa fa fa fa fd fd fd fd fd fd fd fd
  0x001c17df0810: fd fd fd fd fd fa fa fa fa fa fa fa fa fa fa fa
=>0x001c17df0820: 00 00 00 00 00 00 00 00 00 00 00 00[04]fa fa fa
  0x001c17df0830: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x001c17df0840: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x001c17df0850: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x001c17df0860: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa
  0x001c17df0870: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa

可以看到最终出错的shadow memory值为0x4，表示该shadow memroy对应的normal memory中只有前4个bytes是可寻址的。0x4的shadow memory前还有12个0x0，表示其前面的12个memory region（每个region有8个byte）都是完全可寻址的。因此所有可寻址的大小=12×8+4=100，正是代码中malloc的size。之所以此次访问会出错，是因为地址0x60bef84165意图访问最后一个region的第五个byte，而该region只有前四个byte可寻址。由于0x4后面是0xfa，因此此次错误属于HeapBufferOverflow。

1.4 缺陷

自从2011年诞生以来，ASAN已经成功地参与了众多大型项目，譬如Chrome和Android。虽然它的表现很突出，但仍然有些地方不尽如人意，重点表现在以下几点：

ASAN的运行是需要消耗memory和CPU资源的，此外它也会增加代码大小。它的性能相比于之前的工具确实有了质的提升，但仍然无法适用于某些压力测试场景，尤其是需要全局打开的时候。这一点在Android上尤为明显，每当我们想要全局打开ASAN调试某些奇葩问题时，系统总会因为负载过重而跑不起来。
ASAN对于UseAfterFree的检测依赖于隔离区，而隔离时间是非永久的。也就意味着已经free的区域过一段时间后又会重新被分配给其他人。当它被重新分配给其他人后，原先的持有者再次访问此块区域将不会报错。因为这一块区域的shadow memory不再是0xfd。所以这算是ASAN漏检的一种情况。
ASAN对于overflow的检测依赖于安全区，而安全区总归是有大小的。它可能是64bytes，128bytes或者其他什么值，但不管怎么样终归是有限的。如果某次踩踏跨过了安全区，踩踏到另一片可寻址的内存区域，ASAN同样不会报错。这是ASAN的另一种漏检。

2.HWASAN

HWASAN是ASAN工具的“升级版”，它基本上解决了上面所说的ASAN的3个问题。但是它需要64位硬件的支持，也就是说在32位的机器上该工具无法运行。

AArch64是64位的架构，指的是寄存器的宽度是64位，但并不表示内存的寻址范围是2^64。真实的寻址范围和处理器内部的总线宽度有关，实际上ARMv8寻址只用到了低48位。也就是说，一个64bit的指针值，其中真正用于寻址的只有低48位。那么剩下的高16位干什么用呢？答案是随意发挥。AArch64拥有地址标记(Address tagging, or top-byte-ignore)的特性，它表示允许软件使用64bit指针值的高8位开发特定功能。

HWASAN用这8bit来存储一块内存区域的标签(tag)。接下来我们以堆内存示例，展示这8bit到底如何起作用。

堆内存通过malloc分配出来，HWASAN在它返回地址时会更改该有效地址的高8位。更改的值是一个随机生成的单字节值，譬如0xaf。此外，该分配出来的内存对应的shadow memory值也设为0xaf。需要注意的是，HWASAN中normal memory和shadow memory的映射关系是16➡1，而ASAN中二者的映射关系是8➡1。

以下分别讨论UseAfterFree和HeapOverFlow的情况。

2.1 Use-After-Free

当一个堆内存被分配出来时，返回给用户空间的地址便已经带上了标签（存储于地址的高8位）。之后通过该地址进行内存访问，将先检测地址中的标签值和访问地址对应的shadow memory的值是否相等。如果相等则验证通过，可以进行正常的内存访问。

当该内存被free时，HWASAN会为该块区域分配一个新的随机值，存储于其对应的shadow memory中。如果此后再有新的访问，则地址中的标签值必然不等于shadow memory中存储的新的随机值，因此会有错误产生。通过如下图示可以很好地明白这一点（图中只用了4bit记录标记值，但不影响理解，8bit标记值的检测和它一致）。

2.2 Heap-Over-Flow

想要检测HeapOverFlow，有一个前提需要满足：相邻的memory区域需要有不同的shadow memory值，否则将无法分辨两个不同的memory区域。为每个memory区域随机分配将有概率让两个相邻区域具有同样的shadow memory值，虽然概率比较小，但总归是个缺陷。因此工具中会有其他逻辑保证这个前提。

下图展示了HeapOverFlow的检测过程。指针p的标签和访问的地址p[32]所对应的shadow memory值不一致，因此报错（图中只用了4bit记录标记值，但不影响理解，8bit标记值的检测和它一致）。

2.3 错误信息示例

ini复制代码Abort message: '==12528==ERROR: HWAddressSanitizer: tag-mismatch on address 0x003d557e2c20 at pc 0x00748b4a6918
READ of size 4 at 0x003d557e2c20 tags: d1/9b (ptr/mem) in thread T0
    #0 0x748b4a6914  (/system/lib64/libutils.so+0x11914)
    #1 0x748a521bdc  (/apex/com.android.runtime/lib64/bionic/libc.so+0x121bdc)
    #2 0x748a51ad7c  (/apex/com.android.runtime/lib64/bionic/libc.so+0x11ad7c)
    #3 0x748a47f830  (/apex/com.android.runtime/lib64/bionic/libc.so+0x7f830)

[0x003d557e2c20,0x003d557e2c80) is a small unallocated heap chunk; size: 96 offset: 0
Thread: T0 0x006b00002000 stack: [0x007fcd371000,0x007fcdb71000) sz: 8388608 tls: [0x000000000000,0x000000000000)
HWAddressSanitizer can not describe address in more detail.
Memory tags around the buggy address (one tag corresponds to 16 bytes):
   e1  e1  e1  e1  83  83  83  83  83  00  a3  a3  a3  a3  a3  a3   
   b7  b7  b7  b7  b7  00  01  01  01  01  01  00  95  95  95  95   
   95  00  ec  ec  ec  ec  ec  00  c8  c8  c8  c8  c8  00  21  21   
   21  21  21  00  cb  cb  cb  cb  cb  00  b8  b8  b8  b8  b8  00   
   14  14  14  14  14  14  b9  b9  b9  b9  b9  b9  89  89  89  89   
   89  89  95  95  95  95  95  95  47  47  47  47  47  00  fe  fe   
   fe  fe  fe  00  c5  c5  c5  c5  c5  00  8e  8e  8e  8e  8e  8e   
   5c  5c  5c  5c  5c  5c  af  af  af  af  af  af  b0  b0  b0  b0   
=> b0  b0 [9b] 9b  9b  9b  9b  9b  1f  1f  1f  1f  1f  1f  69  69 <=
   69  69  69  a0  7a  7a  7a  7a  7a  ff  eb  eb  eb  eb  eb  eb   
   16  16  16  16  16  16  81  81  81  81  81  81  7f  7f  7f  7f   
   7f  7f  57  57  57  57  57  57  e0  e0  e0  e0  e0  e0  94  94   
   94  94  94  00  35  35  35  35  35  35  98  98  98  98  98  00   
   7d  7d  7d  7d  7d  7d  6e  6e  6e  6e  6e  6e  59  59  59  59   
   59  59  8e  8e  8e  8e  8e  8e  6d  6d  6d  6d  6d  6d  69  69   
   69  69  69  69  d5  d5  d5  d5  d5  d5  63  63  63  63  63  63

0x9b总共有6个，因此该memory区域的总长为6×16=96，与上述提示一致。

1	arduino复制代码[0x003d557e2c20,0x003d557e2c80) is a small unallocated heap chunk; size: 96

2.4 优缺点

和ASAN相比，HWASAN具有如下缺点：

可移植性较差，只适用于64位机器。
需要对Linux Kernel做一些改动以支持工具。
对于所有错误的检测将有一定概率false negative（漏掉一些真实的错误），概率为1/256。原因是tag的生成只能从256（2^8）个数中选一个，因此不同地址的tag将有可能相同。

不过相对于这些缺点，HWASAN所拥有的优点更加引人注目：

不再需要安全区来检测buffer overflow，既极大地降低了工具对于内存的消耗，也不会出现ASAN中某些overflow检测不到的情况。
不再需要隔离区来检测UseAfterFree，因此不会出现ASAN中某些UseAfterFree检测不到的情况。

2.5 一个难题

上述的讨论其实回避了一个问题：如果一个16字节的memory region中只有前几个字节可寻址（假设是5），那么其对应的shadow memory值也是5。这时，如果用地址去访问该region的第2个字节，那么如何判断访问是否合规呢？

此时直接对比地址的tag和shadow memory的值肯定是不行的，因为此时的shadow memory值含义发生了变化，它不再是一个类似于tag的随机值，而是memory region中可访问字节的数目。

为了解决这个难题，HWASAN在这种情况下将memory region的随机值保存在最后一个字节中。所以即便地址的tag和shadow memory的值不等，但只要和memory region中最后一个字节相等，也表明该访问合法。

具体可参考链接：clang.llvm.org/docs/Hardwa…

参考文章：

本文转载自: 掘金

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

程序员必知的 89 个操作系统核心概念

发表于 2020-03-31

操作系统（Operating System，OS）：是管理计算机硬件与软件资源的系统软件，同时也是计算机系统的内核与基石。操作系统需要处理管理与配置内存、决定系统资源供需的优先次序、控制输入与输出设备、操作网络与管理文件系统等基本事务。操作系统也提供一个让用户与系统交互的操作界面。

shell：它是一个程序，可从键盘获取命令并将其提供给操作系统以执行。在过去，它是类似 Unix 的系统上唯一可用的用户界面。如今，除了命令行界面（CLI）外，我们还具有图形用户界面（GUI）。

GUI (Graphical User Interface)：是一种用户界面，允许用户通过图形图标和音频指示符与电子设备进行交互。

内核模式(kernel mode): 通常也被称为 超级模式（supervisor mode），在内核模式下，正在执行的代码具有对底层硬件的完整且不受限制的访问。它可以执行任何 CPU 指令并引用任何内存地址。内核模式通常保留给操作系统的最低级别，最受信任的功能。内核模式下的崩溃是灾难性的；他们将停止整个计算机。超级用户模式是计算机开机时选择的自动模式。
用户模式(user node)：当操作系统运行用户应用程序（例如处理文本编辑器）时，系统处于用户模式。当应用程序请求操作系统的帮助或发生中断或系统调用时，就会发生从用户模式到内核模式的转换。在用户模式下，模式位设置为1。从用户模式切换到内核模式时，它从1更改为0。
计算机架构(computer architecture) ：在计算机工程中，计算机体系结构是描述计算机系统功能，组织和实现的一组规则和方法。它主要包括指令集、内存管理、I/O 和总线结构

SATA(Serial ATA)：串行 ATA (Serial Advanced Technology Attachment)，它是一种电脑总线，负责主板和大容量存储设备（如硬盘及光盘驱动器）之间的数据传输，主要用于个人电脑。
复用(multiplexing)：也称为共享，在操作系统中主要指示了时间和空间的管理。对资源进行复用时，不同的程序或用户轮流使用它。他们中的第一个开始使用资源，然后再使用另一个，依此类推。
大型机(mainframes)：大型机是一类计算机，通常以其大尺寸，存储量，处理能力和高度的可靠性而著称。它们主要由大型组织用于需要大量数据处理的关键任务应用程序。

批处理(batch system): 批处理操作系统的用户不直接与计算机进行交互。每个用户都在打孔卡等脱机设备上准备工作，并将其提交给计算机操作员。为了加快处理速度，将具有类似需求的作业一起批处理并成组运行。程序员将程序留给操作员，然后操作员将具有类似要求的程序分批处理。
OS/360： OS/360，正式称为IBM System / 360操作系统，是由 IBM 为 1964 年发布的其当时新的System/360 大型机开发的已停产的批处理操作系统。
多处理系统(Computer multitasking)：是指计算机同时运行多个程序的能力。多任务的一般方法是运行第一个程序的一段代码，保存工作环境；再运行第二个程序的一段代码，保存环境；……恢复第一个程序的工作环境，执行第一个程序的下一段代码。
分时系统(Time-sharing)：在计算中，分时是通过多程序和多任务同时在许多用户之间共享计算资源的一种系统
相容分时系统(Compatible Time-Sharing System)：最早的分时操作系统，由美国麻省理工学院计算机中心设计与实作。
云计算(cloud computing)：云计算是计算机系统资源（尤其是数据存储和计算能力）的按需可用性，而无需用户直接进行主动管理。这个术语通常用于描述 Internet 上可供许多用户使用的数据中心。如今占主导地位的大型云通常具有从中央服务器分布在多个位置的功能。如果与用户的连接相对较近，则可以将其指定为边缘服务器。

UNIX 操作系统：UNIX 操作系统，是一个强大的多用户、多任务操作系统，支持多种处理器架构，按照操作系统的分类，属于分时操作系统。
UNIX System V：是 UNIX 操作系统的一个分支。
BSD(Berkeley Software Distribution)：UNIX 的衍生系统。
POSIX：可移植操作系统接口，是 IEEE 为要在各种 UNIX 操作系统上运行软件，而定义API的一系列互相关联的标准的总称。
MINIX：Minix，是一个迷你版本的类 UNIX 操作系统。
Linux：终于到了大名鼎鼎的 Linux 操作系统了，太强大了，不予以解释了，大家都懂。

DOS (Disk Operating System)：磁盘操作系统（缩写为DOS）是可以使用磁盘存储设备（例如软盘，硬盘驱动器或光盘）的计算机操作系统。
MS-DOS(MicroSoft Disk Operating System) ：一个由美国微软公司发展的操作系统，运行在Intel x86个人电脑上。它是DOS操作系统家族中最著名的一个，在Windows 95以前，DOS是IBM PC及兼容机中的最基本配备，而MS-DOS则是个人电脑中最普遍使用的DOS操作系统。

MacOS X，怎能少的了苹果操作系统？macOS 是苹果公司推出的基于图形用户界面操作系统，为 Macintosh 的主操作系统

Windows NT(Windows New Technology)：是美国微软公司 1993 年推出的纯 32 位操作系统核心。
Service Pack(SP)：是程序的更新、修复和（或）增强的集合，以一个独立的安装包的形式发布。许多公司，如微软或Autodesk，通常在为某一程序而做的修补程序达到一定数量时，就发布一个Service Pack。
数字版权管理（DRM）：他是工具或技术保护措施（TPM）是一组访问控制技术，用于限制对专有硬件和受版权保护的作品的使用。
x86：x86是一整套指令集体系结构，由 Intel 最初基于 Intel 8086 微处理器及其 8088 变体开发。采用内存分段作为解决方案，用于处理比普通 16 位地址可以覆盖的更多内存。32 位是 x86 默认的位数，除此之外，还有一个 x86-64 位，是x86架构的 64 位拓展，向后兼容于 16 位及 32 位的 x86架构。
FreeBSD：FreeBSD 是一个类 UNIX 的操作系统，也是 FreeBSD 项目的发展成果。
X Window System：X 窗口系统（X11，或简称X）是用于位图显示的窗口系统，在类 UNIX 操作系统上很常见。

Gnome：GNOME 是一个完全由自由软件组成的桌面环境。它的目标操作系统是Linux，但是大部分的 BSD 系统亦支持 GNOME。

网络操作系统(network operating systems)：网络操作系统是用于网络设备（如路由器，交换机或防火墙）的专用操作系统。

分布式网络系统(distributed operating systems)：分布式操作系统是在独立，网络，通信和物理上独立计算节点的集合上的软件。它们处理由多个CPU服务的作业。每个单独的节点都拥有全局集合操作系统的特定软件的一部分。

程序计数器(Program counter)：程序计数器是一个 CPU 中的寄存器，用于指示计算机在其程序序列中的位置。
堆栈寄存器(stack pointer)：堆栈寄存器是计算机 CPU 中的寄存器，其目的是跟踪调用堆栈。
程序状态字(Program Status Word): 它是由操作系统维护的8个字节（或64位）长的数据的集合。它跟踪系统的当前状态。
流水线(Pipeline): 在计算世界中，管道是一组串联连接的数据处理元素，其中一个元素的输出是下一个元素的输入。流水线的元素通常以并行或按时间分割的方式执行。通常在元素之间插入一定数量的缓冲区存储。

超标量(superscalar)：超标量 CPU 架构是指在一颗处理器内核中实行了指令级并发的一类并发运算。这种技术能够在相同的CPU主频下实现更高的 CPU 流量。
系统调用(system call): 指运行在用户空间的程序向操作系统内核请求需要更高权限运行的服务。系统调用提供用户程序与操作系统之间的接口。大多数系统交互式操作需求在内核态运行。如设备 IO 操作或者进程间通信。
多线程(multithreading)：是指从软件或者硬件上实现多个线程并发执行的技术。具有多线程能力的计算机因为有硬件支持而能够在同一时间执行多个线程，进而提升整体处理性能。
CPU 核心(core)：它是 CPU 的大脑，它接收指令，并执行计算或运算以满足这些指令。一个 CPU 可以有多个内核。
图形处理器(Graphics Processing Unit)：又称显示核心、视觉处理器、显示芯片或绘图芯片；它是一种专门在个人电脑、工作站、游戏机和一些移动设备（如平板电脑、智能手机等）上运行绘图运算工作的微处理器。

存储体系结构：顶层的存储器速度最高，但是容量最小，成本非常高，层级结构越向下，其访问效率越慢，容量越大，但是造价也就越便宜。

高速缓存行(cache lines)：其实就是把高速缓存分割成了固定大小的块，其大小是以突发读或者突发写周期的大小为基础的。
缓存命中(cache hit)：当应用程序或软件请求数据时，会首先发生缓存命中。首先，中央处理单元（CPU）在其最近的内存位置（通常是主缓存）中查找数据。如果在缓存中找到请求的数据，则将其视为缓存命中。

L1 cache：一级缓存是 CPU 芯片中内置的存储库。 L1缓存也称为主缓存，是计算机中最快的内存，并且最接近处理器。
L2 cache: 二级缓存存储库，内置在 CPU 芯片中，包装在同一模块中，或者建在主板上。 L2 高速缓存提供给 L1 高速缓存，后者提供给处理器。 L2 内存比 L1 内存慢。
L3 cache: 三级缓存内置在主板上或CPU模块内的存储库。 L3 高速缓存为 L2 高速缓存提供数据，其内存通常比 L2 内存慢，但比主内存快。 L3 高速缓存提供给 L2 高速缓存，后者又提供给 L1 高速缓存，后者又提供给处理器。
RAM((Random Access Memory)：随机存取存储器，也叫主存，是与 CPU 直接交换数据的内部存储器。它可以随时读写，而且速度很快，通常作为操作系统或其他正在运行中的程序的临时数据存储介质。RAM工作时可以随时从任何一个指定的地址写入（存入）或读出（取出）信息。它与 ROM 的最大区别是数据的易失性，即一旦断电所存储的数据将随之丢失。RAM 在计算机和数字系统中用来暂时存储程序、数据和中间结果。
ROM (Read Only Memory)：只读存储器是一种半导体存储器，其特性是一旦存储数据就无法改变或删除，且内容不会因为电源关闭而消失。在电子或电脑系统中，通常用以存储不需经常变更的程序或数据。
EEPROM (Electrically Erasable PROM)：电可擦除可编程只读存储器，是一种可以通过电子方式多次复写的半导体存储设备。
闪存(flash memory)：是一种电子式可清除程序化只读存储器的形式，允许在操作中被多次擦或写的存储器。这种科技主要用于一般性数据存储，以及在电脑与其他数字产品间交换传输数据，如储存卡与U盘。
SSD(Solid State Disks)：固态硬盘，是一种主要以闪存作为永久性存储器的电脑存储设备。

虚拟地址(virtual memory)：虚拟内存是计算机系统内存管理的一种机制。它使得应用程序认为它拥有连续可用的内存（一个连续完整的地址空间），而实际上，它通常是被分隔成多个物理内存碎片，还有部分暂时存储在外部磁盘存储器上，在需要时进行数据交换。与没有使用虚拟内存技术的系统相比，使用这种技术的系统使得大型程序的编写变得更容易，对真正的物理内存（例如RAM）的使用也更有效率。
MMU (Memory Management Unit)：内存管理单元，有时称作分页内存管理单元。它是一种负责处理中央处理器（CPU）的内存访问请求的计算机硬件。它的功能包括虚拟地址到物理地址的转换（即虚拟内存管理）、内存保护、中央处理器高速缓存的控制等。

context switch：上下文切换，又称环境切换。是一个存储和重建 CPU 状态的机制。要交换 CPU 上的进程时，必需先行存储当前进程的状态，然后再将进程状态读回 CPU 中。
驱动程序(device driver)：设备驱动程序，简称驱动程序（driver），是一个允许高级别电脑软件与硬件交互的程序，这种程序创建了一个硬件与硬件，或硬件与软件沟通的接口，经由主板上的总线或其它沟通子系统与硬件形成连接的机制，这样使得硬件设备上的数据交换成为可能。

忙等(busy waiting)：在软件工程中，忙碌等待也称自旋，是一种以进程反复检查一个条件是否为真的条件，这种机制可能为检查键盘输入或某个锁是否可用。
中断(Interrupt)：通常，在接收到来自外围硬件（相对于中央处理器和内存）的异步信号，或来自软件的同步信号之后，处理器将会进行相应的硬件／软件处理。发出这样的信号称为进行中断请求（interrupt request，IRQ）。硬件中断导致处理器通过一个运行信息切换（context switch）来保存执行状态（以程序计数器和程序状态字等寄存器信息为主）；软件中断则通常作为 CPU 指令集中的一个指令，以可编程的方式直接指示这种运行信息切换，并将处理导向一段中断处理代码。中断在计算机多任务处理，尤其是即时系统中尤为有用。
中断向量(interrupt vector)：中断向量位于中断向量表中。中断向量表（IVT）是将中断处理程序列表与中断向量表中的中断请求列表相关联的数据结构。中断向量表的每个条目（称为中断向量）都是中断处理程序的地址。

DMA (Direct Memory Access)：直接内存访问，直接内存访问是计算机科学中的一种内存访问技术。它允许某些电脑内部的硬件子系统（电脑外设），可以独立地直接读写系统内存，而不需中央处理器（CPU）介入处理。
总线(Bus)：总线（Bus）是指计算机组件间规范化的交换数据的方式，即以一种通用的方式为各组件提供数据传送和控制逻辑。
PCIe (Peripheral Component Interconnect Express)：官方简称PCIe，是计算机总线的一个重要分支，它沿用现有的PCI编程概念及信号标准，并且构建了更加高速的串行通信系统标准。
DMI (Direct Media Interface)：直接媒体接口，是英特尔专用的总线，用于电脑主板上南桥芯片和北桥芯片之间的连接。
USB(Universal Serial Bus)：是连接计算机系统与外部设备的一种串口总线标准，也是一种输入输出接口的技术规范，被广泛地应用于个人电脑和移动设备等信息通讯产品，并扩展至摄影器材、数字电视（机顶盒）、游戏机等其它相关领域。

BIOS(Basic Input Output System)：是在通电引导阶段运行硬件初始化，以及为操作系统提供运行时服务的固件。它是开机时运行的第一个软件。

硬实时系统(hard real-time system)：硬实时性意味着你必须绝对在每个截止日期前完成任务。很少有系统有此要求。例如核系统，一些医疗应用（例如起搏器），大量国防应用，航空电子设备等。
软实时系统(soft real-time system)：软实时系统可能会错过某些截止日期，但是如果错过太多，最终性能将下降。一个很好的例子是计算机中的声音系统。
进程(Process)：程序本身只是指令、数据及其组织形式的描述，进程才是程序（那些指令和数据）的真正运行实例。若进程有可能与同一个程序相关系，且每个进程皆可以同步（循序）或异步的方式独立运行。
地址空间(address space)：地址空间是内存中可供程序或进程使用的有效地址范围。也就是说，它是程序或进程可以访问的内存。存储器可以是物理的也可以是虚拟的，用于执行指令和存储数据。
进程表(process table)：进程表是操作系统维护的数据结构，该表中的每个条目（通常称为上下文块）均包含有关进程的信息，例如进程名称和状态，优先级，寄存器以及它可能正在等待的信号灯。
命令行界面(command-line interpreter)：是在图形用户界面得到普及之前使用最为广泛的用户界面，它通常不支持鼠标，用户通过键盘输入指令，计算机接收到指令后，予以执行。

进程间通信(interprocess communication)：指至少两个进程或线程间传送数据或信号的一些技术或方法。
超级用户(superuser)：也被称为管理员帐户，在计算机操作系统领域中指一种用于进行系统管理的特殊用户，其在系统中的实际名称也因系统而异，如 root、administrator 与supervisor。
目录(directory): 在计算机或相关设备中，一个目录或文件夹就是一个装有数字文件系统的虚拟容器。在它里面保存着一组文件和其它一些目录。
路径(path name)：路径是一种电脑文件或目录的名称的通用表现形式，它指向文件系统上的一个唯一位置。
根目录(root directory)：根目录指的就是计算机系统中的顶层目录，比如 Windows 中的 C 盘和 D 盘，Linux 中的 /。
工作目录(Working directory)：它是一个计算机用语。用户在操作系统内所在的目录，用户可在此目录之下，用相对文件名访问文件。
文件描述符(file descriptor)：文件描述符是计算机科学中的一个术语，是一个用于表述指向文件的引用的抽象化概念。
inode：索引节点的缩写，索引节点是 UNIX 系统中包含的信息，其中包含有关每个文件的详细信息，例如节点，所有者，文件，文件位置等。
共享库(shared library)：共享库是一个包含目标代码的文件，执行过程中多个 a.out 文件可能会同时使用该目标代码。
DLLs (Dynamic-Link Libraries)：动态链接库，它是微软公司在操作系统中实现共享函数库概念的一种实现方式。这些库函数的扩展名是 .DLL、.OCX（包含ActiveX控制的库）或者.DRV（旧式的系统驱动程序）。
客户端(clients)：客户端是访问服务器提供的服务的计算机硬件或软件。
服务端(servers)：在计算中，服务器是为其他程序或设备提供功能的计算机程序或设备，称为服务端
主从架构(client-server)：主从式架构也称客户端/服务器架构、C/S 架构，是一种网络架构，它把客户端与服务器区分开来。每一个客户端软件的实例都可以向一个服务器或应用程序服务器发出请求。有很多不同类型的服务器，例如文件服务器、游戏服务器等。

虚拟机(Virtual Machines)：在计算机科学中的体系结构里，是指一种特殊的软件，可以在计算机平台和终端用户之间创建一种环境，而终端用户则是基于虚拟机这个软件所创建的环境来操作其它软件。

Java 虚拟机(Java virtual Machines)：Java虚拟机有自己完善的硬体架构，如处理器、堆栈、寄存器等，还具有相应的指令系统。JVM屏蔽了与具体操作系统平台相关的信息，使得Java程序只需生成在Java虚拟机上运行的目标代码（字节码），就可以在多种平台上不加修改地运行。
目标文件(object file)：目标文件是包含目标代码的文件，这意味着通常无法直接执行的可重定位格式的机器代码。目标文件有多种格式，相同的目标代码可以打包在不同的目标文件中。目标文件也可以像共享库一样工作。
C preprocessor： C 预处理å器是 C 语言、C++ 语言的预处理器。用于在编译器处理程序之前预扫描源代码，完成头文件的包含, 宏扩展, 条件编译, 行控制等操作。

文章参考：

blog.csdn.net/zhangjg_blo…

www.techopedia.com/definition/…

en.wikipedia.org/wiki/Direct…

en.wikipedia.org/wiki/Bus_(c…

en.wikipedia.org/wiki/Interr…

en.wikipedia.org/wiki/Busy_w…

en.wikipedia.org/wiki/Contex…

en.wikipedia.org/wiki/Read-o…

www.techopedia.com/definition/…

zhuanlan.zhihu.com/p/37749443

en.wikipedia.org/wiki/Pipeli…

en.wikipedia.org/wiki/Stack_…

en.wikipedia.org/wiki/Distri…

en.wikipedia.org/wiki/Time-s…

zh.wikipedia.org/wiki/UNIX

zh.wikipedia.org/wiki/UNIX_S…

en.wikipedia.org/wiki/Networ…

zh.wikipedia.org/zh/X86-64

zh.wikipedia.org/zh/X86

en.wikipedia.org/wiki/Cloud_…

www.techopedia.com/definition/…

zh.wikipedia.org/wiki/SATA

blog.codinghorror.com/understandi…

en.wikipedia.org/wiki/Protec…

本文转载自: 掘金

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

Nestjs 从零到壹系列（六）：用 15 行代码实现 R

发表于 2020-03-30

上一篇介绍了如何使用 DTO 和管道对入参进行验证，接下来介绍一下如何用拦截器，实现后台管理系统中最复杂、也最令人头疼的 RBAC。

GitHub 项目地址，欢迎各位大佬 Star。

RBAC

1. 什么是 RBAC ？

RBAC：基于角色的权限访问控制（Role-Based Access Control），是商业系统中最常见的权限管理技术之一。在 RBAC 中，权限与角色相关联，用户通过成为适当角色的成员而得到这些角色的权限。这就极大地简化了权限的管理。

2. RBAC 模型的分类

RBAC 模型可以分为：RBAC 0、RBAC 1、RBAC 2、RBAC 3 四种。

其中 RBAC 0 是基础，也是最简单的，相当于底层逻辑。RBAC 1、RBAC 2、RBAC 3 都是以 RBAC 0 为基础的升级。

2.1 RBAC 0

最简单的用户、角色、权限模型。这里面又包含了2种：

用户和角色是多对一关系，即：一个用户只充当一种角色，一种角色可以有多个用户担当。
用户和角色是多对多关系，即：一个用户可同时充当多种角色，一种角色可以有多个用户担当。

一般情况下，使用 RBAC 0 模型就可以满足常规的权限管理系统设计了。

2.2 RBAC 1

相对于RBAC0模型，增加了子角色，引入了继承概念，即子角色可以继承父角色的所有权限。

2.3 RBAC 2

基于RBAC0模型，增加了对角色的一些限制：角色互斥、基数约束、先决条件角色等。

【角色互斥】：同一用户不能分配到一组互斥角色集合中的多个角色，互斥角色是指权限互相制约的两个角色。案例：财务系统中一个用户不能同时被指派给会计角色和审计员角色。
【基数约束】：一个角色被分配的用户数量受限，它指的是有多少用户能拥有这个角色。例如：一个角色专门为公司 CEO 创建的，那这个角色的数量是有限的。
【先决条件角色】：指要想获得较高的权限，要首先拥有低一级的权限。例如：先有副总经理权限，才能有总经理权限。
【运行时互斥】：例如，允许一个用户具有两个角色的成员资格，但在运行中不可同时激活这两个角色。

2.4 RBAC 3

称为统一模型，它包含了 RBAC 1 和 RBAC 2，利用传递性，也把 RBAC 0 包括在内，综合了 RBAC 0、RBAC 1 和 RBAC 2 的所有特点，这里就不在多描述了。

具体实现

由于是入门教程，这里只演示 RBAC 0 模型的实现，即一个用户只能有一种角色，不存在交叉关系。

正所谓：道生一，一生二，二生三，三生万物。学会 RBAC 0 之后，相信读者们一定能结合概念，继续扩展权限系统的。

其实 RBAC 0 实现起来非常简单，简单到核心代码都不超过 15 行。

1. 拦截器逻辑编写

还记得第三篇签发 Token 的时候，有个 role 字段么？那个就是用户角色，下面我们针对 Token 的 role 字段进行展开。先新建文件：

1	复制代码$ nest g interceptor rbac interceptor

复制代码// src/interceptor/rbac.interceptor.ts
import { CallHandler, ExecutionContext, Injectable, NestInterceptor, ForbiddenException } from '@nestjs/common';
import { Observable } from 'rxjs';

@Injectable()
export class RbacInterceptor implements NestInterceptor {
  // role[用户角色]: 0-超级管理员 | 1-管理员 | 2-开发&测试&运营 | 3-普通用户（只能查看）
  constructor(private readonly role: number) {}
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const req = context.getArgByIndex(1).req;
    if (req.user.role > this.role) {
      throw new ForbiddenException('对不起，您无权操作');
    }
    return next.handle();
  }
}

上面就是验证的核心代码，抛开注释，总共才15行，

构造器里的 role: number 是通过路由传入的可配置参数，表示必须小于等于这个数字的角色才能访问。通过获取用户角色的数字，和传入的角色数字进行比较即可。

2. 测试准备

和第二篇一样，直接复制下列 SQL语句到 navicat 查询模块，运行，创建新表：

复制代码CREATE TABLE `commodity` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '商品ID',
  `ccolumn_id` smallint(6) NOT NULL COMMENT '商品_栏目ID',
  `commodity_name` varchar(10) NOT NULL COMMENT '商品_名称',
  `commodity_desc` varchar(20) NOT NULL COMMENT '商品_介绍',
  `market_price` decimal(7,2) NOT NULL DEFAULT '0.00' COMMENT '市场价',
  `sale_money` decimal(7,2) NOT NULL DEFAULT '0.00' COMMENT '销售价',
  `c_by` varchar(24) NOT NULL COMMENT '创建人',
  `c_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `u_by` varchar(24) NOT NULL DEFAULT '0' COMMENT '修改人',
  `u_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间',
  PRIMARY KEY (`id`),
  KEY `idx_ccid` (`ccolumn_id`),
  KEY `idx_cn` (`commodity_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='商品表';

3. 编写业务逻辑

创建 commodity 模块，之前的教程已经教过，这里不再赘述，直接切入正题，先编写 Service：

复制代码// src/logical/commodity/commodity.service.js
import { Injectable } from '@nestjs/common';
import * as Sequelize from 'sequelize'; // 引入 Sequelize 库
import sequelize from '../../database/sequelize'; // 引入 Sequelize 实例

@Injectable()
export class CommodityService {
  /**
   * 查询商品列表
   * @param {*} body
   * @param {string} username
   * @returns {Promise<any>}
   * @memberof CommodityService
   */
  async queryCommodityList(body: any): Promise<any> {
    const { pageIndex = 1, pageSize = 10, keywords = '' } = body;
    // 分页查询条件
    const currentIndex = (pageIndex - 1) * pageSize < 0 ? 0 : (pageIndex - 1) * pageSize;
    const queryCommodityListSQL = `
      SELECT
        id, ccolumn_id columnId, commodity_name name, commodity_desc description,
        sale_money saleMoney, market_price marketPrice,
        c_by createBy, DATE_FORMAT(c_time, '%Y-%m-%d %H:%i:%s') createTime,
        u_by updateBy, DATE_FORMAT(u_time, '%Y-%m-%d %H:%i:%s') updateTime
      FROM
        commodity
      WHERE
        commodity_name LIKE '%${keywords}%'
      ORDER BY
        id DESC
      LIMIT ${currentIndex}, ${pageSize}
    `;
    const commodityList: any[] = await sequelize.query(queryCommodityListSQL, {
      type: Sequelize.QueryTypes.SELECT,
      raw: true,
      logging: false,
    });
    
    // 统计数据条数
    const countCommodityListSQL = `
      SELECT
        COUNT(*) AS total
      FROM
        commodity
      WHERE
        commodity_name LIKE '%${keywords}%'
    `;
    const count: any = (
      await sequelize.query(countCommodityListSQL, {
        type: Sequelize.QueryTypes.SELECT,
        raw: true,
        logging: false,
      })
    )[0];

    return {
      code: 200,
      data: {
        commodityList,
        total: count.total,
      },
    };
  }

  /**
   * 创建商品
   *
   * @param {*} body
   * @param {string} username
   * @returns {Promise<any>}
   * @memberof CommodityService
   */
  async createCommodity(body: any, username: string): Promise<any> {
    const { columnId = 0, name, description = '', marketPrice = 0, saleMoney = 0 } = body;
    const createCommoditySQL = `
      INSERT INTO commodity
        (ccolumn_id, commodity_name, commodity_desc, market_price, sale_money, c_by)
      VALUES
        ('${columnId}', '${name}', '${description}', ${marketPrice}, ${saleMoney}, '${username}');
    `;
    await sequelize.query(createCommoditySQL, { logging: false });
    return {
      code: 200,
      msg: 'Success',
    };
  }

  /**
   * 修改商品
   *
   * @param {*} body
   * @param {string} username
   * @returns
   * @memberof CommodityService
   */
  async updateCommodity(body: any, username: string) {
    const { id, columnId, name, description, saleMoney, marketPrice } = body;

    const updateCommoditySQL = `
      UPDATE
        commodity
      SET
        ccolumn_id = ${columnId},
        commodity_name = '${name}',
        commodity_desc = '${description}',
        market_price = ${marketPrice},
        sale_money = ${saleMoney},
        u_by = '${username}'
      WHERE
        id = ${id}
    `;
    const transaction = await sequelize.transaction();
    await sequelize.query(updateCommoditySQL, { transaction, logging: false });
    return {
      code: 200,
      msg: 'Success',
    };
  }

  /**
   * 删除商品
   *
   * @param {*} body
   * @returns
   * @memberof CommodityService
   */
  async deleteCommodity(body: any) {
    const { id } = body;
    const deleteCommoditySQL = `
      DELETE FROM
        commodity
      WHERE
        id = ${id}
    `;
    await sequelize.query(deleteCommoditySQL, { logging: false });
    return {
      code: 200,
      msg: 'Success',
    };
  }
}

上面的代码就包含了增、删、改、查，基本就涵盖了平时 80% 的搬砖内容。为了快速验证效果，这里就没有使用 DTO 进行参数验证，平时大家还是要加上比较好。

接下来编写 Controller，并引入 RBAC 拦截器：

复制代码// src/logical/commodity/commodity.controller.js
import { Controller, Request, Post, Body, UseGuards, UseInterceptors } from '@nestjs/common';
import { AuthGuard } from '@nestjs/passport';
import { CommodityService } from './commodity.service';
import { RbacInterceptor } from '../../interceptor/rbac.interceptor';

@Controller('commodity')
export class CommodityController {
  constructor(private readonly commodityService: CommodityService) {}

  // 查询商品列表
  @UseGuards(AuthGuard('jwt'))
  @UseInterceptors(new RbacInterceptor(3)) // 调用 RBAC 拦截器
  @Post('list')
  async queryColumnList(@Body() body: any) {
    return await this.commodityService.queryCommodityList(body);
  }

  // 新建商品
  @UseGuards(AuthGuard('jwt'))
  @UseInterceptors(new RbacInterceptor(2))
  @Post('create')
  async createCommodity(@Body() body: any, @Request() req: any) {
    return await this.commodityService.createCommodity(body, req.user.username);
  }

  // 修改商品
  @UseGuards(AuthGuard('jwt'))
  @UseInterceptors(new RbacInterceptor(2))
  @Post('update')
  async updateCommodity(@Body() body: any, @Request() req: any) {
    return await this.commodityService.updateCommodity(body, req.user.username);
  }

  // 删除商品
  @UseGuards(AuthGuard('jwt'))
  @UseInterceptors(new RbacInterceptor(1))
  @Post('delete')
  async deleteCommodity(@Body() body: any) {
    return await this.commodityService.deleteCommodity(body);
  }
}

和平时的路由没什么区别，就是使用了 @UseInterceptors(new RbacInterceptor())，并把数字传入，这样就可以判断权限了。

4. 验证

这是之前注册的用户表，在没有修改权限的情况下，角色 role 都是 3：

先往商品表插入一些数据：

我将使用 nodejs 用户登录，并请求查询接口：

上图的查询结果，也符合预期，共有 2 条商品名称含有关键字 德玛。

接下来，我们新建商品（英雄）：

上图可以看到，因为权限不足，所以被拦截了。

我们直接去数据库修改角色 role 字段，将 3（普通用户） 改为 2（开发&测试&运营）：

然后，重新登录，重新登录，重新登录，重要的事情说 3 遍，再请求：

返回成功信息，再看看数据库：

如图，创建商品功能测试成功。

但是，“麦林炮手”的价格应该是 1350，我们修改一下价格：

再看看数据库，通过 u_by 字段可以知道是通过接口修改的：

现在问题来了，因为麦林炮手的介绍不太“和谐”，所以需要删除，于是我们请求一下删除接口：

返回“无权操作”，只好提升角色，或者联系管理员帮忙删除啦，剩下的事情和之前的一样，不再赘述。

5. 优化

大家可能发现，因为传入的是数字，所以在 Controller 里写的也都是数字，如果是一个人维护的还好，但是多人协同时，就显得不够友好了。

于是，我们应该创建常量，将角色和数字对应上，这样再看 Controller 的时候，哪些接口有哪些角色可以访问就一目了然了。

我们修改 auth 目录下的 constants.ts

复制代码// src/logical/auth/constants.ts
export const jwtConstants = {
  secret: 'shinobi7414',
};

export const roleConstans = {
  SUPER_ADMIN: 0, // 超级管理员
  ADMIN: 1, // 管理员
  DEVELOPER: 2, // 开发者（测试、运营具有同一权限，若提升为 RBAC 1 以上，则可酌情分开）
  HUMAN: 3 // 普通用户
};

然后修改 Controller，用常量替换数字：

复制代码// src/logical/commodity/commodity.controller.js
import { Controller, Request, Post, Body, UseGuards, UseInterceptors } from '@nestjs/common';
import { AuthGuard } from '@nestjs/passport';
import { CommodityService } from './commodity.service';
import { RbacInterceptor } from '../../interceptor/rbac.interceptor';
import { roleConstans as role } from '../auth/constants'; // 引入角色常量

@Controller('commodity')
export class CommodityController {
  constructor(private readonly commodityService: CommodityService) {}

  // 查询商品列表
  @UseGuards(AuthGuard('jwt'))
  @UseInterceptors(new RbacInterceptor(role.HUMAN))
  @Post('list')
  async queryColumnList(@Body() body: any) {
    return await this.commodityService.queryCommodityList(body);
  }

  // 新建商品
  @UseGuards(AuthGuard('jwt'))
  @UseInterceptors(new RbacInterceptor(role.DEVELOPER))
  @Post('create')
  async createCommodity(@Body() body: any, @Request() req: any) {
    return await this.commodityService.createCommodity(body, req.user.username);
  }

  // 修改商品
  @UseGuards(AuthGuard('jwt'))
  @UseInterceptors(new RbacInterceptor(role.DEVELOPER))
  @Post('update')
  async updateCommodity(@Body() body: any, @Request() req: any) {
    return await this.commodityService.updateCommodity(body, req.user.username);
  }

  // 删除商品
  @UseGuards(AuthGuard('jwt'))
  @UseInterceptors(new RbacInterceptor(role.ADMIN))
  @Post('delete')
  async deleteCommodity(@Body() body: any) {
    return await this.commodityService.deleteCommodity(body);
  }
}

如此一来，什么角色才有权限操作就一目了然。

2020-3-31 更新：使用 Guard 守卫控制权限

评论区有大神指出，应该使用 Guard 来管理角色相关，因此，在这里补充一下 Guard 的实现。

新建 Guard 文件：

1	复制代码$ nest g guard rbac guards

编写守卫逻辑：

复制代码// src/guards/rbac.guard.ts
import { CanActivate, ExecutionContext, Injectable, ForbiddenException } from '@nestjs/common';
import { Observable } from 'rxjs';

@Injectable()
export class RbacGuard implements CanActivate {
  // role[用户角色]: 0-超级管理员 | 1-管理员 | 2-开发&测试&运营 | 3-普通用户（只能查看）
  constructor(private readonly role: number) {}
  canActivate(
    context: ExecutionContext,
  ): boolean | Promise<boolean> | Observable<boolean> {
    const request = context.switchToHttp().getRequest();
    const user = request.user;
    if (user.role > this.role) {
      throw new ForbiddenException('对不起，您无权操作');
    }
    return true;
  }
}

去掉注释和 TSLint 的换行，同样不超过 15 行，接下来，在 Controller 里引入：

复制代码// src/logical/commodity/commodity.controller.ts
import { Controller, Request, Post, Body, UseGuards, UseInterceptors } from '@nestjs/common';
import { AuthGuard } from '@nestjs/passport';
import { CommodityService } from './commodity.service';
import { RbacInterceptor } from '../../interceptor/rbac.interceptor';
import { RbacGuard } from '../../guards/rbac.guard';
import { roleConstans as role } from '../auth/constants';

@Controller('commodity')
export class CommodityController {
  constructor(private readonly commodityService: CommodityService) {}

  // 查询商品列表
  @UseGuards(new RbacGuard(role.HUMAN))
  @UseGuards(AuthGuard('jwt'))
  // @UseInterceptors(new RbacInterceptor(role.HUMAN))
  @Post('list')
  async queryColumnList(@Body() body: any) {
    return await this.commodityService.queryCommodityList(body);
  }

  // 新建商品
  @UseGuards(new RbacGuard(role.DEVELOPER))
  @UseGuards(AuthGuard('jwt'))
  // @UseInterceptors(new RbacInterceptor(role.DEVELOPER))
  @Post('create')
  async createCommodity(@Body() body: any, @Request() req: any) {
    return await this.commodityService.createCommodity(body, req.user.username);
  }

  // 修改商品
  @UseGuards(new RbacGuard(role.DEVELOPER))
  @UseGuards(AuthGuard('jwt'))
  // @UseInterceptors(new RbacInterceptor(role.DEVELOPER))
  @Post('update')
  async updateCommodity(@Body() body: any, @Request() req: any) {
    return await this.commodityService.updateCommodity(body, req.user.username);
  }

  // 删除商品
  @UseGuards(new RbacGuard(role.ADMIN))
  @UseGuards(AuthGuard('jwt'))
  // @UseInterceptors(new RbacInterceptor(role.ADMIN))
  @Post('delete')
  async deleteCommodity(@Body() body: any) {
    return await this.commodityService.deleteCommodity(body);
  }
}

注意：RbacGuard 要在 AuthGuard 的上面，不然获取不到用户信息。

请求一下只有管理员才有权限的删除操作：

涛声依旧。

总结

本篇介绍了 RBAC 的概念，以及如何使用拦截器和守卫实现 RBAC 0，原理简单到 15 行代码就搞定了。

然而这种设计，要求路由必须是一一对应的，遇到复杂的用户关系，还需要再建 3 张表，一张是 权限 表，一张是 用户-权限 对应表，还有一张是 路由-权限 对应表，这样基本能覆盖 RBAC 2 以上的需求了。

但万变不离其宗，基本就是在拦截器或守卫里做文章，用户登录后，将权限列表缓存起来（可以是 Redis），这样就不用每次都查表去判断有没有权限访问路由了。

下一篇，暂时还不知道要介绍什么，清明节前事有点多，可能是使用 Swagger 自动生成接口文档吧。

本篇收录于NestJS 实战教程，更多文章敬请关注。

参考资料

RBAC模型：基于用户 - 角色 - 权限控制的一些思考

本文转载自: 掘金

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

背景

目前痛点

目的

TiDB整体架构

数据存储模型

数据切分

提高索引效率

TIDB特性

1.1 优缺点分析

1.2 事务的重试

2.1 Region分裂

2.2 新增存储节点

TIDB成本

使用场景

注意事项与坑点

总结

欢迎关注，文章更快一步

Cookie 和 Session

Session 是什么

Session 如何判断是否是同一会话

Session 的缺点

Cookies 是什么

创建 Cookie

Set-Cookie 和 Cookie 标头

会话 Cookies

永久性 Cookies

Cookie 的 Secure 和 HttpOnly 标记

Cookie 的作用域

JSON Web Token 和 Session Cookies 的对比

JWT 和 Session Cookies 的相同之处

什么是 Session Cookies

什么是 Json Web Tokens

JWT 的格式

拼凑在一起

JWT 和 Session Cookies 的不同

密码签名

JSON 是无状态的

可扩展性

JWT 支持跨域认证

JWT 和 Session Cookies 的选型

后记

鸣谢

前言

正文

数据库自增长序列或字段

UUID

UUID的变种

Redis生成ID

Twitter的snowflake算法

利用zookeeper生成唯一ID

MongoDB的ObjectId

By the way

前言

絮叨

前文回顾

老年代GC的时候会发生 Concurrent Mode Failure吗？

CMS垃圾回收之后进行内存碎片的整理的频率应该多高？

结尾

日常求赞

I. SSE 扫盲

1. 概念介绍

2. 特点分析

3. 应用场景

II. 手动实现 sse 功能

1. 项目创建

2. 功能实现

III. SseEmitter

1. sse 规范

2. 实现

3. 小结

IV. 其他

0. 项目

1. 一灰灰 Blog

从零开始 Docker 搭建 Redis 集群

Docker 安装

初识 Docker

Docker 的结构

Docker Container

启动你的第一个 Container

查看 Container