微服务或许你没有真正实践过，但一定听说过，虽然已经到了 2022 年，这个词依然很热，可以通过搜索 google 指数看得到。

起源

“微服务”一词源于 2011 年 5 月在威尼斯附近的一次软件架构师研讨会上进行的架构风格的讨论。2012 年 5 月 讨论小组决定将这种架构风格命名为“微服务”。Fred George 同年在一次技术大会上进行自己的微服务实践分享，并说微服务是一种细粒度的 SOA ，但最终将其发扬光大的是 Martin Fowler 2014 年写的博文《 Microservices 》，原文链接如下：

martinfowler.com/articles/microservices.html。

自此以后，微服务就家喻户晓了，Microservices 的 Google 指数也是在 2014 年后就一路飙升。

和微服务相对应的是单体架构，先来看看单体架构是怎样的。

单体架构

大多数人做软件开发应该都是从单体架构开始的，以 .NET 程序员来说，从最早的 WebForm、到后来的 MVC、再到现在的前后端分离，后端使用 .NET 的 WebAPI ，都是整个项目的代码放到一个解决方案中，发布要么直接整个目录进行替换，或者更新有变更的 dll 文件。

包括到现在，这种单体架构的模该还占着很大的比重，凡是存在，必有道理，单体架构有着他的可取之处：

• 开发方便，.NET 程序员只需只需使用宇宙最强 IDE VS 就可以。

• 调试方便，在开发阶段，所有的项目都在一个解决方案下，项目之间是可以直接引用，断点可以到达你想要的任何地方。

• 运行方便，编码完成，只需一个 F5 搞定。

• 部署方便，无论是之前部署 IIS ，还是现在的容器部署，都只涉及到一个发布目录。

不过，随着产品的功能越来越复杂，代码也会变得越来越复杂，团队的人数也会越来越多，这时单体架构就会带来一些问题：

• 因为代码库非常的臃肿，从编译、构建、运行到测试这个时间会越来越长。

• 技术栈几乎是受限的，比如一个 .NET 的工程，基本就是 C# 来开发了，不太可能混杂其他的语言。

• 不方便横向扩展，只能整套程序进行扩，满足所有模块的需求，对资源的利用率非常差。

• 不够敏捷，团队成员越来越多多时，都在同一个代码上进行修改、提交、合并，容易引发冲突和其他问题。

• 一个很小的改动点，容易引发全身问题，导致系统崩溃，因为影响点多，测试成本也会很高。

• 缺乏可靠性，我们就碰到过因为一个序列化的问题导致 CPU 占用很高，结果整个系统瘫痪了。

微服务架构

上面提到的单体架构存在的问题，采用微服务架构可以很好地解决。微服务的核心是为了解耦，构建成一个松耦合的分布式系统。

一个庞大的单体系统拆分成若干个小的服务，每个服务可以由一个小的团队来维护，团队会更加敏捷，构建发布的时间更短，代码也容易维护。

不同的微服务团队可以采用不同的技术栈，比如工作流引擎使用 .NET ，规则引擎可以使用 Java ，一些全新的模块更容易采用新的技术，人员流动和补充上也更加灵活。

每个服务通常采用独立的数据库，代码或者数据库层面的问题不会导致整个系统的崩溃。

扩展方便，这个很重要，如果监控发现流程引擎的压力很大，可以只针对这个服务进行横向扩展，服务器资源可以得到更好的利用。

上面说的都是好处，但没有任何一种技术是银弹，微服务解决问题的同时，也会带来更多的问题。

1、开发调试变得困难了，需要通过日志的方式或者借助一些远程调试工具。

2、单体架构中，模块之间的调用都是进程内，添加类库的引用后，就是本地方法的调用，微服务各自独立部署，就会涉及到进程间的通信。

3、线上问题往往需要多个服务团队一起来协作解决，会存在互相甩锅的问题。

4、在分布式系统中，事务、数据一致性、联合查询等相比较单体更加复杂。

5、持续集成、部署、运维的复杂度也显著提升。

6、随着服务越来越多，客户端怎样去找这些服务呢?

7、进程内的访问不存在网络的问题，拆分后的服务可能在同个机器的不同进程，更多的时候是不同机器的不同进程，网络问题导致服务不稳定怎么办?

为了解决这些问题，各种中间件和框架就应运而生，又会带来更多的学习成本。

在 .NET 技术栈中，会用到下面这些中间件：

• 服务注册与发现：Consul。

• 网关：Ocelot。

• 熔断降级：Polly。

• 服务链路追踪：SkyWalking 或 Twitter 的 Zipkin。

• 配置中心：Apllo。

• 鉴权中心：IdentityServer4。

在 Java 中也有 Spring Cloud 和 Spring Cloud Alibaba 这种全家桶套件可以使用。

要不要转微服务呢?

从单体到微服务是一个权衡和取舍的问题，切记不要跟风。以我的经验来看，可以分为两类：

1. 做企业级系统。

2. 做互联网系统。

做企业级应用大多都是项目交付型的，客户关系维系的好，后面可以做二期、三期，当然也有一锤子买卖的。这其中一个关键点是要快，单从快速来看，采用单体架构，开发、调试、部署都是最快的。

从客户角度来说，只要能满足业务，是单体还是微服务其实不太关心。

做互联网应用，也就是我们常说的 SaaS，也分为两种情况：

1、将现有的私有化部署的系统(单体架构)改造成支持 SaaS 的模式。

这种我也不建议一上来就大刀阔斧地进行微服务改造，可以在代码的结构上做一些调整，比如按照领域去拆分目录，不同领域之间的调用可以再进行一层抽象，目的是为了未来向微服务架构转化。

当团队的技术栈变得丰富了，比如原先只有 .NET ,现在有些模块采用的是 Java ,这时已然是朝着微服务架构发展了，只是粒度比较大而已，相应的一些中间件也需要引入，比如服务网关、服务发现、服务间通信等。

2、从零开始做一个 SaaS 系统。

互联网系统和企业级系统有很大的差别，如果说企业级系统更多关注功能性需求，那么互联网系统除了功能性需求，还需要关注非功能性需求，比如：横向扩展、限流降级、日志追踪、预警、灰度发布等。

即便因为时间关系，一开始是单体架构，我觉得也应该是微服务架构的单体，随着持续迭代和发展，根据实际情况逐步进行拆分。

如果时间上比较充裕，可以一开始就按照微服务架构进行分离，但粒度不要太小。

总结

1. 解决常说的的三高问题(高并发、高性能、高可用)，一个核心的思路就是拆，分而治之，所以说微服务肯定是能解决掉我们的很多问题，也是发展方向。

2. 实践微服务需要根据当前的实际情况，如果单体运行的很好，也没什么问题，也不要为了炫技进行微服务改造。

3. 如果决定要实践微服务，先做好单体架构的设计，让代码遵循面向对象的设计原则，否则即便形式上变成了微服务，也不能尝到微服务的甜头。

作者：不止dotNET

程序员是一个创作型的职业，频繁的加班并不能增加产出，而国内 996 的公司文化，真的一言难尽。但是如果你进到一家公司，你能从哪些观察来判断这家公司的工作强度的（加班文化）？是看大家走得早不早吗？有一定的参考意义，但是如果走得晚呢，可能是大家不敢提前走而在公司耗时间。

今天要推荐一个代码分析工具 code996，它可以统计 Git 项目的 commit 时间分布，进而推导出这个项目的编码工作强度。这算是一种对项目更了解的方式，杜绝 996 从了解数据开始。

我们先来看 code996 分析出来的结果示例，以下是分析项目的基本情况：

通过图表查看 commit 提交分布：

对比项目工作时间类型：

如果你对 code996 是如何工作的，以下是作者的说明：

因为代码是公司的很重要的资产，泄露是肯定不行的，为了解决大家的后顾之忧，该项目是完全安全的。

code996 除了能够分析项目的实际工作强度，也能用来分析我们代码编写的情况，对自身了解自己代码编写效率的时段、最近的工作强度等都是非常好的一个输入。

更多项目详情请查看如下链接。

开源项目地址：https://github.com/hellodigua/code996

开源项目作者：hellodigua

这篇文章我想和你聊一聊 Redis 的架构演化之路。

现如今 Redis 变得越来越流行，几乎在很多项目中都要被用到，不知道你在使用 Redis 时，有没有思考过，Redis 到底是如何稳定、高性能地提供服务的？

你也可以尝试回答一下以下这些问题：

• 我使用 Redis 的场景很简单，只使用单机版 Redis 会有什么问题吗？
• 我的 Redis 故障宕机了，数据丢失了怎么办？如何能保证我的业务应用不受影响？
• 为什么需要主从集群？它有什么优势？
• 什么是分片集群？我真的需要分片集群吗？
• …

如果你对 Redis 已经有些了解，肯定也听说过数据持久化、主从复制、哨兵这些概念，它们之间又有什么区别和联系呢？

如果你存在这样的疑惑，这篇文章，我会从 0 到 1，再从 1 到 N，带你一步步构建出一个稳定、高性能的 Redis 集群。

在这个过程中，你可以了解到 Redis 为了做到稳定、高性能，都采取了哪些优化方案，以及为什么要这么做？

掌握了这些原理，这样平时你在使用 Redis 时，就能够做到「游刃有余」。

这篇文章干货很多，希望你可以耐心读完。

从最简单的开始：单机版 Redis

首先，我们从最简单的场景开始。

假设现在你有一个业务应用，需要引入 Redis 来提高应用的性能，此时你可以选择部署一个单机版的 Redis 来使用，就像这样：

这个架构非常简单，你的业务应用可以把 Redis 当做缓存来使用，从 MySQL 中查询数据，然后写入到 Redis 中，之后业务应用再从 Redis 中读取这些数据，由于 Redis 的数据都存储在内存中，所以这个速度飞快。

如果你的业务体量并不大，那这样的架构模型基本可以满足你的需求。是不是很简单？

随着时间的推移，你的业务体量逐渐发展起来了，Redis 中存储的数据也越来越多，此时你的业务应用对 Redis 的依赖也越来越重。

但是，突然有一天，你的 Redis 因为某些原因宕机了，这时你的所有业务流量，都会打到后端 MySQL 上，这会导致你的 MySQL 压力剧增，严重的话甚至会压垮 MySQL。

这时你应该怎么办？

我猜你的方案肯定是，赶紧重启 Redis，让它可以继续提供服务。

但是，因为之前 Redis 中的数据都在内存中，尽管你现在把 Redis 重启了，之前的数据也都丢失了。重启后的 Redis 虽然可以正常工作，但是由于 Redis 中没有任何数据，业务流量还是都会打到后端 MySQL 上，MySQL 的压力还是很大。

这可怎么办？你陷入了沉思。

有没有什么好的办法解决这个问题？

既然 Redis 只把数据存储在内存中，那是否可以把这些数据也写一份到磁盘上呢？

如果采用这种方式，当 Redis 重启时，我们把磁盘中的数据快速恢复到内存中，这样它就可以继续正常提供服务了。

是的，这是一个很好的解决方案，这个把内存数据写到磁盘上的过程，就是「数据持久化」。

数据持久化：有备无患

现在，你设想的 Redis 数据持久化是这样的：

但是，数据持久化具体应该怎么做呢？

我猜你最容易想到的一个方案是，Redis 每一次执行写操作，除了写内存之外，同时也写一份到磁盘上，就像这样：

没错，这是最简单直接的方案。

但仔细想一下，这个方案有个问题：客户端的每次写操作，既需要写内存，又需要写磁盘，而写磁盘的耗时相比于写内存来说，肯定要慢很多！这势必会影响到 Redis 的性能。

如何规避这个问题？

我们可以这样优化：Redis 写内存由主线程来做，写内存完成后就给客户端返回结果，然后 Redis 用另一个线程去写磁盘，这样就可以避免主线程写磁盘对性能的影响。

这确实是一个好方案。除此之外，我们可以换个角度，思考一下还有什么方式可以持久化数据？

这时你就要结合 Redis 的使用场景来考虑了。

回忆一下，我们在使用 Redis 时，通常把它用作什么场景？

是的，缓存。

把 Redis 当做缓存来用，意味着尽管 Redis 中没有保存全量数据，对于不在缓存中的数据，我们的业务应用依旧可以通过查询后端数据库得到结果，只不过查询后端数据的速度会慢一点而已，但对业务结果其实是没有影响的。

基于这个特点，我们的 Redis 数据持久化还可以用「数据快照」的方式来做。

那什么是数据快照呢？

简单来讲，你可以这么理解：

1. 你把 Redis 想象成一个水杯，向 Redis 写入数据，就相当于往这个杯子里倒水
2. 此时你拿一个相机给这个水杯拍一张照片，拍照的这一瞬间，照片中记录到这个水杯中水的容量，就是水杯的数据快照

也就是说，Redis 的数据快照，是记录某一时刻下 Redis 中的数据，然后只需要把这个数据快照写到磁盘上就可以了。

它的优势在于，只在需要持久化时，把数据「一次性」写入磁盘，其它时间都不需要操作磁盘。

基于这个方案，我们可以定时给 Redis 做数据快照，把数据持久化到磁盘上。

其实，上面说的这些持久化方案，就是 Redis 的「RDB」和「AOF」：

• RDB：只持久化某一时刻的数据快照到磁盘上（创建一个子进程来做）
• AOF：每一次写操作都持久到磁盘（主线程写内存，根据策略可以配置由主线程还是子线程进行数据持久化）

它们的区别除了上面讲到的，还有以下特点：

1. RDB 采用二进制 + 数据压缩的方式写磁盘，这样文件体积小，数据恢复速度也快
2. AOF 记录的是每一次写命令，数据最全，但文件体积大，数据恢复速度慢

如果让你来选择持久化方案，你可以这样选择：

1. 如果你的业务对于数据丢失不敏感，采用 RDB 方案持久化数据
2. 如果你的业务对数据完整性要求比较高，采用 AOF 方案持久化数据

假设你的业务对 Redis 数据完整性要求比较高，选择了 AOF 方案，那此时你又会遇到这些问题：

1. AOF 记录每一次写操作，随着时间增长，AOF 文件体积会越来越大
2. 这么大的 AOF 文件，在数据恢复时变得非常慢

这怎么办？数据完整性要求变高了，恢复数据也变困难了？有没有什么方法，可以缩小文件体积？提升恢复速度呢？

我们继续来分析 AOF 的特点。

由于 AOF 文件中记录的都是每一次写操作，但对于同一个 key 可能会发生多次修改，我们只保留最后一次被修改的值，是不是也可以？

是的，这就是我们经常听到的「AOF rewrite」，你也可以把它理解为 AOF 「瘦身」。

我们可以对 AOF 文件定时 rewrite，避免这个文件体积持续膨胀，这样在恢复时就可以缩短恢复时间了。

再进一步思考一下，还有没有办法继续缩小 AOF 文件？

回顾一下我们前面讲到的，RDB 和 AOF 各自的特点：

1. RDB 以二进制 + 数据压缩方式存储，文件体积小
2. AOF 记录每一次写命令，数据最全

我们可否利用它们各自的优势呢？

当然可以，这就是 Redis 的「混合持久化」。

具体来说，当 AOF rewrite 时，Redis 先以 RDB 格式在 AOF 文件中写入一个数据快照，再把在这期间产生的每一个写命令，追加到 AOF 文件中。因为 RDB 是二进制压缩写入的，这样 AOF 文件体积就变得更小了。

此时，你在使用 AOF 文件恢复数据时，这个恢复时间就会更短了！

Redis 4.0 以上版本才支持混合持久化。

这么一番优化，你的 Redis 再也不用担心实例宕机了，当发生宕机时，你就可以用持久化文件快速恢复 Redis 中的数据。

但这样就没问题了吗？

仔细想一下，虽然我们已经把持久化的文件优化到最小了，但在恢复数据时依旧是需要时间的，在这期间你的业务应用还是会受到影响，这怎么办？

我们来分析有没有更好的方案。

一个实例宕机，只能用恢复数据来解决，那我们是否可以部署多个 Redis 实例，然后让这些实例数据保持实时同步，这样当一个实例宕机时，我们在剩下的实例中选择一个继续提供服务就好了。

没错，这个方案就是接下来要讲的「主从复制：多副本」。

主从复制：多副本

此时，你可以部署多个 Redis 实例，架构模型就变成了这样：

我们这里把实时读写的节点叫做 master，另一个实时同步数据的节点叫做 slave。

采用多副本的方案，它的优势是：

1. 缩短不可用时间：master 发生宕机，我们可以手动把 slave 提升为 master 继续提供服务
2. 提升读性能：让 slave 分担一部分读请求，提升应用的整体性能

这个方案不错，不仅节省了数据恢复的时间，还能提升性能，那它有什么问题吗？

你可以思考一下。

其实，它的问题在于：当 master 宕机时，我们需要「手动」把 slave 提升为 master，这个过程也是需要花费时间的。

虽然比恢复数据要快得多，但还是需要人工介入处理。一旦需要人工介入，就必须要算上人的反应时间、操作时间，所以，在这期间你的业务应用依旧会受到影响。

怎么解决这个问题？我们是否可以把这个切换的过程，变成自动化呢？

对于这种情况，我们需要一个「故障自动切换」机制，这就是我们经常听到的「哨兵」所具备的能力。

哨兵：故障自动切换

现在，我们可以引入一个「观察者」，让这个观察者去实时监测 master 的健康状态，这个观察者就是「哨兵」。

具体如何做？

1. 哨兵每间隔一段时间，询问 master 是否正常
2. master 正常回复，表示状态正常，回复超时表示异常
3. 哨兵发现异常，发起主从切换

有了这个方案，就不需要人去介入处理了，一切就变得自动化了，是不是很爽？

但这里还有一个问题，如果 master 状态正常，但这个哨兵在询问 master 时，它们之间的网络发生了问题，那这个哨兵可能会误判。

这个问题怎么解决？

答案是，我们可以部署多个哨兵，让它们分布在不同的机器上，它们一起监测 master 的状态，流程就变成了这样：

1. 多个哨兵每间隔一段时间，询问 master 是否正常
2. master 正常回复，表示状态正常，回复超时表示异常
3. 一旦有一个哨兵判定 master 异常（不管是否是网络问题），就询问其它哨兵，如果多个哨兵（设置一个阈值）都认为 master 异常了，这才判定 master 确实发生了故障
4. 多个哨兵经过协商后，判定 master 故障，则发起主从切换

所以，我们用多个哨兵互相协商来判定 master 的状态，这样一来，就可以大大降低误判的概率。

哨兵协商判定 master 异常后，这里还有一个问题：由哪个哨兵来发起主从切换呢？

答案是，选出一个哨兵「领导者」，由这个领导者进行主从切换。

问题又来了，这个领导者怎么选？

想象一下，在现实生活中，选举是怎么做的？

是的，投票。

在选举哨兵领导者时，我们可以制定这样一个选举规则：

1. 每个哨兵都询问其它哨兵，请求对方为自己投票
2. 每个哨兵只投票给第一个请求投票的哨兵，且只能投票一次
3. 首先拿到超过半数投票的哨兵，当选为领导者，发起主从切换

其实，这个选举的过程就是我们经常听到的：分布式系统领域中的「共识算法」。

什么是共识算法？

我们在多个机器部署哨兵，它们需要共同协作完成一项任务，所以它们就组成了一个「分布式系统」。

在分布式系统领域，多个节点如何就一个问题达成共识的算法，就叫共识算法。

在这个场景下，多个哨兵共同协商，选举出一个都认可的领导者，就是使用共识算法完成的。

这个算法还规定节点的数量必须是奇数个，这样可以保证系统中即使有节点发生了故障，剩余超过「半数」的节点状态正常，依旧可以提供正确的结果，也就是说，这个算法还兼容了存在故障节点的情况。

共识算法在分布式系统领域有很多，例如 Paxos、Raft，哨兵选举领导者这个场景，使用的是 Raft 共识算法，因为它足够简单，且易于实现。

现在，我们用多个哨兵共同监测 Redis 的状态，这样一来，就可以避免误判的问题了，架构模型就变成了这样：

好了，到这里我们先小结一下。

你的 Redis 从最简单的单机版，经过数据持久化、主从多副本、哨兵集群，这一路优化下来，你的 Redis 不管是性能还是稳定性，都越来越高，就算节点发生故障，也不用担心了。

你的 Redis 以这样的架构模式部署，基本上就可以稳定运行很长时间了。

…

随着时间的发展，你的业务体量开始迎来了爆炸性增长，此时你的架构模型，还能够承担这么大的流量吗？

我们一起来分析一下：

1. 稳定性：Redis 故障宕机，我们有哨兵 + 副本，可以自动完成主从切换
2. 性能：读请求量增长，我们可以再部署多个 slave，读写分离，分担读压力
3. 性能：写请求量增长，但我们只有一个 master 实例，这个实例达到瓶颈怎么办？

看到了么，当你的写请求量越来越大时，一个 master 实例可能就无法承担这么大的写流量了。

要想完美解决这个问题，此时你就需要考虑使用「分片集群」了。

分片集群：横向扩展

什么是「分片集群」？

简单来讲，一个实例扛不住写压力，那我们是否可以部署多个实例，然后把这些实例按照一定规则组织起来，把它们当成一个整体，对外提供服务，这样不就可以解决集中写一个实例的瓶颈问题吗？

所以，现在的架构模型就变成了这样：

现在问题又来了，这么多实例如何组织呢？

我们制定规则如下：

1. 每个节点各自存储一部分数据，所有节点数据之和才是全量数据
2. 制定一个路由规则，对于不同的 key，把它路由到固定一个实例上进行读写

而分片集群根据路由规则所在位置的不同，还可以分为两大类：

1. 客户端分片
2. 服务端分片

客户端分片指的是，key 的路由规则放在客户端来做，就是下面这样：

这个方案的缺点是，客户端需要维护这个路由规则，也就是说，你需要把路由规则写到你的业务代码中。

如何做到不把路由规则耦合在业务代码中呢？

你可以这样优化，把这个路由规则封装成一个模块，当需要使用时，集成这个模块就可以了。

这就是 Redis Cluster 的采用的方案。

Redis Cluster 内置了哨兵逻辑，无需再部署哨兵。

当你使用 Redis Cluster 时，你的业务应用需要使用配套的 Redis SDK，这个 SDK 内就集成好了路由规则，不需要你自己编写了。

再来看服务端分片。

这种方案指的是，路由规则不放在客户端来做，而是在客户端和服务端之间增加一个「中间代理层」，这个代理就是我们经常听到的 Proxy。

而数据的路由规则，就放在这个 Proxy 层来维护。

这样一来，你就无需关心服务端有多少个 Redis 节点了，只需要和这个 Proxy 交互即可。

Proxy 会把你的请求根据路由规则，转发到对应的 Redis 节点上，而且，当集群实例不足以支撑更大的流量请求时，还可以横向扩容，添加新的 Redis 实例提升性能，这一切对于你的客户端来说，都是透明无感知的。

业界开源的 Redis 分片集群方案，例如 Twemproxy、Codis 就是采用的这种方案。

分片集群在数据扩容时，还涉及到了很多细节，这块内容不是本文章重点，所以暂不详述。

至此，当你使用分片集群后，对于未来更大的流量压力，都可以从容面对了！

总结

好了，我们来总结一下，我们是如何一步步构建一个稳定、高性能的 Redis 集群的。

首先，在使用最简单的单机版 Redis 时，我们发现当 Redis 故障宕机后，数据无法恢复的问题，因此我们想到了「数据持久化」，把内存中的数据也持久化到磁盘上一份，这样 Redis 重启后就可以从磁盘上快速恢复数据。

在进行数据持久化时，我们又面临如何更高效地将数据持久化到磁盘的问题。之后我们发现 Redis 提供了 RDB 和 AOF 两种方案，分别对应了数据快照和实时的命令记录。当我们对数据完整性要求不高时，可以选择 RDB 持久化方案。如果对于数据完整性要求较高，那么可以选择 AOF 持久化方案。

但是我们又发现，AOF 文件体积会随着时间增长变得越来越大，此时我们想到的优化方案是，使用 AOF rewrite 的方式对其进行瘦身，减小文件体积，再后来，我们发现可以结合 RDB 和 AOF 各自的优势，在 AOF rewrite 时使用两者结合的「混合持久化」方式，又进一步减小了 AOF 文件体积。

之后，我们发现尽管可以通过数据恢复的方式还原数据，但恢复数据也是需要花费时间的，这意味着业务应用还是会受到影响。我们进一步优化，采用「多副本」的方案，让多个实例保持实时同步，当一个实例故障时，可以手动把其它实例提升上来继续提供服务。

但是这样也有问题，手动提升实例上来，需要人工介入，人工介入操作也需要时间，我们开始想办法把这个流程变得自动化，所以我们又引入了「哨兵」集群，哨兵集群通过互相协商的方式，发现故障节点，并可以自动完成切换，这样就大幅降低了对业务应用的影响。

最后，我们把关注点聚焦在如何支撑更大的写流量上，所以，我们又引入了「分片集群」来解决这个问题，让多个 Redis 实例分摊写压力，未来面对更大的流量，我们还可以添加新的实例，横向扩展，进一步提升集群的性能。

至此，我们的 Redis 集群才得以长期稳定、高性能的为我们的业务提供服务。

这里我画了一个思维导图，方便你更好地去理解它们之间的关系，以及演化的过程。

后记

看到这里，我想你对如何构建一个稳定、高性能的 Redis 集群问题时，应该会有自己的见解了。

其实，这篇文章所讲的优化思路，围绕的主题就是「架构设计」的核心思想：

• 高性能：读写分离、分片集群
• 高可用：数据持久化、多副本、故障自动切换
• 易扩展：分片集群、横向扩展

当我们讲到哨兵集群、分片集群时，这还涉及到了「分布式系统」相关的知识：

• 分布式共识：哨兵领导者选举
• 负载均衡：分片集群数据分片、数据路由

当然，除了 Redis 之外，对于构建任何一个数据集群，你都可以沿用这个思路去思考、去优化，看看它们到底是如何做的。

例如当你在使用 MySQL 时，你可以思考一下 MySQL 与 Redis 有哪些不同？MySQL 为了做到高性能、高可用，又是如何做的？其实思路都是类似的。

我们现在到处可见分布式系统、数据集群，我希望通过这篇文章，你可以理解这些软件是如何一步步演化过来的，在演化过程中，它们遇到了哪些问题，为了解决这些问题，这些软件的设计者设计了怎样的方案，做了哪些取舍？

你只有了解了其中的原理，掌握了分析问题、解决问题的能力，这样在以后的开发过程中，或是学习其它优秀软件时，就能快速地找到「重点」，在最短的时间掌握它，并能在实际应用中发挥它们的优势。

其实这个思考过程，也是做「架构设计」的思路。在做软件架构设计时，你面临的场景就是发现问题、分析问题、解决问题，一步步去演化、升级你的架构，最后在性能、可靠性方面达到一个平衡。虽然各种软件层出不穷，但架构设计的思想不会变，我希望你真正吸收的是这些思想，这样才可以做到以不变应万变。

来源：mp.weixin.qq.com/s/q79ji-cgfUMo7H0p254QRg

1 事件回顾

上周，谷歌Go语言项目负责人Steve Francia宣布辞去职务，而他给出理由是：Go项目的工作停滞不前，让他感到难受。

有意思的是，部分国内的Gopher（Go语言爱好者的自称）对Go语言也产生了新想法。比如，国内第一批Go语言爱好者之一的柴树杉、全球Go贡献者榜上长期排名TOP 50的史斌等Gopher，他们决定以Go语言为蓝本，发起新的编程语言：凹语言™（凹读音“Wa”）。

目前凹语言™的代码已经在Github开源，并且提供了简单可执行的示例。根据其仓库的介绍，凹语言™的设计目标有以下几个：

1、披着Go和Rust语法外衣的C++语言

2、凹语言™源码文件后缀为.wa

3、凹语言™编译器兼容WaGo语法，凹语法与WaGo语法在AST层面一致（二者可生成相同的AST并无损的互相转换）

4、凹语言™支持中文/英文双语关键字，即任一关键字均有中文版和英文版，二者在语法层面等价

凹语言™示意，图片来源@GitHub

据柴树杉、史斌等人的说法，Go语言“克制”的风格是他们对编程语言审美的最大公约数。因此，凹语言™项目启动时大量借鉴了Go的设计思想和具体实现。

当然，他们也表示，选择Go语言作为初始的蓝本，是在有限投入下不得不作出的折衷。他们希望随着项目的发展，积累更多原创的设计，为自主创新的大潮贡献一点力量。

虽说柴树杉、史斌等人是资深的Gopher，偏爱Go语言并不难理解，但我们还是忍不住好奇：究竟Go语言有多神奇，让他们对Go语言这么着迷？

2 为什么选中Go语言

从许多使用过Go语言的开发者对Go的评价上看，Go语言在设计上有以下四个特点。

1、简单易用

不同于那些通过相互借鉴而不断增加新特性的主流编程语言（如C++、Java等），Go的设计者们在语言设计之初就拒绝走语言特性融合的道路，而选择了“做减法”。

他们把复杂留给了语言自身的设计和实现，留给了Go核心开发组，而将简单、易用和清晰留给了广大使用Go语言的开发者。因此，Go语言呈现出：

• 简洁、常规的语法（不需要解析符号表），仅有25个关键字；

• 没有头文件；

• 显式依赖（package）；

• 没有循环依赖（package）；

• 常量只是数字；

• 首字母大小写决定可见性；

• 任何类型都可以拥有方法（没有类）；

• 没有子类型继承（没有子类）；

• 没有算术转换；

• 没有构造函数或析构函数；

• 赋值不是表达式；

• 在赋值和函数调用中定义的求值顺序（无“序列点”概念）；

• 没有指针算术；

• 内存总是初始化为零值；

• 没有类型注解语法（如C++中的const、static等）

• ……

2、偏好组合

C++、Java等主流面向对象语言，通过庞大的自上而下的类型体系、继承、显式接口实现等机制，将程序的各个部分耦合起来，但在Go语言中我们找不到经典面向对象的语法元素、类型体系和继承机制。

那Go语言是如何将程序的各个部分耦合在一起呢？是组合。

在语言设计层面，Go使用了正交的语法元素，包括Go语言无类型体系，类型之间是独立的，没有子类型的概念；每个类型都可以有自己的方法集合，类型定义与方法实现是正交独立的。

各类型之间通过类型嵌入，将已经实现的功能嵌入新类型中，以快速满足新类型的功能需求。在通过新类型实例调用方法时，方法的匹配取决于方法名字，而不是类型。

另外，通过在接口的定义中嵌入接口类型来实现接口行为的聚合，组成大接口，这种方式在标准库中尤为常用，并且已经成为Go语言的一种惯用法。

这是Go语言的一个创新设计：接口只是方法集合，且与实现者之间的关系是隐式的，如此可让程序各个部分之间的耦合降至最低。

3、并发和轻量

Go语言的三位设计者Rob Pike、Robert Griesemer和Ken Thompson曾认为C++标准委员会在思路上是短视的，因为硬件很可能在未来十年内发生重大变化，将语言与当时的硬件紧密耦合起来是十分不明智的，是没法给开发人员在编写大规模并发程序时带去太多帮助的。

因而他们把将面向多核、原生内置并发支持作为新语言的设计原则之一。

Go语言原生支持并发的设计哲学体现在下面两点。

（1）Go语言采用轻量级协程并发模型，使得Go应用在面向多核硬件时更具可扩展性。

（2）Go语言为开发者提供的支持并发的语法元素和机制。

4、面向工程

Go语言的设计者在Go语言最初设计阶段，就将解决工程问题作为Go的设计原则之一，进而考虑Go语法、工具链与标准库的设计，这也是Go与那些偏学院派、偏研究性编程语言在设计思路上的一个重大差异。

这让Go语言的规范足够简单灵活，有其他语言基础的程序员都能迅速上手。更重要的是Go自带完善的工具链，大大提高了团队协作的一致性。比如Gofmt自动排版Go代码，很大程度上杜绝了不同人写的代码排版风格不一致的问题。把编辑器配置成在编辑存档的时候自动运行Gofmt，这样在编写代码的时候可以随意摆放位置，存档的时候自动变成正确排版的代码。此外还有Gofix，Govet等非常有用的工具。

总之，Go在语言层面的简单让Go收获了不逊于C++/Java等的表现力的同时，还获得了更好的可读性、更高的开发效率等在软件工程领域更为重要的元素。

3 凹语言™的未来

虽然今天，Go凭借其优越的性能，已经成为主流编程语言之一（超过75%的CNCF项目，包括Kubernetes和Istio，都是用Go编写的，另外，Go也是主要的云应用程序语言之一），Go语言在中国也相当受欢迎，但我们还是不禁担心脱胎于Go的凹语言™，会有美好的未来吗？

Go语言搜索热度，图片来源@Google Trend

预测未来从来都是困难的，不过，好在凹语言™的前面有一个先行者——Go+语言，我们不妨基于Go+的发展，来大致推测凹语言™的未来。

Go+是七牛云CEO许式伟发明的编程语言，于2020年7月正式发布，2021年10月推出1.0版本，目前最新发布版本是今年6月13日发布的1.1版本。也就是说，从正式发布到现在，经过近两年的时间，Go+还处于初始阶段，距离大规模应用还有一定距离，那么可以预见，凹语言™在未来相当长的时间里，不会进入广大开发者的视野中。

另外，据ECUG Con 2022大会上许式伟发表的看法，虽然大家都比较看重编程语言的性能，但单从性能来看的话，许式伟认为Python在脚本语言里面只能算二流，Python其实并不快。

在许式伟看来，对新生的语言来说，最重要它选择的目标人群。

Go+选择的目标人群是全民，许式伟称其为“连儿童也能掌握的语言”，因而Go+从工程与STEM教育的一体化开始奠定用户基础。

正是Go+的这几个特性，让一部分开发者看好Go+的未来。而对Go+的正向预期，会成为Go+进一步发展的助力。

对凹语言™来说，这个道理也是适用的：凹语言™的发展重点可能不在于性能，而在于其选择哪些人群作为目标受众，以及通过何种方式获得种子用户。

如果日后凹语言™的项目方会公布这些消息，那么凹语言™的未来还是可以期待的。

来源：武穆、信远（51CTO技术栈）

最近发现网易号有盗掘金文章的，xdm有空可以关注一下这个问题，希望帮助到大家同时能够保障自己权益。

前言

本文记录一下我是如何使用Gateway搭建网关服务及实现动态路由的，帮助大家学习如何快速搭建一个网关服务，了解路由相关配置，鉴权的流程及业务处理，有兴趣的一定看到最后，非常适合没接触过网关服务的同学当作入门教程。

搭建服务

框架

• SpringBoot 2.1

<parent>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-parent</artifactId>
   <version>2.1.0.RELEASE</version>
</parent>

• Spring-cloud-gateway-core

<dependency>
   <groupId>org.springframework.cloud</groupId>
   <artifactId>spring-cloud-gateway-core</artifactId>
</dependency>

• common-lang3

<dependency>
   <groupId>org.apache.commons</groupId>
   <artifactId>commons-lang3</artifactId>
</dependency>

路由配置

网关作为请求统一入口，路由就相当于是每个业务系统的入口，通过路由规则则可以匹配到对应微服务的入口，将请求命中到对应的业务系统中

server:
  port: 8080

spring:
  cloud:
    gateway:
      enabled: true
      routes:
      - id: demo-server
        uri: http://localhost:8081
        predicates:
        - Path=/demo-server/**
        filters:
          - StripPrefix= 1

routes

解读配置

• 现在有一个服务demo-server部署在本机，地址和端口为127.0.0.1:8081，所以路由配置uri为http://localhost:8081

• 使用网关服务路由到此服务，predicates -Path=/demo-server/**,网关服务的端口为8080，启动网关服务，访问localhost:8080/demo-server,路由断言就会将请求路由到demo-server

• 直接访问demo-server的接口localhost:8081/api/test,通过网关的访问地址则为localhost:8080/demo-server/api/test,predicates配置将请求断言到此路由，filters-StripPrefix=1代表将地址中/后的第一个截取，所以demo-server就截取掉了

使用gateway通过配置文件即可完成路由的配置，非常方便，我们只要充分的了解配置项的含义及规则就可以了;但是这些配置如果要修改则需要重启服务，重启网关服务会导致整个系统不可用，这一点是无法接受的，下面介绍如何通过Nacos实现动态路由

动态路由

使用nacos结合gateway-server实现动态路由，我们需要先部署一个nacos服务，可以使用docker部署或下载源码在本地启动,具体操作可以参考官方文档即可

Nacos配置

groupId: 使用网关服务名称即可

dataId: routes

配置格式： json

[{
      "id": "xxx-server",
      "order": 1, #优先级
      "predicates": [{ #路由断言
          "args": {
              "pattern": "/xxx-server/**"
          },
          "name": "Path"
      }],
      "filters":[{ #过滤规则
          "args": {
              "parts": 0 #k8s服务内部访问容器为http://xxx-server/xxx-server的话,配置0即可
          },
          "name": "StripPrefix" #截取的开始索引
      }],
      "uri": "http://localhost:8080/xxx-server" #目标地址
  }]

json格式配置项与yaml中对应，需要了解配置在json中的写法

比对一下json配置与yaml配置

{
    "id":"demo-server",
    "predicates":[
        {
            "args":{
                "pattern":"/demo-server/**"
            },
            "name":"Path"
        }
    ],
    "filters":[
        {
            "args":{
                "parts":1
            },
            "name":"StripPrefix"
        }
    ],
    "uri":"http://localhost:8081"
}
spring:
  cloud:
    gateway:
      enabled: true
      routes:
      - id: demo-server
        uri: http://localhost:8081
        predicates:
        - Path=/demo-server/**
        filters:
          - StripPrefix= 1

代码实现

Nacos实现动态路由的方式核心就是通过Nacos配置监听，配置发生改变后执行网关相关api创建路由

@Component
public class NacosDynamicRouteService implements ApplicationEventPublisherAware {

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

    @Autowired
    private RouteDefinitionWriter routeDefinitionWriter;

    private ApplicationEventPublisher applicationEventPublisher;

    /** 路由id */
    private static List<String> routeIds = Lists.newArrayList();

    /**
     * 监听nacos路由配置，动态改变路由
     * @param configInfo
     */
    @NacosConfigListener(dataId = "routes", groupId = "gateway-server")
    public void routeConfigListener(String configInfo) {
        clearRoute();
        try {
            List<RouteDefinition> gatewayRouteDefinitions = JSON.parseArray(configInfo, RouteDefinition.class);
            for (RouteDefinition routeDefinition : gatewayRouteDefinitions) {
                addRoute(routeDefinition);
            }
            publish();
            LOGGER.info("Dynamic Routing Publish Success");
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
        
    }


    /**
     * 清空路由
     */
    private void clearRoute() {
        for (String id : routeIds) {
            routeDefinitionWriter.delete(Mono.just(id)).subscribe();
        }
        routeIds.clear();
    }

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        this.applicationEventPublisher = applicationEventPublisher;
    }

    /**
     * 添加路由
     * 
     * @param definition
     */
    private void addRoute(RouteDefinition definition) {
        try {
            routeDefinitionWriter.save(Mono.just(definition)).subscribe();
            routeIds.add(definition.getId());
        } catch (Exception e) {
            LOGGER.error(e.getMessage(), e);
        }
    }

    /**
     * 发布路由、使路由生效
     */
    private void publish() {
        this.applicationEventPublisher.publishEvent(new RefreshRoutesEvent(this.routeDefinitionWriter));
    }
}

过滤器

gateway提供GlobalFilter及Ordered两个接口用来定义过滤器，我们自定义过滤器只需要实现这个两个接口即可

• GlobalFilter filter() 实现过滤器业务

• Ordered getOrder() 定义过滤器执行顺序

通常一个网关服务的过滤主要包含 鉴权（是否登录、是否黑名单、是否免登录接口...） 限流（ip限流等等）功能，我们今天简单介绍鉴权过滤器的流程实现

鉴权过滤器

需要实现鉴权过滤器，我们先得了解登录及鉴权流程，如下图所示

由图可知，我们鉴权过滤核心就是验证token是否有效，所以我们网关服务需要与业务系统在同一个redis库，先给网关添加redis依赖及配置

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>
spring:
  redis:
    host: redis-server
    port: 6379
    password:
    database: 0

代码实现

• 1.定义过滤器AuthFilter

• 2.获取请求对象 从请求头或参数或cookie中获取token(支持多种方式传token对于客户端更加友好，比如部分web下载请求会新建一个页面，在请求头中传token处理起来比较麻烦)

• 3.没有token,返回401

• 4.有token，查询redis是否有效

• 5.无效则返回401，有效则完成验证放行

• 6.重置token过期时间、添加内部请求头信息方便业务系统权限处理

@Component

public class AuthFilter implements GlobalFilter, Ordered {



    @Autowired

    private RedisTemplate<String, String> redisTemplate;



    private static final String TOKEN_HEADER_KEY = "auth_token";



    @Override

    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        // 1.获取请求对象

        ServerHttpRequest request = exchange.getRequest();

        // 2.获取token

        String token = getToken(request);

        ServerHttpResponse response = exchange.getResponse();

        if (StringUtils.isBlank(token)) {

            // 3.token为空 返回401

            response.setStatusCode(HttpStatus.UNAUTHORIZED);

            return response.setComplete();

        }

        // 4.验证token是否有效

        String userId = getUserIdByToken(token);

        if (StringUtils.isBlank(userId)) {

            // 5.token无效 返回401

            response.setStatusCode(HttpStatus.UNAUTHORIZED);

            return response.setComplete();

        }

        // token有效，后续业务处理

        // 从写请求头，方便业务系统从请求头获取用户id进行权限相关处理

        ServerHttpRequest.Builder builder = exchange.getRequest().mutate();

        request = builder.header("user_id", userId).build();

        // 延长缓存过期时间-token缓存用户如果一直在操作就会一直重置过期

        // 这样避免用户操作过程中突然过期影响业务操作及体验，只有用户操作间隔时间大于缓存过期时间才会过期

        resetTokenExpirationTime(token, userId);

        // 完成验证

        return chain.filter(exchange);

    }





    @Override

    public int getOrder() {

        // 优先级 越小越优先

        return 0;

    }



    /**

     * 从redis中获取用户id

     * 在登录操作时候 登陆成功会生成一个token, redis得key为auth_token:token 值为用户id

     *

     * @param token

     * @return

     */

    private String getUserIdByToken(String token) {

        String redisKey = String.join(":", "auth_token", token);

        return redisTemplate.opsForValue().get(redisKey);

    }



    /**

     * 重置token过期时间

     *

     * @param token

     * @param userId

     */

    private void resetTokenExpirationTime(String token, String userId) {

        String redisKey = String.join(":", "auth_token", token);

        redisTemplate.opsForValue().set(redisKey, userId, 2, TimeUnit.HOURS);

    }





    /**

     * 获取token

     *

     * @param request

     * @return

     */

    private static String getToken(ServerHttpRequest request) {

        HttpHeaders headers = request.getHeaders();

        // 从请求头获取token

        String token = headers.getFirst(TOKEN_HEADER_KEY);

        if (StringUtils.isBlank(token)) {

            // 请求头无token则从url获取token

            token = request.getQueryParams().getFirst(TOKEN_HEADER_KEY);

        }

        if (StringUtils.isBlank(token)) {

            // 请求头和url都没有token则从cookies获取

            HttpCookie cookie = request.getCookies().getFirst(TOKEN_HEADER_KEY);

            if (cookie != null) {

                token = cookie.getValue();

            }

        }

        return token;

    }

}

总结

Gateway通过配置项可以实现路由功能，整合Nacos及配置监听可以实现动态路由，实现GlobalFilter, Ordered两个接口可以快速实现一个过滤器，文中也详细的介绍了登录后的请求鉴权流程，如果有不清楚地方可以评论区见咯。

来源：juejin.cn/post/7004756545741258765

这篇文章的前提是，你得有个女朋友，没有就先收藏着吧！

七夕节的来源是梁山伯与祝英台的美丽传说，化成了一对蝴蝶~ 美丽的神话！虽然现在一般是过214的情人节了，但是不得不说，古老的传统的文化遗产，还是要继承啊~

在互联网公司中，主要的程序员品种包括：前端工程师，后端工程师，算法工程师。

对于具体的职业职能划分还不是很清楚的，我们简单的介绍一下不同程序员岗位的职责：

前端程序员：绘制UI界面，与设计和产品经理进行需求的对接，绘制特定的前端界面推向用户

后端程序员：接收前端json字符串，与数据库对接，将json推向前端进行显示

算法工程师：进行特定的规则映射，优化函数的算法模型，改进提高映射准确率。

七夕节到了，怎么结合自身的的专业技能，哄女朋友开心呢？

前端工程师：我先来，画个动态的晚霞页面！

1.定义样式风格：

.star {
  width: 2px;
  height: 2px;
  background: #f7f7b6;
  position: absolute;
  left: 0;
  top: 0;
  backface-visibility: hidden;
}

2.定义动画特性

@keyframes rotate {
  0% {
    transform: perspective(400px) rotateZ(20deg) rotateX(-40deg) rotateY(0);
  }

  100% {
    transform: perspective(400px) rotateZ(20deg) rotateX(-40deg) rotateY(-360deg);
  }
}

3.定义星空样式数据

export default {
  data() {
    return {
      starsCount: 800, //星星数量
      distance: 900, //间距
    }
  }
}

4.定义星星运行速度与规则：

starNodes.forEach((item) => {
      let speed = 0.2 + Math.random() * 1;
      let thisDistance = this.distance + Math.random() * 300;
      item.style.transformOrigin = `0 0 ${thisDistance}px`;
      item.style.transform =
          `
                translate3d(0,0,-${thisDistance}px)
                rotateY(${Math.random() * 360}deg)
                rotateX(${Math.random() * -50}deg)
                scale(${speed},${speed})`;
    });

前端预览效果图：

后端工程师看后，先点了点头，然后表示不服，画页面太肤浅了，我开发一个接口，定时在女朋友生日的时候发送祝福邮件吧！

1.导入pom.xml 文件

        <!-- mail邮件服务启动器 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-mail</artifactId>
        </dependency>

2.application-dev.properties内部增加配置链接

#QQ\u90AE\u7BB1\u90AE\u4EF6\u53D1\u9001\u670D\u52A1\u914D\u7F6E
spring.mail.host=smtp.qq.com
spring.mail.port=587

## qq邮箱
spring.mail.username=#yourname#@qq.com
## 这里填邮箱的授权码
spring.mail.password=#yourpassword#

3.配置邮件发送工具类 MailUtils.java

@Component
public class MailUtils {
    @Autowired
    private JavaMailSenderImpl mailSender;
    
    @Value("${spring.mail.username}")
    private String mailfrom;

    // 发送简单邮件
    public void sendSimpleEmail(String mailto, String title, String content) {
        //  定制邮件发送内容
        SimpleMailMessage message = new SimpleMailMessage();
        message.setFrom(mailfrom);
        message.setTo(mailto);
        message.setSubject(title);
        message.setText(content);
        // 发送邮件
        mailSender.send(message);
    }
}

4.测试使用定时注解进行注释

@Component
class DemoApplicationTests {

    @Autowired
    private MailUtils mailUtils;

    /**
     * 定时邮件发送任务，每月1日中午12点整发送邮件
     */
    @Scheduled(cron = "0 0 12 1 * ?")
    void sendmail(){
        //  定制邮件内容
        StringBuffer content = new StringBuffer();
        content.append("HelloWorld");
        //分别是接收者邮箱，标题，内容
        mailUtils.sendSimpleEmail("123456789@qq.com","自定义标题",content.toString());
    }
}

@scheduled注解 使用方法： cron：秒，分，时，天，月，年，* 号表示 所有的时间均匹配

5.工程进行打包，部署在服务器的容器中运行即可。

算法工程师，又开发接口，又画页面，我就训练一个自动写诗机器人把！

1.定义神经网络RNN结构

def neural_network(model = 'gru', rnn_size = 128, num_layers = 2):
    cell = tf.contrib.rnn.BasicRNNCell(rnn_size, state_is_tuple = True)
    cell = tf.contrib.rnn.MultiRNNCell([cell] * num_layers, state_is_tuple = True)
    initial_state = cell.zero_state(batch_size, tf.float32)
    with tf.variable_scope('rnnlm'):
        softmax_w = tf.get_variable("softmax_w", [rnn_size, len(words)])
        softmax_b = tf.get_variable("softmax_b", [len(words)])
        embedding = tf.get_variable("embedding", [len(words), rnn_size])
        inputs = tf.nn.embedding_lookup(embedding, input_data)
    outputs, last_state = tf.nn.dynamic_rnn(cell, inputs, initial_state = initial_state, scope = 'rnnlm')
    output = tf.reshape(outputs, [-1, rnn_size])
    logits = tf.matmul(output, softmax_w) + softmax_b
    probs = tf.nn.softmax(logits)
    return logits, last_state, probs, cell, initial_state

2.定义模型训练方法：

def train_neural_network():
    logits, last_state, _, _, _ = neural_network()
    targets = tf.reshape(output_targets, [-1])
    loss = tf.contrib.legacy_seq2seq.sequence_loss_by_example([logits], [targets], \
        [tf.ones_like(targets, dtype = tf.float32)], len(words))
    cost = tf.reduce_mean(loss)
    learning_rate = tf.Variable(0.0, trainable = False)
    tvars = tf.trainable_variables()
    grads, _ = tf.clip_by_global_norm(tf.gradients(cost, tvars), 5)
    #optimizer = tf.train.GradientDescentOptimizer(learning_rate)
    optimizer = tf.train.AdamOptimizer(learning_rate)
    train_op = optimizer.apply_gradients(zip(grads, tvars))

    Session_config = tf.ConfigProto(allow_soft_placement = True)
    Session_config.gpu_options.allow_growth = True

    trainds = DataSet(len(poetrys_vector))

    with tf.Session(config = Session_config) as sess:
        sess.run(tf.global_variables_initializer())

        saver = tf.train.Saver(tf.global_variables())
        last_epoch = load_model(sess, saver, 'model/')

        for epoch in range(last_epoch + 1, 100):
            sess.run(tf.assign(learning_rate, 0.002 * (0.97 ** epoch)))
            #sess.run(tf.assign(learning_rate, 0.01))

            all_loss = 0.0
            for batche in range(n_chunk):
                x,y = trainds.next_batch(batch_size)
                train_loss, _, _ = sess.run([cost, last_state, train_op], feed_dict={input_data: x, output_targets: y})

                all_loss = all_loss + train_loss

                if batche % 50 == 1:
                    print(epoch, batche, 0.002 * (0.97 ** epoch),train_loss)

            saver.save(sess, 'model/poetry.module', global_step = epoch)
            print (epoch,' Loss: ', all_loss * 1.0 / n_chunk)

3.数据集预处理

poetry_file ='data/poetry.txt'
# 诗集
poetrys = []
with open(poetry_file, "r", encoding = 'utf-8') as f:
    for line in f:
        try:
            #line = line.decode('UTF-8')
            line = line.strip(u'\n')
            title, content = line.strip(u' ').split(u':')
            content = content.replace(u' ',u'')
            if u'_' in content or u'(' in content or u'（' in content or u'《' in content or u'[' in content:
                continue
            if len(content) < 5 or len(content) > 79:
                continue
            content = u'[' + content + u']'
            poetrys.append(content)
        except Exception as e:
            pass

poetry.txt文件中存放这唐诗的数据集，用来训练模型

4.测试一下训练后的模型效果：

藏头诗创作：“七夕快乐”

模型运算的结果

哈哈哈，各种节日都是程序员的表（zhuang）演（bi) 时间，不过这些都是锦上添花，只有实实在在，真心，才会天长地久啊~

提前祝各位情侣七夕节快乐！

作者：千与编程
来源：juejin.cn/post/6995491512716918814

一段时间以来，人们都“知道”，作为移动平台的 Android 不如 iOS 安全，这已经成为常识。似乎除了消费者，每个人都知道。2021 年 8 月对 10000 名移动消费者进行的一项全球调查发现，iOS 和 Android 用户的安全期望基本一致。

然而，尽管消费者有这样的期望，而且从本质上讲，一个移动平台并不一定比另一个平台更安全，但移动应用很少能实现 Android 和 iOS 的安全功能对等。事实上，许多移动应用甚至缺少最基本的安全保护措施。让我们看看这是为什么。

1

移动应用安全需要多层次防御

大多数安全专家和第三方标准组织都会同意，移动应用安全需要多层次防御，包括以下核心领域的多种安全特性：

• 代码混淆和应用护盾（Application Shielding）：保护移动应用的二进制文件和源代码，防止逆向工程。

• 数据加密：保护应用中存储和使用的数据。

• 安全通信：保护在应用和应用后端之间传递的数据，包括确保用于建立可信连接的数字证书的真实性和有效性。

• 操作系统防护：保护应用免受未经授权的操作系统修改（如 rooting 和越狱）所影响。

开发人员应该在应用的 iOS 和 Android 版本中均衡地实现这些功能的组合，形成一致的安全防御。而且，他们应该在开发周期的早期添加这些功能——这个概念被称为安全“左移”。听起来很容易吧？理论上，是的，但在实践中，如果使用“传统”方法，要实现移动应用多层次安全防御实际上是相当困难的。

多年来，移动开发人员一直试图使用传统的工具集来实现应用内移动应用安全，包括第三方开源库、商业移动应用安全 SDK 或专用编译器。第一个主要的挑战是，移动应用的安全从来无法通过“银弹”实现。由于移动应用在不受保护的环境中运行，并存储和处理大量有价值的信息，有许多方法可以攻击它们。黑客有无穷无尽的、免费提供而又非常强大的工具集可以使用，而且可以全天候地研究和攻击应用而不被发现。

2

移动安全要求

因此，为了构建一个强大的防御体系，移动开发人员需要实施一个既“广”且“深”的多层次防御。所谓“广”，我指的是不同保护类别的多种安全特性，它们彼此相互补充，如加密和混淆。所谓“深”，我指的是每个安全特性都应该有多种检测或保护方法。例如，一个越狱检测 SDK 如果只在应用启动时进行检查，就不会很有效，因为攻击者很容易绕过。

或者考虑下反调试，这是一种重要的运行时防御，可以防止攻击者使用调试器来执行动态分析——他们会在一个受控的环境中运行应用，为的是了解或修改应用的行为。有许多类型的调试器——有一些基于 LLDB——是针对原生代码的，如 C++ 或 Objective C，其他的在 Java 或 Kotlin 层进行检查，诸如此类。每个调试器连接和分析应用的工作方式都略有不同。因此，为了使反调试防御奏效，应用需要识别正在使用的多种调试方法，并动态地进行恰当的防御，因为黑客会继续尝试不同的调试工具或方法，直到他们找到一个可以成功的。

3

防篡改

安全要求清单并不仅限于此。每个应用都需要防篡改功能，如校验和验证、预防二进制补丁，以及应用的重新打包、重新签名、模拟器和仿真器，等等。毫不夸张，仅是针对一个操作系统研究和实现这些功能或保护方法中的一项，就需要至少几个人周的开发时间。而且还要有一个前提，就是移动开发人员已经拥有特定安全领域的专业知识，但情况往往并非如此。复杂度可能会快速增加，到目前为止，我们只讨论了一个保护类别——运行时或动态保护。想象一下，如果提到的每个功能都需要一到两周的开发时间，那么实现全部安全特性得付出多大的时间成本。

4

防越狱 /Rooting

接下来，你还需要操作系统层面的保护，如防越狱 /rooting，在移动操作系统遭破坏的情况下保护应用。越狱 /rooting 使移动应用容易受到攻击，因为它允许对操作系统和文件系统进行完全的管理控制，破坏了整个安全模型。而且，仅仅检测越狱 /rooting 已经不够了，因为黑客们一直在不断地改进他们的工具。要说最先进的越狱和 rooting 工具，在 iOS 上是 Checkra1n，在 Android 上是 Magisk——还有许多其他的工具。其中，还有一些工具用于隐藏或掩盖活动及管理超级用户权限——通常授予恶意应用。朋友们，如果你使用 SDK 或第三方库实现了越狱或 rooting 检测，那么你的保护措施很有可能已经过时或者很容易被绕过，尤其是在没有对应用的源代码进行充分混淆的情况下。

5

代码混淆

如果你使用 SDK 或第三方库来实现安全防护，那在未混淆的应用中几乎没什么用——为什么？因为黑客使用 Hopper、IDA-pro 等开源工具，就可以很容易地反编译或反汇编，找到 SDK 的源代码，或使用类似 Frida 这样的动态二进制工具箱，注入他们自己的恶意代码，修改应用的行为，或简单地禁用安全 SDK。

代码混淆可以防止攻击者了解移动应用的源代码。而且，我们总是建议使用多种混淆方法，包括混淆本地代码或非本地代码和库，以及混淆应用的逻辑结构或控制流。例如，可以使用控制流混淆或重命名函数、类、方法、变量等来实现。不要忘了还要混淆调试信息。

从现实世界的数据中可以看出，大多数移动应用都缺乏足够的混淆，只混淆了应用的一小部分代码，这项对超过 100 万个 Android 应用的研究清楚地说明了这一点。正如该研究指出的那样，造成这种情况的原因是，对于大多数移动开发人员来说，依赖专用编译器的传统混淆方法实在是太复杂和费时，难以全面实施。相反，许多开发人员只实现了单一的混淆功能，或者只混淆了代码库的一小部分。在这项研究中，研究人员发现，大多数应用只实现了类名混淆，这本身很容易被攻陷。拿书打个比方，类名混淆本身就像是混淆了一本书的“目录”，但书中所有实际的页和内容却并没有混淆。这种表面的混淆相当容易被绕过。

6

数据保护和加密

接着说数据保护，你还需要借助加密来保护应用和用户数据——在移动应用中，有很多地方存储着数据，包括沙盒、内存以及应用的代码或字符串。要自己实现加密，有很多棘手的问题需要解决，包括密钥衍生（key derivation）、密码套件和加密算法组合、密钥大小及强度。许多应用使用了多种编程语言，每一种都需要不同的 SDK，或者会导致你无法控制的不兼容性，又或是需要你无法访问的依赖。而数据类型的差异也有复杂性增加和性能下降的风险。

然后，还有一个典型的问题，即在哪里存储加密密钥。如果密钥存储在应用内部，那它们可能会被反向工程的攻击者发现，然后他们就可以用来解密数据。这就是为什么我们说动态密钥生成是一个非常重要的功能。通过动态密钥生成，加密密钥只在运行时生成，而不会存储在应用或移动设备上。此外，密钥只使用一次，可以防止攻击者发现或截获它们。

那么传输中的数据呢？仅靠 TLS 是不够的，因为有很多方法可以侵入应用的连接。检查和验证 TLS 会话和证书很重要，这可以确保所有的证书和 CA 都是有效且真实的，受到行业标准加密的保护。这可以防止黑客获得 TLS 会话的控制权。然后还有证书固定，可以防止连接到遭到入侵的服务器，或保护服务器，拒绝遭到入侵的应用连接（例如，如果你的应用被变成了一个恶意机器人）。

7

欺诈、恶意软件、防盗版

最后，还有反欺诈、反恶意软件和反盗版保护，你可以在上述基线保护的基础上增加防护层，用于防止非常高级或专门的威胁。这些保护措施可能包括可以防止应用覆盖攻击、自动点击器、钩子框架和动态二进制工具、内存注入、键盘记录器、密钥注入或可访问性滥用的功能，所有这些都是移动欺诈或移动恶意软件的常用武器。

不难想象，即使是实现上述功能的一个子集，也需要大量的时间和资源。到目前为止，我只是谈了一个强大的安全防御所需的特性和功能。即使你内部有资源和所需的技能组合，那么拼凑出一个防御体系的行动挑战又是什么呢？让我们探讨一下开发团队可能会遇到的一些实施挑战。

8

不同平台和框架之间的实现差异

鉴于用于构建移动应用的 SDK/ 库及原生或非原生编程语言之间存在无数的框架差异和不兼容，开发人员将面临的下一个问题是如何分别为 Android 和 iOS 实现这些安全功能。虽然软件开发工具包（SDK）提供了一些标准安全功能，但没有 SDK 能普遍覆盖所有的平台或框架。

当开发人员试图使用 SDK 或开源库来实现移动应用安全时，所面临的一个主要挑战在于，这些方法都依赖于源代码，需要对应用代码进行修改。而结果是，这些方法中的每一个都明确地与应用所使用的特定编程语言绑定，并且还暴露给了各种编程语言或是这些语言和框架的包“依赖”。

通常，iOS 应用使用 Objective-C 或 Swift 构建，而 Android 应用使用 Java 或 Kotlin 以及使用 C 和 C++ 编写原生库。例如，假如你想对存储在 Android 和 iOS 应用中的数据进行加密。如果你找到了一些第三方 Android 加密库亦或是 Java 或 Kotlin 的 SDK，它们不一定适用于应用中使用的 C 或 C++ 代码部分（原生库）。

在 iOS 中也是如此。你浏览 StackOverflow 时可能会发现，在 Swift 中常用的 Cryptokit 框架对 Objective C 不起作用。

那么，非原生或跨平台应用呢？它们是完全不同的赛道，因为你要处理的是 JavaScript 等 Web 技术和 React Native、Cordova、Flutter 或 Xamarin 等非原生框架，它们无法直接（或根本不能）使用为原生语言构建的 SDK 或库。此外，对于非原生应用，你可能无法获得相关的源代码文件，从源头实现加密。

关于这个问题，有一个真实的例子，请看 Stack Overflow 上的这篇帖子。开发人员需要在一个 iOS 应用中实现代码混淆，其中 React Native（一个非原生框架）和 Objective C（一种原生编码语言）之间存在多个依赖关系。由于 iOS 项目中没有可以混淆 React Native 代码的内置库，开发人员需要使用一个外部包（依赖关系 #1）。此外，该外部包还依赖下游的一个库或包来混淆 JavaScript 代码（依赖关系 #2）。现在，如果第三方库的开发人员决定废弃该解决方案，会发生什么？我们的一个客户就面临着这样的问题，这导致他们的应用不符合 PCI 标准。

那么，你认为需要多少开发人员来实现我刚才描述的哪怕是一小部分功能？又需要多长时间？你有足够的时间在现有的移动应用发布过程中实现所需的安全功能吗？

9

DevOps 是敏捷 + 自动化，传统安全是单体 + 手动

移动应用是在一个快节奏、灵活且高度自动化的敏捷模式下开发和发布的。为了使构建和发布更快速、更简单，大多数 Android 和 iOS DevOps 团队都围绕 CI/CD 和其他自动化工具构建了最佳管道。另一方面，安全团队无法访问或查看 DevOps 系统，而且大多数安全工具并不是针对敏捷方法构建的，因为它们在很大程度上依赖于手动编程或实施，在这种情况下，单个安全功能的实施时间可能会长于发布时间表允许的时间。

为了弥补这些不足，一些组织在向公共应用商店发布应用之前，会使用代码扫描和渗透测试，以深入探查漏洞和其他移动应用问题。当发现漏洞时，企业就会面临一个艰难的决定：是在未进行必要保护的情况下发布应用，还是推迟发布，让开发人员有时间来解决安全问题。当这种情况发生时，推荐的安全保护措施往往会被忽视。

开发人员并不懒惰，而是他们用于实现安全保护的系统和工具根本无法匹配现代敏捷 /DevOps 开发的快节奏。

10

实现强大的移动应用安全和平台对等的五个步骤

一般来说，自动化是实现安全对等和强大的移动应用安全的关键所在。以下是在应用发布周期内将移动应用安全打造为应用组成部分的五个步骤。

开发、运营和安全团队必须就移动安全预期达成一致。对于组织作为起点的安全目标，人们要有一个共同的理解，如 OWASP Mobile Top 10、TRM 移动应用安全指南和移动应用安全验证标准（MASVS）。一旦确定了目标并选择了标准，所有团队成员都要知道这对他们的工作流有何影响。

安全非常复杂，手动编码很慢，而且容易出错。评估并利用自动化系统，借助人工智能和机器学习（ML）将安全集成到移动应用中。通常情况下，这些都是无代码平台，可以自动将安全构建到移动应用中，它们通常被称为安全构建系统。

移动应用安全模型左移是指，移动开发人员需要在构建应用的同时构建安全特性。

一旦选择了自动化安全实施平台，就应该将其整合到团队的持续集成（CI）和持续交付（CD）流程中，这可以加速开发生命周期，所有团队——开发、运营和安全——在整个冲刺期间都应该保持密切合作。此外，企业可以为每个 Android 和 iOS 应用所需的特定安全特性创建可重复使用的移动安全模板，从而更接近实现平台对等。

如果没有办法即时验证所需的安全功能是否包含在发布中，那么在发布会议上就会出现争执，可能导致应用发布或更新延期。验证和确认应该自动记录，防止最后一刻的发布混乱。

开发团队需要可预测性和明确的预算。通过采用自动化的安全方法，应用开发团队可以减少人员和开发费用的意外变化，因为它消除了手动将安全编码到移动应用时固有的不确定性。

11

小结

安全对等问题是一个大问题，但它是一个更大问题的一部分，即移动应用普遍缺乏安全性。通过在安全实现中采用与特性开发相同或更高程度的自动化，开发人员可以确保他们针对每个平台发布的每一个应用都免受黑客、骗子和网络犯罪分子的侵害。

作者简介：

Alan Bavosa 是 Appdome 的安全产品副总裁。长期以来，他一直担任安全产品执行官，曾是 Palerra（被 Oracle 收购）和 Arcsight（被 HP 收购）的产品主管。

原文链接：

https://www.infoq.com/articles/secure-mobile-apps-parity-problem/

max_allowed_packet 表示 MySQL Server 或者客户端接收的 packet 的最大大小，packet 即数据包，MySQL Server 和客户端上都有这个限制。

数据包

每个数据包，都由包头、包体两部分组成，包头由 3 字节的包体长度、1 字节的包编号组成。3 字节最多能够表示 2 ^ 24 = 16777216 字节（16 M），也就是说，一个数据包的包体长度必须小于等于 16M 。

如果要发送超过 16M 的数据怎么办？

当要发送大于 16M 的数据时，会把数据拆分成多个 16M 的数据包，除最后一个数据包之外，其它数据包大小都是 16M。而 MySQL Server 收到这样的包后，如果发现包体长度等于 16M ，它就知道本次接收的数据由多个数据包组成，会先把当前数据包的内容写入缓冲区，然后接着读取下一个数据包，并把下一个数据包的内容追加到缓冲区，直到读到结束数据包，就接收到客户端发送的完整数据了。

那怎样算一个数据包？

• 一个 SQL 是一个数据包

• 返回查询结果时，一行数据算一个数据包

• 解析的 binlog ，如果用 mysql 客户端导入，一个 SQL 算一个数据包

• 在复制中，一个 event 算一个数据包

下面我们通过测试来讨论 max_allowed_packet 的实际影响。

导入 SQL 文件受 max_allowed_packet 限制吗？

如果 SQL 文件中有单个 SQL 大小超过 max_allowed_packet ，会报错：

##导出时设置 mysqldump --net-buffer-length=16M，这样保证导出的sql文件中单个 multiple-row INSERT 大小为 16M
mysqldump -h127.0.0.1 -P13306 -uroot -proot --net-buffer-length=16M \
--set-gtid-purged=off sbtest sbtest1 > /data/backup/sbtest1.sql

##设置max_allowed_packet=1M

##导入报错
[root@localhost data]# mysql -h127.0.0.1 -P13306 -uroot -proot db3 < /data/backup/sbtest1.sql
mysql: [Warning] Using a password on the command line interface can be insecure.
ERROR 1153 (08S01) at line 41: Got a packet bigger than 'max_allowed_packet' bytes

导入解析后的 binlog 受 max_allowed_packet 限制吗？

row 格式的 binlog，单个SQL修改的数据产生的 binlog 如果超过 max_allowed_packet，也会报错。

在恢复数据到指定时间点的场景，解析后的binlog单个事务大小超过1G，并且这个事务只包含一个SQL，此时一定会触发 max_allowed_packet 的报错。但是恢复数据的任务又很重要，怎么办呢？可以将 binlog 改名成 relay log，用 sql 线程回放来绕过这个限制。

查询结果受 max_allowed_packet 限制吗？

查询结果中，只要单行数据不超过客户端设置的 max_allowed_packet 即可：

##插入2行20M大小的数据
[root@localhost tmp]# dd if=/dev/zero of=20m.img bs=1 count=0 seek=20M
记录了0+0 的读入
记录了0+0 的写出
0字节(0 B)已复制，0.000219914 秒，0.0 kB/秒
[root@localhost tmp]# ll -h 20m.img
-rw-r--r-- 1 root root 20M 6月   6 15:15 20m.img

mysql> create table t1(id int auto_increment primary key,a longblob);
Query OK, 0 rows affected (0.03 sec)

mysql> insert into t1 values(NULL,load_file('/tmp/20m.img'));
Query OK, 1 row affected (0.65 sec)

mysql> insert into t1 values(NULL,load_file('/tmp/20m.img'));
Query OK, 1 row affected (0.65 sec)

##mysql客户端默认 --max-allowed-packet=16M,读取失败
mysql> select * from t1;
ERROR 2020 (HY000): Got packet bigger than 'max_allowed_packet' bytes

##设置 mysql 客户端 --max-allowed-packet=22M，读取成功
[root@localhost ~]# mysql -h127.0.0.1 -P13306 -uroot -proot --max-allowed-packet=23068672 sbtest -e "select * from t1;" > /tmp/t1.txt

[root@localhost ~]# ll  -h /tmp/t1.txt
-rw-r--r-- 1 root root 81M 6月   6 15:30 /tmp/t1.txt

load data 文件大小受 max_allowed_packet 限制吗？

load data 文件大小、单行大小都不受 max_allowed_packet 影响：

##将上一个测试中的数据导出，2行数据一共81M
mysql> select * int o outfile '/tmp/t1.csv' from t1;
Query OK, 2 rows affected (0.57 sec)

[root@localhost ~]# ll -h /tmp/t1.csv
-rw-r----- 1 mysql mysql 81M 6月   6 15:32 /tmp/t1.csv

##MySQL Server max_allowed_packet=16M
mysql> select @@max_allowed_packet;
+----------------------+
| @@max_allowed_packet |
+----------------------+
|             16777216 |
+----------------------+
1 row in set (0.00 sec)

##load data 成功,不受 max_allowed_packet 限制
mysql> load data infile '/tmp/t1.csv' into table t1;
Query OK, 2 rows affected (1.10 sec)
Records: 2  Deleted: 0  Skipped: 0  Warnings: 0

binlog 中超过 1G 的 SQL ，是如何突破 max_allowed_packet 复制到从库的？

从库 slave io 线程、slave sql 线程可以处理的最大数据包大小由参数 slave_max_allowed_packet 控制。这是限制 binlog event 大小，而不是单个 SQL 修改数据的大小。

主库 dump 线程会自动设置 max_allowed_packet为1G，不会依赖全局变量 max_allowed_packet。用来控制主库 DUMP 线程每次读取 event 的最大大小。

具体可以参考：mp.weixin.qq.com/s/EfNY_UwEthiu-DEBO7TrsA

另外超过 4G 的大事务，从库心跳会报错：https://opensource.actionsky.com/20201218-mysql/

作者：胡呈清，爱可生 DBA 团队成员，擅长故障分析、性能优化

来源：jianshu.com/u/a95ec11f67a8

背景介绍

从系统设计角度看，一个系统从设计搭建到数据逐步增长，SQL 执行效率可能会出现劣化，为继续支撑业务发展，我们需要对慢 SQL 进行分析和优化，严峻的情况下甚至需要对整个系统进行重构。所以我们往往需要在系统设计前对业务进行充分调研、遵守系统设计规范，在系统运行时定期结合当前业务发展情况进行系统瓶颈的分析。

从数据库角度看，每个 SQL 执行都需要消耗一定 I/O 资源，SQL 执行的快慢，决定了资源被占用时间的长短。假如有一条慢 SQL 占用了 30%的资源共计 1 分钟。那么在这 1 分钟时间内，其他 SQL 能够分配的资源总量就是 70%，如此循环，当资源分配完的时候，所有新的 SQL 执行将会排队等待。所以往往一条慢 SQL 会影响到整个业务。

本文仅讨论 MySQL-InnoDB 的情况。

优化方式

SQL 语句执行效率的主要因素

• 数据量

• SQL 执行后返回给客户端的数据量的大小；

• 数据量越大需要扫描的 I/O 次数越多，数据库服务器的 IO 更容易成为瓶颈。

• 取数据的方式

• 数据在缓存中还是在磁盘上；

• 是否能够通过全局索引快速寻址；

• 是否结合谓词条件命中全局索引加速扫描。

• 数据加工的方式

• 排序、子查询、聚合、关联等，一般需要先把数据取到临时表中，再对数据进行加工；

• 对于数据量比较多的计算，会消耗大量计算节点的 CPU 资源，让数据加工变得更加缓慢；

• 是否选择了合适的 join 方式

优化思路

• 减少数据扫描（减少磁盘访问）

• 尽量在查询中加入一些可以提前过滤数据的谓词条件，比如按照时间过滤数据等，可以减少数据的扫描量，对查询更友好；

• 在扫描大表数据时是否可以命中索引，减少回表代价，避免全表扫描。

• 返回更少数据（减少网络传输或磁盘访问）

• 减少交互次数（减少网络传输）

• 将数据存放在更快的地方

• 某条查询涉及到大表，无法进一步优化，如果返回的数据量不大且变化频率不高但访问频率很高，此时应该考虑将返回的数据放在应用端的缓存当中或者 Redis 这样的缓存当中，以提高存取速度。

• 减少服务器 CPU 开销（减少 CPU 及内存开销）

• 避免大事务操作

• 利用更多资源（增加资源）

优化案例

数据分页优化

sele ct * from table_demo where type = ? limit ?,?;

优化方式一：偏移 id

lastId = 0 or min(id)
do {
sele ct * from table_demo where type = ? and id >{#lastId} limit ?;
lastId = max(id)
} while (isNotEmpty)

优化方式二：分段查询

该方式较方式一的优点在于可并行查询，每个分段查询互不依赖；较方式一的缺点在于较依赖数据的连续性，若数据过于分散，代价较高。

minId = min(id) maxId = max(id)
for(int i = minId; i<= maxId; i+=pageSize){
sele ct * from table_demo where type = ? and id between i and i+ pageSize;
}

优化 GROU P BY

提高 GROU P BY 语句的效率, 可以通过将不需要的记录在 GROU P BY 之前过滤掉.下面两个查询返回相同结果但第二个明显就快了许多。

低效:

sele ct job , avg(sal) from table_demo grou p by job having job = ‘manager'

高效:

sele ct job , avg(sal) from table_demo where job = ‘manager' grou p by job

范围查询

联合索引中如果有某个列存在范围（大于小于）查询，其右边的列是否还有意义？

expla in sele ct count(1) from statement where org_code='1012' and trade_date_time >= '2019-05-01 00:00:00' and trade_date_time<='2020-05-01 00:00:00'
expla in sele ct * from statement where org_code='1012' and trade_date_time >= '2019-05-01 00:00:00' and trade_date_time<='2020-05-01 00:00:00' limit 0, 100
expla in sele ct * from statement where org_code='1012' and trade_date_time >= '2019-05-01 00:00:00' and trade_date_time<='2020-05-01 00:00:00'

• 使用单键索引 trade_date_time 的情况下

• 从索引里找到所有 trade_date_time 在'2019-05-01' 到'2020-05-01' 区间的主键 id。假设有 100 万个。

• 对这些 id 进行排序（为的是在下面一步回表操作中优化 I/O 操作，因为很多挨得近的主键可能一次磁盘 I/O 就都取到了）

• 回表，查出 100 万行记录，然后逐个扫描，筛选出 org_code='1020'的行记录

• 使用联合索引 trade_date_time, org_code -联合索引 trade_date_time, org_code 底层结构推导如下：

以查找 trade_date_time >='2019-05-01' and trade_date_time <='2020-05-01' and org_code='1020'为例：

1. 在范围查找的时候,直接找到最大,最小的值,然后进行链表遍历，故仅能用到 trade_date_time 的索引，无法使用到 org_code 索引

2. 基于 MySQL5.6+的索引下推特性，虽然 org_code 字段无法使用到索引树，但是可以用于过滤回表的主键 id 数。

小结：对于该 case, 索引效果[org_code,trade_date_time] > [trade_date_time, org_code]>[trade_date_time]。实际业务场景中，检索条件中 trade_date_time 基本上肯定会出现，但 org_code 却不一定，故索引的设计还需要结合实际业务需求。

优化 Order by

索引：

KEY `idx_account_trade_date_time` (`account_number`,`trade_date_time`),
KEY `idx_trade_date_times` (`trade_date_time`)
KEY `idx_createtime` (`create_time`),

慢 SQL:

SELE CT id,....,creator,modifier,create_time,update_time FROM statement
WHERE (account_number = 'XXX' AND create_time >= '2022-04-24 06:03:44' AND create_time <= '2022-04-24 08:03:44' AND dc_flag = 'C') ORDER BY trade_date_time DESC,id DESC LIMIT 0,1000;

优化前：SQL 执行超时被 kill 了

SELE CT id,....,creator,modifier,create_time,upda te_time FROM statement
WHERE (account_number = 'XXX' AND create_time >= '2022-04-24 06:03:44' AND create_time <= '2022-04-24 08:03:44' AND dc_flag = 'C') ORDER BY create_time DESC,id DESC LIMIT 0,1000;

优化后：执行总行数为:6 行，耗时 34ms。

MySQL使不使用索引与所查列无关，只与索引本身，where条件，order by 字段，grou p by 字段有关。索引的作用一个是查找，一个是排序。

业务拆分

sele ct * from order where status='S' and update_time < now-5min limit 500

拆分优化：

随着业务数据的增长 status='S'的数据基本占据数据的 90%以上，此时该条件无法走索引。我们可以结合业务特征，对数据获取按日期进行拆分。

date = now; minDate = now - 10 days
while(date > minDate) {
sele ct * from order where order_date={#date} and status='S' and upda te_time < now-5min limit 500
date = data + 1
}

数据库结构优化

1. 范式优化：表的设计合理化（符合 3NF），比如消除冗余（节省空间）；

2. 反范式优化：比如适当加冗余等（减少 join）

3. 拆分表：分区将数据在物理上分隔开，不同分区的数据可以制定保存在处于不同磁盘上的数据文件里。这样，当对这个表进行查询时，只需要在表分区中进行扫描，而不必进行全表扫描，明显缩短了查询时间，另外处于不同磁盘的分区也将对这个表的数据传输分散在不同的磁盘 I/O，一个精心设置的分区可以将数据传输对磁盘 I/O 竞争均匀地分散开。对数据量大的表可采取此方法，可按月建表分区。

SQL 语句优化

SQL 检查状态及分数计算逻辑

1. 尽量避免使用子查询

2. 用 IN 来替换 OR

3. 读取适当的记录 LIMIT M,N，而不要读多余的记录

4. 禁止不必要的 Order By 排序

5. 总和查询可以禁止排重用 union all

6. 避免随机取记录

7. 将多次插入换成批量 Insert 插入

8. 只返回必要的列，用具体的字段列表代替 sele ct * 语句

9. 区分 in 和 exists

10. 优化 Grou p By 语句

11. 尽量使用数字型字段

12. 优化 Join 语句

大表优化

• 分库分表（水平、垂直）

• 读写分离

• 数据定期归档

原理剖析

MySQL 逻辑架构图：

索引的优缺点

优点

• 提高查询语句的执行效率，减少 IO 操作的次数

• 创建唯一性索引，可以保证数据库表中每一行数据的唯一性

• 加了索引的列会进行排序，在使用分组和排序子句进行查询时，可以显著减少查询中分组和排序的时间

缺点

• 索引需要占物理空间

• 创建索引和维护索引要耗费时间，这种时间随着数据量的增加而增加

• 当对表中的数据进行增删改查时，索引也要动态的维护，这样就降低了数据的更新效率

索引的数据结构

主键索引

普通索引

组合索引

索引页结构

索引页由七部分组成，其中 Infimum 和 Supremum 也属于记录，只不过是虚拟记录，这里为了与用户记录区分开，还是决定将两者拆开。

数据行格式：

MySQL 有 4 种存储格式：

1. Compact

2. Redundant （5.0 版本以前用，已废弃）

3. Dynamic （MySQL5.7 默认格式）

4. Compressed

Dynamic 行存储格式下，对于处理行溢出（当一个字段存储长度过大时，会发生行溢出）时，仅存放溢出页内存地址。

索引的设计原则

哪些情况适合建索引

• 数据又数值有唯一性的限制

• 频繁作为 where 条件的字段

• 经常使用 grou p by 和 order by 的字段，既有 gro up by 又有 order by 的字段时，建议建联合索引

• 经常作为 upda te 或 dele te 条件的字段

• 经常需要 distinct 的字段

• 多表连接时的字段建议创建索引，也有注意事项

• 连接表数量最好不要超过 3 张，每增加一张表就相当于增加了一次嵌套循环，数量级增长会非常快

• 对多表查询时的 where 条件创建索引

• 对连接字段创建索引，并且数据类型保持一致

• 在确定数据范围的情况下尽量使用数据类型较小的，因为索引会也会占用空间

• 对字符串创建索引时建议使用字符串的前缀作为索引

• 这样做的好处是：

• 能节省索引的空间，

• 虽然不能精确定位，但是能够定位到相同的前缀，然后通过主键查询完整的字符串，这样既能节省空间，又减少了字符串的比较时间，还能解决排序问题。

• 区分度高（散列性高）的字段适合作为索引。

• 在多个字段需要创建索引的情况下，联合索引优先于单值索引。使用最频繁的列作为索引的最左侧 。

哪些情况下不需要使用索引

• 在 where 条件中用不到的字段不需要。

• 数据量小的不需要建索引，比如数据少于 1000 条。

• 由大量重复数据的列上不要建索引，比如性别字段中只有男和女时。

• 避免在经常更新的表或字段中创建过多的索引。

• 不建议主键使用无序的值作为索引，比如 uuid。

• 不要定义冗余或重复的索引

• 例如：已经创建了联合索引 key(id,name)后就不需要再单独建一个 key(id)的索引

索引优化之 MRR

例如有一张表 user，主键 id，普通字段 age，为 age 创建非聚集索引，有一条查询语句 sele ct* user from table where age > 18;(注意查询语句中的结果是*)

在 MySQL5.5 以及之前的版本中如何查询呢？先通过非聚集索引查询到 age>18 的第一条数据，获取到了主键 id；然后根据非聚集索引中的叶子节点存储的主键 id 去聚集索引中查询行数据；根据 age>18 的数据条数每次查询聚集索引，这个过程叫做回表。

上述的步骤有什么缺点呢？如何 age>18 的数据非常多，那么每次回表都需要经过 3 次 IO(假设 B+树的高度是 3)，那么会导致查询效率过低。

在 MySQL5.6 时针对上述问题进行了优化，优化器先查询到 age>3 的所有数据的主键 id，对所有主键的 id 进行排序，排序的结果缓存到 read_rnd_buffer，然后通过排好序的主键在聚簇索引中进行查询。

如果两个主键的范围相近，在同一个数据页中就可以之间按照顺序获取，那么磁盘 io 的过程将会大大降低。这个优化的过程就叫做 Multi Range Read(MRR) 多返回查询。

索引下推

假设有索引(name, age), 执行 SQL: sele ct * from tuser where name like '张%' and age=10;

MySQL 5.6 以后， 存储引擎根据（name，age）联合索引，找到，由于联合索引中包含列，所以存储引擎直接在联合索引里按照age=10过滤。按照过滤后的数据再一一进行回表扫描。

索引下推使用条件

• 只能用于range、 ref、 eq_ref、ref_or_null访问方法；

• 只能用于InnoDB和 MyISAM存储引擎及其分区表；

• 对存储引擎来说，索引下推只适用于二级索引（也叫辅助索引）;

索引下推的目的是为了减少回表次数，也就是要减少 IO 操作。对于的聚簇索引来说，数据和索引是在一起的，不存在回表这一说。

• 引用了子查询的条件不能下推；

• 引用了存储函数的条件不能下推，因为存储引擎无法调用存储函数。

思考：

1. MySQL 一张表到底能存多少数据？

2. 为什么要控制单行数据大小?

3. 优化案例 4 中优化前的 SQL 为什么走不到索引？

总结

抛开数据库硬件层面，数据库表设计、索引设计、业务代码逻辑、分库分表策略、数据归档策略都对 SQL 执行效率有影响，我们只有在整个设计、开发、运维阶段保持高度敏感、追求极致，才能让我们系统的可用性、伸缩性不会随着业务增长而劣化。

参考资料

1. https://help.aliyun.com/document_detail/311122.html

2. https://blog.csdn.net/qq_32099833/article/details/123150701

3. https://www.cnblogs.com/tufujie/p/9413852.html

来源：字节跳动技术团队

至暗时刻

2021年7月13日22:52，SRE收到大量服务和域名的接入层不可用报警，客服侧开始收到大量用户反馈B站无法使用，同时内部同学也反馈B站无法打开，甚至APP首页也无法打开。基于报警内容，SRE第一时间怀疑机房、网络、四层LB、七层SLB等基础设施出现问题，紧急发起语音会议，拉各团队相关人员开始紧急处理（为了方便理解，下述事故处理过程做了部分简化）。

初因定位

22:55 远程在家的相关同学登陆VPN后，无法登陆内网鉴权系统（B站内部系统有统一鉴权，需要先获取登录态后才可登陆其他内部系统），导致无法打开内部系统，无法及时查看监控、日志来定位问题。

22:57 在公司Oncall的SRE同学（无需VPN和再次登录内网鉴权系统）发现在线业务主机房七层SLB（基于OpenResty构建） CPU 100%，无法处理用户请求，其他基础设施反馈未出问题，此时已确认是接入层七层SLB故障，排除SLB以下的业务层问题。

23:07 远程在家的同学紧急联系负责VPN和内网鉴权系统的同学后，了解可通过绿色通道登录到内网系统。

23:17 相关同学通过绿色通道陆续登录到内网系统，开始协助处理问题，此时处理事故的核心同学（七层SLB、四层LB、CDN）全部到位。

故障止损

23:20 SLB运维分析发现在故障时流量有突发，怀疑SLB因流量过载不可用。因主机房SLB承载全部在线业务，先Reload SLB未恢复后尝试拒绝用户流量冷重启SLB，冷重启后CPU依然100%，未恢复。

23:22 从用户反馈来看，多活机房服务也不可用。SLB运维分析发现多活机房SLB请求大量超时，但CPU未过载，准备重启多活机房SLB先尝试止损。

23:23 此时内部群里同学反馈主站服务已恢复，观察多活机房SLB监控，请求超时数量大大降低，业务成功率恢复到50%以上。此时做了多活的业务核心功能基本恢复正常，如APP推荐、APP播放、评论&弹幕拉取、动态、追番、影视等。非多活服务暂未恢复。

23:25 - 23:55 未恢复的业务暂无其他立即有效的止损预案，此时尝试恢复主机房的SLB。

• 我们通过Perf发现SLB CPU热点集中在Lua函数上，怀疑跟最近上线的Lua代码有关，开始尝试回滚最近上线的Lua代码。

• 近期SLB配合安全同学上线了自研Lua版本的WAF，怀疑CPU热点跟此有关，尝试去掉WAF后重启SLB，SLB未恢复。

• SLB两周前优化了Nginx在balance_by_lua阶段的重试逻辑，避免请求重试时请求到上一次的不可用节点，此处有一个最多10次的循环逻辑，怀疑此处有性能热点，尝试回滚后重启SLB，未恢复。

• SLB一周前上线灰度了对 HTTP2 协议的支持，尝试去掉 H2 协议相关的配置并重启SLB，未恢复。

新建源站SLB

00:00 SLB运维尝试回滚相关配置依旧无法恢复SLB后，决定重建一组全新的SLB集群，让CDN把故障业务公网流量调度过来，通过流量隔离观察业务能否恢复。

00:20 SLB新集群初始化完成，开始配置四层LB和公网IP。

01:00 SLB新集群初始化和测试全部完成，CDN开始切量。SLB运维继续排查CPU 100%的问题，切量由业务SRE同学协助。

01:18 直播业务流量切换到SLB新集群，直播业务恢复正常。

01:40 主站、电商、漫画、支付等核心业务陆续切换到SLB新集群，业务恢复。

01:50 此时在线业务基本全部恢复。

恢复SLB

01:00 SLB新集群搭建完成后，在给业务切量止损的同时，SLB运维开始继续分析CPU 100%的原因。

01:10 - 01:27 使用Lua 程序分析工具跑出一份详细的火焰图数据并加以分析，发现 CPU 热点明显集中在对 lua-resty-balancer 模块的调用中，从 SLB 流量入口逻辑一直分析到底层模块调用，发现该模块内有多个函数可能存在热点。

01:28 - 01:38 选择一台SLB节点，在可能存在热点的函数内添加 debug 日志，并重启观察这些热点函数的执行结果。

01:39 - 01:58 在分析 debug 日志后，发现 lua-resty-balancer模块中的 _gcd 函数在某次执行后返回了一个预期外的值：nan，同时发现了触发诱因的条件：某个容器IP的weight=0。

01:59 - 02:06 怀疑是该 _gcd 函数触发了 jit 编译器的某个 bug，运行出错陷入死循环导致SLB CPU 100%，临时解决方案：全局关闭 jit 编译。

02:07 SLB运维修改SLB 集群的配置，关闭 jit 编译并分批重启进程，SLB CPU 全部恢复正常，可正常处理请求。同时保留了一份异常现场下的进程core文件，留作后续分析使用。

02:31 - 03:50 SLB运维修改其他SLB集群的配置，临时关闭 jit 编译，规避风险。

根因定位

11:40 在线下环境成功复现出该 bug，同时发现SLB 即使关闭 jit 编译也仍然存在该问题。此时我们也进一步定位到此问题发生的诱因：在服务的某种特殊发布模式中，会出现容器实例权重为0的情况。

12:30 经过内部讨论，我们认为该问题并未彻底解决，SLB 仍然存在极大风险，为了避免问题的再次产生，最终决定：平台禁止此发布模式；SLB 先忽略注册中心返回的权重，强制指定权重。

13:24 发布平台禁止此发布模式。

14:06 SLB 修改Lua代码忽略注册中心返回的权重。

14:30 SLB 在UAT环境发版升级，并多次验证节点权重符合预期，此问题不再产生。

15:00 - 20:00 生产所有 SLB 集群逐渐灰度并全量升级完成。

原因说明

背景

B站在19年9月份从Tengine迁移到了OpenResty，基于其丰富的Lua能力开发了一个服务发现模块，从我们自研的注册中心同步服务注册信息到Nginx共享内存中，SLB在请求转发时，通过Lua从共享内存中选择节点处理请求，用到了OpenResty的lua-resty-balancer模块。到发生故障时已稳定运行快两年时间。

在故障发生的前两个月，有业务提出想通过服务在注册中心的权重变更来实现SLB的动态调权，从而实现更精细的灰度能力。SLB团队评估了此需求后认为可以支持，开发完成后灰度上线。

诱因

• 在某种发布模式中，应用的实例权重会短暂的调整为0，此时注册中心返回给SLB的权重是字符串类型的"0"。此发布模式只有生产环境会用到，同时使用的频率极低，在SLB前期灰度过程中未触发此问题。

• SLB 在balance_by_lua阶段，会将共享内存中保存的服务IP、Port、Weight 作为参数传给lua-resty-balancer模块用于选择upstream server，在节点 weight = "0" 时，balancer 模块中的 _gcd 函数收到的入参 b 可能为 "0"。

根因

• Lua 是动态类型语言，常用习惯里变量不需要定义类型，只需要为变量赋值即可。

• Lua在对一个数字字符串进行算术操作时，会尝试将这个数字字符串转成一个数字。

• 在 Lua 语言中，如果执行数学运算 n % 0，则结果会变为 nan（Not A Number）。

• _gcd函数对入参没有做类型校验，允许参数b传入："0"。同时因为"0" != 0，所以此函数第一次执行后返回是 _gcd("0",nan)。如果传入的是int 0，则会触发[ if b == 0 ]分支逻辑判断，不会死循环。

• _gcd("0",nan)函数再次执行时返回值是 _gcd(nan,nan)，然后Nginx worker开始陷入死循环，进程 CPU 100%。

问题分析

\1. 为何故障刚发生时无法登陆内网后台？

事后复盘发现，用户在登录内网鉴权系统时，鉴权系统会跳转到多个域名下种登录的Cookie，其中一个域名是由故障的SLB代理的，受SLB故障影响当时此域名无法处理请求，导致用户登录失败。流程如下：

事后我们梳理了办公网系统的访问链路，跟用户链路隔离开，办公网链路不再依赖用户访问链路。

\2. 为何多活SLB在故障开始阶段也不可用？

多活SLB在故障时因CDN流量回源重试和用户重试，流量突增4倍以上，连接数突增100倍到1000W级别，导致这组SLB过载。后因流量下降和重启，逐渐恢复。此SLB集群日常晚高峰CPU使用率30%左右，剩余Buffer不足两倍。如果多活SLB容量充足，理论上可承载住突发流量， 多活业务可立即恢复正常。此处也可以看到，在发生机房级别故障时，多活是业务容灾止损最快的方案，这也是故障后我们重点投入治理的一个方向。

\3. 为何在回滚SLB变更无效后才选择新建源站切量，而不是并行？

我们的SLB团队规模较小，当时只有一位平台开发和一位组件运维。在出现故障时，虽有其他同学协助，但SLB组件的核心变更需要组件运维同学执行或review，所以无法并行。

\4. 为何新建源站切流耗时这么久？

我们的公网架构如下：

此处涉及三个团队：

• SLB团队：选择SLB机器、SLB机器初始化、SLB配置初始化

• 四层LB团队：SLB四层LB公网IP配置

• CDN团队：CDN更新回源公网IP、CDN切量

SLB的预案中只演练过SLB机器初始化、配置初始化，但和四层LB公网IP配置、CDN之间的协作并没有做过全链路演练，元信息在平台之间也没有联动，比如四层LB的Real Server信息提供、公网运营商线路、CDN回源IP的更新等。所以一次完整的新建源站耗时非常久。在事故后这一块的联动和自动化也是我们的重点优化方向，目前一次新集群创建、初始化、四层LB公网IP配置已经能优化到5分钟以内。

\5. 后续根因定位后证明关闭jit编译并没有解决问题，那当晚故障的SLB是如何恢复的？

当晚已定位到诱因是某个容器IP的weight="0"。此应用在1:45时发布完成，weight="0"的诱因已消除。所以后续关闭jit虽然无效，但因为诱因消失，所以重启SLB后恢复正常。

如果当时诱因未消失，SLB关闭jit编译后未恢复，基于定位到的诱因信息：某个容器IP的weight=0，也能定位到此服务和其发布模式，快速定位根因。

优化改进

此事故不管是技术侧还是管理侧都有很多优化改进。此处我们只列举当时制定的技术侧核心优化改进方向。

1. 多活建设

在23:23时，做了多活的业务核心功能基本恢复正常，如APP推荐、APP播放、评论&弹幕拉取、动态、追番、影视等。故障时直播业务也做了多活，但当晚没及时恢复的原因是：直播移动端首页接口虽然实现了多活，但没配置多机房调度。导致在主机房SLB不可用时直播APP首页一直打不开，非常可惜。通过这次事故，我们发现了多活架构存在的一些严重问题：

多活基架能力不足

• 机房与业务多活定位关系混乱。

• CDN多机房流量调度不支持用户属性固定路由和分片。

• 业务多活架构不支持写，写功能当时未恢复。

• 部分存储组件多活同步和切换能力不足，无法实现多活。

业务多活元信息缺乏平台管理

• 哪个业务做了多活？

• 业务是什么类型的多活，同城双活还是异地单元化？

• 业务哪些URL规则支持多活，目前多活流量调度策略是什么？

• 上述信息当时只能用文档临时维护，没有平台统一管理和编排。

多活切量容灾能力薄弱

• 多活切量依赖CDN同学执行，其他人员无权限，效率低。

• 无切量管理平台，整个切量过程不可视。

• 接入层、存储层切量分离，切量不可编排。

• 无业务多活元信息，切量准确率和容灾效果差。

我们之前的多活切量经常是这么一个场景：业务A故障了，要切量到多活机房。SRE跟研发沟通后确认要切域名A+URL A，告知CDN运维。CDN运维切量后研发发现还有个URL没切，再重复一遍上面的流程，所以导致效率极低，容灾效果也很差。

所以我们多活建设的主要方向：

多活基架能力建设

• 优化多活基础组件的支持能力，如数据层同步组件优化、接入层支持基于用户分片，让业务的多活接入成本更低。

• 重新梳理各机房在多活架构下的定位，梳理Czone、Gzone、Rzone业务域。

• 推动不支持多活的核心业务和已实现多活但架构不规范的业务改造优化。

多活管控能力提升

• 统一管控所有多活业务的元信息、路由规则，联动其他平台，成为多活的元数据中心。

• 支持多活接入层规则编排、数据层编排、预案编排、流量编排等，接入流程实现自动化和可视化。

• 抽象多活切量能力，对接CDN、存储等组件，实现一键全链路切量，提升效率和准确率。

• 支持多活切量时的前置能力预检，切量中风险巡检和核心指标的可观测。

2. SLB治理

架构治理

• 故障前一个机房内一套SLB统一对外提供代理服务，导致故障域无法隔离。后续SLB需按业务部门拆分集群，核心业务部门独立SLB集群和公网IP。

• 跟CDN团队、四层LB&网络团队一起讨论确定SLB集群和公网IP隔离的管理方案。

• 明确SLB能力边界，非SLB必备能力，统一下沉到API Gateway，SLB组件和平台均不再支持，如动态权重的灰度能力。

运维能力

• SLB管理平台实现Lua代码版本化管理，平台支持版本升级和快速回滚。

• SLB节点的环境和配置初始化托管到平台，联动四层LB的API，在SLB平台上实现四层LB申请、公网IP申请、节点上线等操作，做到全流程初始化5分钟以内。

• SLB作为核心服务中的核心，在目前没有弹性扩容的能力下，30%的使用率较高，需要扩容把CPU降低到15%左右。

• 优化CDN回源超时时间，降低SLB在极端故障场景下连接数。同时对连接数做极限性能压测。

自研能力

• 运维团队做项目有个弊端，开发完成自测没问题后就开始灰度上线，没有专业的测试团队介入。此组件太过核心，需要引入基础组件测试团队，对SLB输入参数做完整的异常测试。

• 跟社区一起，Review使用到的OpenResty核心开源库源代码，消除其他风险。基于Lua已有特性和缺陷，提升我们Lua代码的鲁棒性，比如变量类型判断、强制转换等。

• 招专业做LB的人。我们选择基于Lua开发是因为Lua简单易上手，社区有类似成功案例。团队并没有资深做Nginx组件开发的同学，也没有做C/C++开发的同学。

3. 故障演练

本次事故中，业务多活流量调度、新建源站速度、CDN切量速度&回源超时机制均不符合预期。所以后续要探索机房级别的故障演练方案：

• 模拟CDN回源单机房故障，跟业务研发和测试一起，通过双端上的业务真实表现来验收多活业务的容灾效果，提前优化业务多活不符合预期的隐患。

• 灰度特定用户流量到演练的CDN节点，在CDN节点模拟源站故障，观察CDN和源站的容灾效果。

• 模拟单机房故障，通过多活管控平台，演练业务的多活切量止损预案。

4. 应急响应

B站一直没有NOC/技术支持团队，在出现紧急事故时，故障响应、故障通报、故障协同都是由负责故障处理的SRE同学来承担。如果是普通事故还好，如果是重大事故，信息同步根本来不及。所以事故的应急响应机制必须优化：

• 优化故障响应制度，明确故障中故障指挥官、故障处理人的职责，分担故障处理人的压力。

• 事故发生时，故障处理人第一时间找backup作为故障指挥官，负责故障通报和故障协同。在团队里强制执行，让大家养成习惯。

• 建设易用的故障通告平台，负责故障摘要信息录入和故障中进展同步。

本次故障的诱因是某个服务使用了一种特殊的发布模式触发。我们的事件分析平台目前只提供了面向应用的事件查询能力，缺少面向用户、面向平台、面向组件的事件分析能力：

• 跟监控团队协作，建设平台控制面事件上报能力，推动更多核心平台接入。

• SLB建设面向底层引擎的数据面事件变更上报和查询能力，比如服务注册信息变更时某个应用的IP更新、weight变化事件可在平台查询。

• 扩展事件查询分析能力，除面向应用外，建设面向不同用户、不同团队、不同平台的事件查询分析能力，协助快速定位故障诱因。

总结

此次事故发生时，B站挂了迅速登上全网热搜，作为技术人员，身上的压力可想而知。事故已经发生，我们能做的就是深刻反思，吸取教训，总结经验，砥砺前行。

此篇作为“713事故”系列之第一篇，向大家简要介绍了故障产生的诱因、根因、处理过程、优化改进。后续文章会详细介绍“713事故”后我们是如何执行优化落地的，敬请期待。

最后，想说一句：多活的高可用容灾架构确实生效了。

来源：哔哩哔哩技术

大家好啊，世界您好啊，请多关照哈，，，，，，，，，，，

许多将数据驱动到 Elasticsearch 中的系统将利用 Elasticsearch 为新插入的文档自动生成的 id 值。 但是，如果数据源意外地将同一文档多次发送到Elasticsearch，并且如果将这种自动生成的 id 值用于 Elasticsearch 插入的每个文档，则该同一文档将使用不同的id值多次存储在 Elasticsearch 中。 如果发生这种情况，那么可能有必要找到并删除此类重复项。 因此，在此博客文章中，我们介绍如何通过

• 使用 Logstash

• 使用 Python 编写的自定义代码从 Elasticsearch 中检测和删除重复文档

示例文档结构

就本博客而言，我们假设 Elasticsearch 集群中的文档具有以下结构。 这对应于包含代表股票市场交易的文档的数据集。

{
    "_index": "stocks",
    "_type": "doc",
    "_id": "6fo3tmMB_ieLOlkwYclP",
    "_version": 1,
    "found": true,
    "_source": {
        "CAC": 1854.6,
        "host": "Alexanders-MBP",
        "SMI": 2061.7,
        "@timestamp": "2017-01-09T02:30:00.000Z",
        "FTSE": 2827.5,
        "DAX": 1527.06,
        "time": "1483929000",
        "message": "1483929000,1527.06,2061.7,1854.6,2827.5\r",
        "@version": "1"
    }
}

给定该示例文档结构，出于本博客的目的，我们任意假设如果多个文档的 [“CAC”，“FTSE”，“SMI”] 字段具有相同的值，则它们是彼此重复的。

使用 Logstash 对 Elasticsearch 文档进行重复数据删除

这种方法已经在之前的文章 “Logstash：处理重复的文档” 已经描述过了。Logstash 可用于检测和删除 Elasticsearch 索引中的重复文档。 在那个文章中，我们已经对这个方法进行了详述，也做了展示。我们也无妨做一个更进一步的描述。

在下面的示例中，我编写了一个简单的 Logstash 配置，该配置从 Elasticsearch 集群上的索引读取文档，然后使用指纹过滤器根据 ["CAC", "FTSE", "SMI"] 字段的哈希值为每个文档计算唯一的 _id 值，最后将每个文档写回到同一 Elasticsearch 集群上的新索引，这样重复的文档将被写入相同的 _id 并因此被消除。

此外，通过少量修改，相同的 Logstash 过滤器也可以应用于写入新创建的索引的将来文档，以确保几乎实时删除重复项。这可以通过更改以下示例中的输入部分以接受来自实时输入源的文档，而不是从现有索引中提取文档来实现。

请注意，使用自定义 id 值（即不是由 Elasticsearch 生成的 _id）将对索引操作的[写入性能产生一些影响](https://www.elastic.co/guide/en/elasticsearch/reference/master/tune-for-indexing-speed.html#use_auto_generated_ids)。

另外，值得注意的是，根据所使用的哈希算法，此方法理论上可能会导致 id 值的[哈希冲突数](https://en.wikipedia.org/wiki/Collision(computer_science))不为零，这在理论上可能导致两个不相同的文档映射到相同的_id，因此导致这些文档之一丢失。对于大多数实际情况，哈希冲突的可能性可能非常低。对不同哈希函数的详细分析不在本博客的讨论范围之内，但是应仔细考虑指纹过滤器中使用的哈希函数，因为它将影响提取性能和哈希冲突次数。

下面给出了使用指纹过滤器对现有索引进行重复数据删除的简单 Logstash 配置。

input {
  # Read all documents from Elasticsearch 
  elasticsearch {
    hosts => "localhost"
    index => "stocks"
    query => '{ "sort": [ "_doc" ] }'
  }
}
# This filter has been updated on February 18, 2019
filter {
    fingerprint {
        key => "1234ABCD"
        method => "SHA256"
        source => ["CAC", "FTSE", "SMI"]
        target => "[@metadata][generated_id]"
        concatenate_sources => true # <-- New line added since original post date
    }
}
output {
    stdout { codec => dots }
    elasticsearch {
        index => "stocks_after_fingerprint"
        document_id => "%{[@metadata][generated_id]}"
    }
}

用于 Elasticsearch 文档重复数据删除的自定义 Python 脚本

内存有效的方法

如果不使用 Logstash，则可以使用自定义 python 脚本有效地完成重复数据删除。 对于这种方法，我们计算定义为唯一标识文档的["CAC"，"FTSE"，"SMI"] 字段的哈希值 （Hash）。 然后，我们将此哈希用作 python 字典中的键，其中每个字典条目的关联值将是映射到同一哈希的文档 _id 的数组。

如果多个文档具有相同的哈希，则可以删除映射到相同哈希的重复文档。 另外，如果你担心哈希值冲突的可能性，则可以检查映射到同一散列的文档的内容，以查看文档是否确实相同，如果是，则可以消除重复项。

检测算法分析

对于 50GB 的索引，如果我们假设索引包含平均大小为 0.4 kB 的文档，则索引中将有1.25亿个文档。 在这种情况下，使用128位 md5 哈希将重复数据删除数据结构存储在内存中所需的内存量约为128位x 125百万= 2GB 内存，再加上160位_id将需要另外160位x 125百万= 2.5 GB 的内存。 因此，此算法将需要4.5GB 的 RAM 数量级，以将所有相关的数据结构保留在内存中。 如果可以应用下一节中讨论的方法，则可以大大减少内存占用。

算法增强

在本节中，我们对算法进行了增强，以减少内存使用以及连续删除新的重复文档。

如果你要存储时间序列数据，并且知道重复的文档只会在彼此之间的一小段时间内出现，那么您可以通过在文档的子集上重复执行该算法来改善该算法的内存占用量在索引中，每个子集对应一个不同的时间窗口。例如，如果您有多年的数据，则可以在datetime字段（在过滤器上下文中以获得最佳性能）上使用范围查询，一次仅一周查看一次数据集。这将要求算法执行52次（每周一次）-在这种情况下，这种方法将使最坏情况下的内存占用减少52倍。

在上面的示例中，你可能会担心没有检测到跨星期的重复文档。假设你知道重复的文档间隔不能超过2小时。然后，您需要确保算法的每次执行都包含与之前算法执行过的最后一组文档重叠2小时的文档。对于每周示例，因此，您需要查询170小时（1周+ 2小时）的时间序列文档，以确保不会丢失任何重复项。

如果你希望持续定期从索引中清除重复的文档，则可以对最近收到的文档执行此算法。与上述逻辑相同-确保分析中包括最近收到的文档以及与稍旧的文档的足够重叠，以确保不会无意中遗漏重复项。

用于检测重复文档的 Python 代码

以下代码演示了如何可以有效地评估文档以查看它们是否相同，然后根据需要将其删除。 但是，为了防止意外删除文档，在本示例中，我们实际上并未执行删除操作。 这样的功能的实现将是非常直接的。

可以在 github 上找到用于从 Elasticsearch 中删除文档重复数据的代码。

#!/usr/local/bin/python3
import hashlib
from elasticsearch import Elasticsearch
es = Elasticsearch(["localhost:9200"])
dict_of_duplicate_docs = {}
# The following line defines the fields that will be
# used to determine if a document is a duplicate
keys_to_include_in_hash = ["CAC", "FTSE", "SMI"]
# Process documents returned by the current search/scroll
def populate_dict_of_duplicate_docs(hits):
    for item in hits:
        combined_key = ""
        for mykey in keys_to_include_in_hash:
            combined_key += str(item['_source'][mykey])
        _id = item["_id"]
        hashval = hashlib.md5(combined_key.encode('utf-8')).digest()
        # If the hashval is new, then we will create a new key
        # in the dict_of_duplicate_docs, which will be
        # assigned a value of an empty array.
        # We then immediately push the _id onto the array.
        # If hashval already exists, then
        # we will just push the new _id onto the existing array
        dict_of_duplicate_docs.setdefault(hashval, []).append(_id)
# Loop over all documents in the index, and populate the
# dict_of_duplicate_docs data structure.
def scroll_over_all_docs():
    data = es.search(index="stocks", scroll='1m',  body={"query": {"match_all": {}}})
    # Get the scroll ID
    sid = data['_scroll_id']
    scroll_size = len(data['hits']['hits'])
    # Before scroll, process current batch of hits
    populate_dict_of_duplicate_docs(data['hits']['hits'])
    while scroll_size > 0:
        data = es.scroll(scroll_id=sid, scroll='2m')
        # Process current batch of hits
        populate_dict_of_duplicate_docs(data['hits']['hits'])
        # Update the scroll ID
        sid = data['_scroll_id']
        # Get the number of results that returned in the last scroll
        scroll_size = len(data['hits']['hits'])
def loop_over_hashes_and_remove_duplicates():
    # Search through the hash of doc values to see if any
    # duplicate hashes have been found
    for hashval, array_of_ids in dict_of_duplicate_docs.items():
      if len(array_of_ids) > 1:
        print("********** Duplicate docs hash=%s **********" % hashval)
        # Get the documents that have mapped to the current hashval
        matching_docs = es.mget(index="stocks", doc_type="doc", body={"ids": array_of_ids})
        for doc in matching_docs['docs']:
            # In this example, we just print the duplicate docs.
            # This code could be easily modified to delete duplicates
            # here instead of printing them
            print("doc=%s\n" % doc)
def main():
    scroll_over_all_docs()
    loop_over_hashes_and_remove_duplicates()
main()

结论

在此博客文章中，我们展示了两种在 Elasticsearch 中对文档进行重复数据删除的方法。 第一种方法使用 Logstash 删除重复的文档，第二种方法使用自定义的 Python 脚本查找和删除重复的文档。

来源：https://blog.csdn.net/UbuntuTouch/article/details/106643400

原文： How to Find and Remove Duplicate Documents in Elasticsearch | Elastic Blog

引言

对于客户或者需求方来说，可以集成交付的软件才是有价值的。

每个软件都有集成的过程，如果软件规模比较小，比如只有一个人而且没有外部依赖，那集成没什么问题。随着软件变得复杂，依赖变多，开发人员变多，那么早集成、常集成，就可以尽早暴露问题，做出相应的调整，防止在软件后期才发现问题，从而导致软件失败。

定义

大师Martin Fowler对持续集成是这样定义的:

持续集成是一种软件开发实践，即团队开发成员经常集成它们的工作，通常每个成员每天至少集成一次，也就意味着每天可能会发生多次集成。每次集成都通过自动化的构建（包括编译，发布，自动化测试)来验证，从而尽快地发现集成错误。许多团队发现这个过程可以大大减少集成的问题，让团队能够更快的开发内聚的软件。

优点

• 降低软件风险 早集成，常集成，并且做了有效的测试，有利于尽早暴露问题和软件缺陷，了解软件的健康情况。假定越少的软件，对于维护和新业务开发都是有利的。 如果botslab每一个品类的设备都单独分支开发，不能及时集成进来，依赖问题、冲突问题、业务复用问题不能尽早解决，问题累积到一定程度解决成本会变大，业务发展是不会为代码重构让路的，那么软件质量的降低是必然的，很多项目都是这样失败的

• 减少重复过程 软件集成的过程看起来简单，但是做起来难。软件的编译，测试，审查，部署，反馈，这些重复劳动是非常耗时，且没有意义的，自动化集成可以让开发解放出来，做一些用脑袋的事情。 如果出现疏漏，会给下游的参与者带来额外工作量和项目质量的误判。软件的输出质量是会影响项目计划的，所以持续集成很重要的一点就是自动化

• 随时生成可以部署的软件 持续集成可以随时随地输出可以部署的软件，这一点对于需求方或客户是明显的好处。我们可以对客户说软件有多么好的架构，多么高质量的代码，但是对于客户来说，一个可以使用的软件才是他的实际资产。持续交付可以尽早的暴露产品问题和开发方向，客户才能给出有效的意见和开发重点。

• 软件是透明的 持续集成会生成软件构建状态和品质信息，经常集成可以看到一些趋势，预测一些软件质量走向。

• 团队信心 持续集成可以建立团队的信心，开发清楚的知道自己的代码产生了什么影响，测试对软件质量的预测稳定，产品或客户可以放心的需求了

步骤

1. 统一的代码库

2. 自动构建

3. 自动测试

4. 每个人每天都要向代码库主干提交代码

5. 每次代码递交后都会在持续集成服务器上触发一次构建

6. 保证快速构建

7. 模拟生产环境的自动测试

8. 每个人都可以很容易的获取最新可执行的应用程序

9. 每个人都清楚正在发生的状况

10. 自动化的部署

原则

1. 所有的开发人员需要在本地机器上做本地构建，然后再提交的版本控制库中，从而确保他们的变更不会导致持续集成失败。

2. 开发人员每天至少向版本控制库中提交一次代码。

3. 开发人员每天至少需要从版本控制库中更新一次代码到本地机器。

4. 需要有专门的集成服务器来执行集成构建,每天要执行多次构建。

5. 每次构建都要100%通过。

6. 每次构建都可以生成可发布的产品。

7. 修复失败的构建是优先级最高的事情。

作者：QiShare
来源：juejin.cn/post/6986884632222384141

背景

在 PHP 安全测试中最单调乏味的任务之一就是检查不安全的 PHP 配置项。作为一名 PHP 安全海报的继承者，我们创建了一个脚本用来帮助系统管理员如同安全专家一样尽可能快速且全面地评估 php.ini 和相关主题的状态。在下文中，该脚本被称作“PHP 安全配置项检查器”，或者 pcc。

https://github.com/sektioneins/pcc

概念

• 一个便于分发的单文件
• 有对每个安全相关的 ini 条目的简单测试
• 包含一些其他测试 - 但不太复杂
• 兼容 PHP >= 5.4, 或者 >= 5.0
• 没有复杂/过度设计的代码，例如没有类/接口，测试框架，类库等等。它应该第一眼看上去是显而易见的-甚至对于新手-这个工具怎么使用能用来做什么。
• 没有（或者少量的）依赖

使用 / 安装

• CLI：简单调用 php phpconfigcheck.php。然后，添加参数 -a 以便更好的查看隐藏结果， -h 以 HTML 格式输出， -j 以 JSON 格式输出.
• WEB： 复制这个脚本文件到你的服务器上的任意一个可访问目录，比如 root 目录。参见下面的“防护措施”。
  在非 CLI 模式下默认输出 HTML 格式。可以通过修改设置环境变量PCC_OUTPUT_TYPE=text 或者 PCC_OUTPUT_TYPE=json改变这个行为。
  一些测试用例默认是被隐藏的，特别是skipped、ok和 unknown/untested这些。要显示全部结果，可以用 phpconfigcheck.php?showall=1，但这并不适用于 JSON 输出，它默认返回全部结果。
  在 WEB 模式下控制输出格式用 phpconfigcheck.php?format=...， format的值可以是 text, html 或者 json中的一个，例如： phpconfigcheck.php?format=text。 format 参数优先于 PCC_OUTPUT_TYPE。
  

保障措施

大多数情况下，最好是自己来关注与安全性相关的问题比如PHP的配置。脚本已实现下列保障措施：

• mtime检查：脚本在非CLI环境中只能工作两天。可以通过touch phpconfigcheck.php或者将脚本文件再次复制到你的服务器(例如通过SCP)来重新进行mtime检查。可以通过设置环境量： PCC_DISABLE_MTIME=1，比如在apache的.htaccess文件中设置SetEnv PCC_DISABLE_MTIME 1来禁用mtime检查。
• 来源IP检查：默认情况下，只有localhost (127.0.0.1 和 ::1)才能访问这个脚本。其他主机可以通过在PCC_ALLOW_IP中添加IP地址或者通配符表达式的方式来访问脚本，比如在.htaccess文件中设置SetEnv PCC_ALLOW_IP 10.0.0.*。你还可以选择通过SSH端口转发访问您的web服务器， 比如 ssh -D 或者 ssh -L。

下载

可以通过github下载第一个完整的开发版： https://github.com/sektioneins/pcc

如果有好的建议或者遇到bug请给我们提issue：

截图

HTML输出的列表是根据问题严重性排序的，通过颜色代码的形式列出了所有建议。列表顶部的状态行会显示问题的数量。

转载自： https://cloud.tencent.com/developer/article/1911011

都说Vue2简单上手容易，的确，看了官方文档确实觉得上手很快，除了ES6语法和webpack的配置让你感到陌生，重要的是思路的变换，以前用jq随便拿全局变量和修改dom的锤子不能用了，vue只用关心数据本身，不用再频繁繁琐的操作dom，注册事件、监听事件、取消事件。。。。（确实很烦）。vue的官方文档还是不错的，由浅到深，如果不使用构建工具确实用的很爽，但是这在实际项目应用中是不可能的，当用vue-cli构建一个工程的时候，发现官方文档还是不够用，需要熟练掌握es6，而vue的全家桶(vue-cli,vue-router,vue-resource,vuex)还是都要上的。

vue.js有著名的全家桶系列，包含了vue-router，vuex， vue-resource，再加上构建工具vue-cli，就是一个完整的vue项目的核心构成。

vue-cli这个构建工具大大降低了webpack的使用难度，支持热更新，有webpack-dev-server的支持，相当于启动了一个请求服务器，给你搭建了一个测试环境，只关注开发就OK。

1.安装vue-cli

① 使用npm（需要安装node环境）全局安装webpack，打开命令行工具输入：npm install webpack -g或者（npm install -g webpack），安装完成之后输入 webpack -v，如下图，如果出现相应的版本号，则说明安装成功。

② 全局安装vue-cli，在cmd中输入命令:

npm install --global vue-cli

安装成功：

安装完成之后输入 vue -V（注意这里是大写的“V”），如下图，如果出现相应的版本号，则说明安装成功。

打开C:\Users\Andminster\AppData\Roaming\npm目录下可以看到：

打开node_modules也可以看到：

2.用vue-cli来构建项目

① 我首先在D盘新建一个文件夹（dxl_vue）作为项目存放地，然后使用命令行cd进入到项目目录输入：

vue init webpack baoge

baoge是自定义的项目名称，命令执行之后，会在当前目录生成一个以该名称命名的项目文件夹。
输入命令后，会跳出几个选项让你回答：

• Project name (baoge)： -----项目名称，直接回车，按照括号中默认名字（注意这里的名字不能有大写字母，如果有会报错Sorry, name can no longer contain capital letters），阮一峰老师博客为什么文件名要小写 ，可以参考一下。
• Project description (A Vue.js project)： ----项目描述，也可直接点击回车，使用默认名字
• Author ()： ----作者，输入dongxili
  接下来会让用户选择：
• Runtime + Compiler: recommended for most users 运行加编译，既然已经说了推荐，就选它了
  Runtime-only: about 6KB lighter min+gzip, but templates (or any Vue-specificHTML) are ONLY allowed in .vue files - render functions are required elsewhere 仅运行时，已经有推荐了就选择第一个了
• Install vue-router? (Y/n) 是否安装vue-router，这是官方的路由，大多数情况下都使用，这里就输入“y”后回车即可。
• Use ESLint to lint your code? (Y/n) 是否使用ESLint管理代码，ESLint是个代码风格管理工具，是用来统一代码风格的，一般项目中都会使用。
  接下来也是选择题Pick an ESLint preset (Use arrow keys) 选择一个ESLint预设，编写vue项目时的代码风格，直接y回车
• Setup unit tests with Karma + Mocha? (Y/n) 是否安装单元测试，我选择安装y回车
• Setup e2e tests with Nightwatch(Y/n)? 是否安装e2e测试 ，我选择安装y回车

回答完毕后上图就开始构建项目了。

② 配置完成后，可以看到目录下多出了一个项目文件夹baoge，然后cd进入这个文件夹：
安装依赖：

npm install

 ( 如果安装速度太慢。可以安装淘宝镜像，打开命令行工具，输入：
npm install -g cnpm --registry=https://registry.npm.taobao.org
 然后使用cnpm来安装 )

npm install ：安装所有的模块，如果是安装具体的哪个个模块，在install 后面输入模块的名字即可。而只输入install就会按照项目的根目录下的package.json文件中依赖的模块安装（这个文件里面是不允许有任何注释的），每个使用npm管理的项目都有这个文件，是npm操作的入口文件。因为是初始项目，还没有任何模块，所以我用npm install 安装所有的模块。安装完成后，目录中会多出来一个node_modules文件夹，这里放的就是所有依赖的模块。

然后现在，baoge文件夹里的目录是这样的：

解释下每个文件夹代表的意思(仔细看一下这张图）：

image.png

3.启动项目

npm run dev

如果浏览器打开之后，没有加载出页面，有可能是本地的 8080 端口被占用，需要修改一下配置文件 config里的index.js

还有，如果本地调试项目时，建议将build 里的assetsPublicPath的路径前缀修改为 ' ./ '（开始是 ' / '），因为打包之后，外部引入 js 和 css 文件时，如果路径以 ' / ' 开头，在本地是无法找到对应文件的（服务器上没问题）。所以如果需要在本地打开打包后的文件，就得修改文件路径。
我的端口没有被占用，直接成功（服务启动成功后浏览器会默认打开一个“欢迎页面”）：

注意：在进行vue页面调试时，一定要去谷歌商店下载一个vue-tool扩展程序。

4.vue-cli的webpack配置分析

• 从package.json可以看到开发和生产环境的入口。
  
  
• 可以看到dev中的设置，build/webpack.dev.conf.js，该文件是开发环境中webpack的配置入口。
• 在webpack.dev.conf.js中出现webpack.base.conf.js，这个文件是开发环境和生产环境，甚至测试环境，这些环境的公共webpack配置。可以说，这个文件相当重要。
• 还有config/index.js 、build/utils.js 、build/build.js等，具体请看这篇介绍：
  https://segmentfault.com/a/1190000008644830

5.打包上线

注意，自己的项目文件都需要放到 src 文件夹下。
在项目开发完成之后，可以输入 npm run build 来进行打包工作。

npm run build

另：

1.npm 开启了npm run dev以后怎么退出或关闭？

ctrl+c

2.--save-dev

自动把模块和版本号添加到模块配置文件package.json中的依赖里devdependencies部分

3. --save-dev 与 --save 的区别

--save     安装包信息将加入到dependencies（生产阶段的依赖）

--save-dev 安装包信息将加入到devDependencies（开发阶段的依赖），所以开发阶段一般使用它

打包完成后，会生成 dist 文件夹，如果已经修改了文件路径，可以直接打开本地文件查看。
项目上线时，只需要将 dist 文件夹放到服务器就行了。

转载自： https://cloud.tencent.com/developer/article/1896690

Java异常

异常指不期而至的各种状况，如：文件找不到、网络连接失败、非法参数等。异常是一个事件，它发生在程序运行期间，干扰了正常的指令流程。Java通 过API中Throwable类的众多子类描述各种不同的异常。因而，Java异常都是对象，是Throwable子类的实例，描述了出现在一段编码中的 错误条件。当条件生成时，错误将引发异常。
Java异常类层次结构图：

try {  

    // 可能会发生异常的程序代码  

} catch (Type1 id1){  

    // 捕获并处置try抛出的异常类型Type1  

}  

catch (Type2 id2){  

     //捕获并处置try抛出的异常类型Type2  

}  

public class TestException {  

    public static void main(String[] args) {  

        int a = 6;  

        int b = 0;  

        try { // try监控区域  

              

            if (b == 0) throw new ArithmeticException(); // 通过throw语句抛出异常  

            System.out.println("a/b的值是：" + a / b);  

        }  

        catch (ArithmeticException e) { // catch捕捉异常  

            System.out.println("程序出现异常，变量b不能为0。");  

        }  

        System.out.println("程序正常结束。");  

    }  

}  制

public static void main(String[] args) {  

        int a = 6;  

        int b = 0;  

        try {  

            System.out.println("a/b的值是：" + a / b);  

        } catch (ArithmeticException e) {  

            System.out.println("程序出现异常，变量b不能为0。");  

        }  

        System.out.println("程序正常结束。");  

    }  

}  制

public class TestException {  

    public static void main(String[] args) {  

        int a, b;  

        a = 6;  

        b = 0; // 除数b 的值为0  

        System.out.println(a / b);  

    }  

}  复制

public class TestException {  

    public static void main(String[] args) {  

        int[] intArray = new int[3];  

        try {  

            for (int i = 0; i <= intArray.length; i++) {  

                intArray[i] = i;  

                System.out.println("intArray[" + i + "] = " + intArray[i]);  

                System.out.println("intArray[" + i + "]模 " + (i - 2) + "的值:  "  

                        + intArray[i] % (i - 2));  

            }  

        } catch (ArrayIndexOutOfBoundsException e) {  

            System.out.println("intArray数组下标越界异常。");  

        } catch (ArithmeticException e) {  

            System.out.println("除数为0异常。");  

        }  

        System.out.println("程序正常结束。");  

    }  

}  

catch (ArithmeticException e){  

      System.out.println("除数为0异常。");  

 } 

项目需求

最近在开发一个 electron 程序，其中有用到和硬件通讯部分；硬件厂商给的是 .dll 链接库做通讯桥接，
第一版本使用 C 写的 Node.js 扩展 😁；由于有异步任务的关系，实现使用了 N-API 提供的多线程做异步任务调度，
虽然功能实现了，但是也有些值得思考的点。

• 纯 C 编程效率低，木有 trycatch 的语言调试难度也大 (磕磕绊绊的)
• 编写好的 .node 扩展文件，放在 electron 主进程中运行会有一定的隐患稍有差错会导致软件闪退 (后来用子进程隔离运行)
• 基于 N-API 方式去编写 Node.js 插件会显得有所束缚，木有那种随心所欲写 C 的那种“顺畅”；尤其是多线程部分

综上考虑，加上通讯功能又是调用 .dll 文件，索性转战 C#，对于 windows 来说再合适不过了；但是问题是 C# 咋编译到 Node.js 中？
答案是“编译不了”。
插件实现的功能只是收到命令后调用 .dll 去操作硬件，再时时能把结果返回即可。
基于这个需求我们用 C# 去调用 .dll 文件，然后再解决派发命令、实时获取结果的通讯问题就OK了，剩下的就都是好处啦

• C# 编写难度低于 C，又是 windows 亲儿子，基于 .NET Framework 编译后的程序仅 19KB (C实现同样功能编出来的.node文件 565KB)
• 基于 C# 的插件独立于 Node.js 运行环境，程序出了问题不会影响 electron 应用
• 木有任何的编程束缚，~亲想咋写就咋写

通讯问题

说这个之前我们还忽略了一个问题，这个 C# 的程序(.exe文件)如果启动？
既然是一个程序(.exe文件)，我们双击即可执行；既然双击即可执行，我们就可以用 child_process 模块提供的
spawn 去拉起程序(代替鼠标双击)；

好！程序已经启动了，那么该到了如果通讯的环节了。
spawn 的执行就是开启了一个单独的进程，通讯问题也就是进程通讯问题。之前如果你用过 spawn 启动过 Node.js 程序(.js文件)，那么你肯定知道通讯使用 send 方法即可；这个是 Node.js 内置的方式

我们启动的进程是 C# 程序，通讯问题只能我们自己来解决了；进程通讯的方式有好多这里不展开。对于前端(web)攻城狮来讲，我们最熟悉的莫过于 http 通讯方式了；就用它！

• C# 程序端启动开启一个 http 服务等待 Node.js 端发送请求过来；根据参数决定要干啥
• spawn 启动的应用(进程)，会返回一个 ChildProcessWithoutNullStreams (这个我也不能很明确的理解)；能够接收到标准的 stdio 输入/输出
  那我们就利用这点使用 ChildProcessWithoutNullStreams.stdout.on('data', chunk => console.log(chunk.toString())) 的方式就可以收到 C# 通过 stdio 即 Console.WriteLine() 发过来的数据；
  哇！好方便~
• 可能有人会想到用双工的 web socket 实现通讯，很棒！实现方式确实有很多种，这里用 Console.WriteLine() 通过标准的 stdio 方式实现，算不算是一个开发成本不高的讨巧做法呢！

大致流程

• 如果觉得这篇文章有难度，可以看简单版的哦 Node.js 利用 stdio 标准输入/输出实现与 C# 程序通讯

开发环境

• C# 代码部分使用 Visual Studio 2017
• test.js 代码部分使用 VsCode

using System; 

using System.Collections.Generic; 

using System.Linq; using System.Text;  

using System.Net; 

using System.Net.Sockets; 

using System.Threading; 

using System.Text.RegularExpressions;  

namespace NodeAddons {     

    class Program {        

         static TcpListener listener;         

         static int port = 8899;         

         static string now = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss");  

         static void Main(string[] args){             

                 listener = new TcpListener(IPAddress.Any, port);             

                 listener.Start();              

                 // 启用服务器线程             

                 new Thread(new ThreadStart(StartServer)).Start();              

                 Console.WriteLine("Http server run at {0}.", port);         

        }          

        // Http 服务器         

        static void StartServer() {             

            while(true) {                 

            // 这里会阻塞线程，直到接受到一个请求                 

            Socket socket = listener.AcceptSocket();                 

            // 将请求单独开一个线程处理；while(true)会回到等待下一个请求状态，周而复始

            new Thread(new ParameterizedThreadStart(HandleRequest)).Start(socket);       }                      }          // 处理一个请求         static void HandleRequest(object args) { Socket socket = (Socket)args;           byte[] receive = new byte[1024];             socket.Receive(receive, receive.Length, SocketFlags.None);             string httpRawTxt = Encoding.ASCII.GetString(receive);              // 通过 stdio(Console.WriteLine) 实现与 node.js 通讯             // ## 开头、结尾，方便区分这个条输出是给 node.js 通讯用的             Console.WriteLine("##" + httpRawTxt + "##");             SendToBrowser(ref socket, now); }          // 发送数据         static void SendToBrowser(ref Socket socket, string body)         {             string header = "HTTP/1.1 200 OK\r\n" + "Content-Type: text/html\r\n"                 + "Content-Length: " + body.Length + "\r\n" + "Access-Control-Allow-Origin:*\r\n" // 支持跨域 + "\r\n"; // 响应头与响应体分界             byte[] data = Encoding.ASCII.GetBytes(header + body);              if (socket.Connected) {                 int res = socket.Send(data, data.Length, SocketFlags.None);                 if (res == -1){                          Console.WriteLine("Socket Error cannot Send Packet.");}else{                   Console.WriteLine(">> [{0}]", now);                 }                 socket.Close();}}}}

Node.js 部分

const http = require('http'); 

const cp = require('child_process');

 const path = require('path');  

// const handel = cp.spawn(path.join(__dirname, 'dist/NodeAddons.exe')); 

const handel = cp.spawn(path.join(__dirname, 'dist/NodeAddons_WithConsole.exe')); handel.stdout.on('data', chunk => {   const str = chunk.toString();   

// 约定 ##数据## 的字符串为通讯数据   

let res = str.match(/##([\S\s]*)##/g);   

if (!Array.isArray(res)) return;   

res = res[0].match(/(?<=(\?))(.*)(?=(\sHTTP\/1.1))/);   

if (!Array.isArray(res)) return;   

console.log('[stdout queryString]', res[0]); });  

function query(param, cb) {   

http.get(`http://127.0.0.1:8899/?${(new URLSearchParams(param)).toString()}`,     res => { res.on('data', chunk => {         cb(chunk.toString());       }); 

}); }  

query({ name: 'anan', age: 29, time: Date.now() }, httpRawTxt => {   console.log('[http response]', httpRawTxt); });  

// 监听 Ctrl + c process.on('SIGINT', () => {   handel.kill();   process.exit(0); });

测试一下

• 当然程序不会自己停下来哈，毕竟子进程的 http 服务一直在运行！
  $ node test.js [stdout queryString] name=anan&age=29&time=1595134635733 [http response] 2020-07-19 12:57:15

本文转载自 https://www.jianshu.com/p/9ac4f9ef9625

Node.js进程管理器是一个有用的工具，可以确保Node.js进程或脚本连续（永久）运行，并使其能够在系统引导时自动启动。

它允许您监视正在运行的服务，它有助于执行常见的系统管理任务（例如重新启动失败，停止，重新加载配置而无需停机，修改环境变量/设置，显示性能指标等等）。它还支持应用程序日志记录，群集和负载平衡，以及许多其他有用的流程管理功能。

另请参阅：2019年为开发人员提供的14个最佳NodeJS框架

包管理器尤其适用于在生产环境中部署Node.js应用程序。 在本文中，我们将回顾Linux系统中Node.js应用程序管理的四个进程管理器。

1. PM2

PM2是一个开源，高级，功能丰富，跨平台和最流行的Node.js生产级流程管理器，内置负载均衡器。它允许您列出，监视和处理所有已启动的Nodejs进程，并支持群集模式。

安装PM2以在Linux中运行Nodejs应用程序

它支持应用程序监视：提供一种监视应用程序资源（内存和CPU）使用情况的简单方法。它支持您的流程管理工作流，允许您通过流程文件配置和调整每个应用程序的行为（支持的格式包括Javascript，JSON和YAML）。

应用程序日志始终是生产环境中的关键，在这方面，PM2允许您轻松管理应用程序的日志。它提供了分别处理和显示日志的不同方式和格式。您可以实时显示日志，刷新日志，并在需要时重新加载日志。

重要的是，PM2支持启动脚本，您可以将其配置为在预期或意外的计算机重新启动时自动启动进程。它还支持在当前目录或其子目录中修改文件时自动重新启动应用程序。

此外，PM2还带有一个模块系统，允许用户为Nodejs进程管理创建自定义模块。例如，您可以为日志轮换模块或负载平衡创建模块等等。

最后但同样重要的是，如果您使用Docker容器，PM2允许容器集成，并提供允许您以编程方式使用它的API系统。

2. StrongLoop PM

StrongLoop PM也是一个开源的高级生产过程管理器，用于Node.js应用程序，内置负载平衡，就像PM2一样，它可以通过命令行或图形界面使用。

用于Nodejs的StrongLoop PM进程管理器

它支持应用程序监视(查看性能指标，如事件循环时间、CPU和内存消耗)、多主机部署、集群模式、零停机应用程序重启和升级、故障时自动进程重启以及日志聚合和管理。

此外，它附带Docker支持，允许您将性能指标导出到与状态兼容的服务器，并在第三方控制台(如DataDog、石墨、Splunk以及Syslog和原始日志文件)中查看。

3. Forever

Forever是一个开源，简单且可配置的命令行界面工具，可以连续（Forever）运行给定的脚本。它适用于运行Node.js应用程序和脚本的较小部署。您可以通过两种方式永久使用：通过命令行或将其嵌入代码中。

Forever运行脚本

它允许您管理（启动，列出，停止，停止所有，重新启动，重新启动所有等等。）Node.js进程，它支持监视文件更改，调试模式，应用程序日志，终止进程和退出信号自定义等等。此外，它还支持多种使用选项，您可以直接从命令行传递或将它们传递到JSON文件中。

4. Systemd - 服务和系统管理器

在Linux中，Systemd是一个守护程序，用于管理系统资源，例如进程和文件系统的其他组件。 systemd管理的任何资源都称为一个单元。有不同类型的单元，包括服务，设备，插座，安装，目标和许多其他单元。

Systemd通过称为单元文件的配置文件管理单元。因此，为了像任何其他系统服务一样管理Node.js服务器，您需要为它创建一个单元文件，在这种情况下它将是一个服务文件。

为Node.js服务器创建服务文件后，可以启动它，启用它以在系统引导时自动启动，检查其状态，重新启动（停止并再次启动它）或重新加载其配置，甚至像任何其他系统服务一样停止它。

摘要

Node.js包管理器是在生产环境中部署项目的有用工具。它使应用程序永远存在，并简化了如何控制它。在本文中，我们回顾了Node.js的四个包管理器。如果您有任何疑问或问题，请使用下面的反馈表与我们联系。

本文转载自: https://www.jianshu.com/p/ee49e600dd16

Node.js编写组件的几种方式

本文主要备忘为Node.js编写组件的三种实现：纯js实现、v8 API实现（同步&异步）、借助swig框架实现。

关键字：Node.js、C++、v8、swig、异步、回调。

简介

首先介绍使用v8 API跟使用swig框架的不同：

（1）v8 API方式为官方提供的原生方法，功能强大而完善，缺点是需要熟悉v8 API，编写起来比较麻烦，是js强相关的，不容易支持其它脚本语言。

（2）swig为第三方支持，一个强大的组件开发工具，支持为python、lua、js等多种常见脚本语言生成C++组件包装代码，swig使用者只需要编写C++代码和swig配置文件即可开发各种脚本语言的C++组件，不需要了解各种脚本语言的组件开发框架，缺点是不支持javascript的回调，文档和demo代码不完善，使用者不多。

二、纯JS实现Node.js组件

（1）到helloworld目录下执行npm init 初始化package.json，各种选项先不管，默认即可，更多package.json信息参见：https://docs.npmjs.com/files/package.json。

（2）组件的实现index.js，例如:

module.exports.Hello = function(name) {

        console.log('Hello ' + name);

}

（3）在外层目录执行：npm install ./helloworld，helloworld于是安装到了node_modules目录中。

（4）编写组件使用代码：

var m = require('helloworld');

m.Hello('zhangsan'); //输出： Hello zhangsan

三、 使用v8 API实现JS组件——同步模式

（1）编写binding.gyp， eg：

    { "target_name": "hello", "sources": [ "hello.cpp" ]

    }

  ]

}

关于binding.gyp的更多信息参见：https://github.com/nodejs/node-gyp

（2）编写组件的实现hello.cpp，eg:

#include <node.h>



namespace cpphello { using v8::FunctionCallbackInfo; using v8::Isolate; using v8::Local; using v8::Object; using v8::String; using v8::Value; void Foo(const FunctionCallbackInfo<Value>& args) {

        Isolate* isolate = args.GetIsolate();

        args.GetReturnValue().Set(String::NewFromUtf8(isolate, "Hello World"));

    } void Init(Local<Object> exports) {

        NODE_SET_METHOD(exports, "foo", Foo);

    }



    NODE_MODULE(cpphello, Init)

}

（3）编译组件

node-gyp configure

node-gyp build

./build/Release/目录下会生成hello.node模块。

（4）编写测试js代码

const m = require('./build/Release/hello')

console.log(m.foo());  //输出 Hello World

（5）增加package.json 用于安装 eg：

{ "name": "hello", "version": "1.0.0", "description": "", "main": "index.js", "scripts": { "test": "node test.js" }, "author": "", "license": "ISC" }

（5）安装组件到node_modules

进入到组件目录的上级目录，执行:npm install ./helloc //注：helloc为组件目录

会在当前目录下的node_modules目录下安装hello模块，测试代码这样子写：

var m = require('hello');

console.log(m.foo());

四、 使用v8 API实现JS组件——异步模式

上面三的demo描述的是同步组件，foo()是一个同步函数，也就是foo()函数的调用者需要等待foo()函数执行完才能往下走，当foo()函数是一个有IO耗时操作的函数时，异步的foo()函数可以减少阻塞等待，提高整体性能。

异步组件的实现只需要关注libuv的uv_queue_work API，组件实现时，除了主体代码hello.cpp和组件使用者代码，其它部分都与上面三的demo一致。

hello.cpp：

/* * Node.js cpp Addons demo: async call and call back.

* gcc 4.8.2

* author:cswuyg

* Date:2016.02.22

* */ #include <iostream> #include <node.h> #include <uv.h> #include <sstream> #include <unistd.h> #include <pthread.h>



namespace cpphello { using v8::FunctionCallbackInfo; using v8::Function; using v8::Isolate; using v8::Local; using v8::Object; using v8::Value; using v8::Exception; using v8::Persistent; using v8::HandleScope; using v8::Integer; using v8::String; // async task

    struct MyTask{

        uv_work_t work; int a{0}; int b{0}; int output{0};

        unsigned long long work_tid{0};

        unsigned long long main_tid{0};

        Persistent<Function> callback;

    }; // async function

    void query_async(uv_work_t* work) {

        MyTask* task = (MyTask*)work->data;

        task->output = task->a + task->b;

        task->work_tid = pthread_self();

        usleep(1000 * 1000 * 1); // 1 second

 } // async complete callback

    void query_finish(uv_work_t* work, int status) {

        Isolate* isolate = Isolate::GetCurrent();

        HandleScope handle_scope(isolate);

        MyTask* task = (MyTask*)work->data; const unsigned int argc = 3;

        std::stringstream stream;

        stream << task->main_tid;

        std::string main_tid_s{stream.str()};

        stream.str("");

        stream << task->work_tid;

        std::string work_tid_s{stream.str()};



        Local<Value> argv[argc] = {

            Integer::New(isolate, task->output), 

            String::NewFromUtf8(isolate, main_tid_s.c_str()),

            String::NewFromUtf8(isolate, work_tid_s.c_str())

        };

        Local<Function>::New(isolate, task->callback)->Call(isolate->GetCurrentContext()->Global(), argc, argv);

        task->callback.Reset(); delete task;

    } // async main

    void async_foo(const FunctionCallbackInfo<Value>& args) {

        Isolate* isolate = args.GetIsolate();

        HandleScope handle_scope(isolate); if (args.Length() != 3) {

            isolate->ThrowException(Exception::TypeError(String::NewFromUtf8(isolate, "arguments num : 3"))); return;

        } if (!args[0]->IsNumber() || !args[1]->IsNumber() || !args[2]->IsFunction()) {

            isolate->ThrowException(Exception::TypeError(String::NewFromUtf8(isolate, "arguments error"))); return;

        }

        MyTask* my_task = new MyTask;

        my_task->a = args[0]->ToInteger()->Value();

        my_task->b = args[1]->ToInteger()->Value();

        my_task->callback.Reset(isolate, Local<Function>::Cast(args[2]));

        my_task->work.data = my_task;

        my_task->main_tid = pthread_self();

        uv_loop_t *loop = uv_default_loop();

        uv_queue_work(loop, &my_task->work, query_async, query_finish); 

    } void Init(Local<Object> exports) {

        NODE_SET_METHOD(exports, "foo", async_foo);

    }



    NODE_MODULE(cpphello, Init)

}