免费可以测试的vps服务器资源?

xiaodu2017 回复了问题 • 6 人关注 • 10 个回复 • 997 次浏览 • 3 天前 • 来自相关话题

OK Log设计思路

cdh0805010118 发表了文章 • 4 个评论 • 201 次浏览 • 3 天前 • 来自相关话题

OK Log 姊妹篇

设计

在这个文档中,我们首先在顶层设计上描述这个系统。然后,我们再引入约束和不变量来确定问题域。我们会一步步地提出一... 查看全部

OK Log 姊妹篇


设计


在这个文档中,我们首先在顶层设计上描述这个系统。然后,我们再引入约束和不变量来确定问题域。我们会一步步地提出一个具体的解决方案,描述框架中的关键组件和组件之间的行为。


生产者与消费者


我们有一个大且动态地生产者集,它们会生产大量的日志记录流。这些记录应该可供消费者查找到的。


     +-----------+
P -> | |
P -> | ? | -> C
P -> | |
+-----------+

生产者主要关心日志被消费的速度尽可能地快。如果这个速度没有控制好,有一些策略可以提供,包括:背压策略(ps: 流速控制), 例如:事件日志、缓冲和数据丢弃(例如:应用程序日志)。在这些情况下,接收日志记录流的组件需要优化顺序写操作。


消费者主要关心尽快地响应用户端的日志查询,保证尽可能快的日志持久化。因为我们定义了查询必须带时间边界条件,我们要确保我们可以通过时间分隔数据文件,来解决grep问题。所以存储在磁盘上的最终数据格式,应该是一个按照时间划分的数据文件格式,且这些文件内的数据是由所有生产者的日志记录流全局归并得到的。如下图所示:


     +-------------------+
P -> | R |
P -> | R ? R R R | -> C
P -> | R |
+-------------------+

设计细节


我们有上千个有序的生产者。(一个生产者是由一个应用进程,和一个forward代理构成)。我们的日志系统有必要比要服务的生产系统小得多。因此我们会有多个ingest节点,每个ingest节点需要处理来自多个生产者的写请求。


我们也想要服务于有大量日志产生的生产系统。因此,我们不会对数据量做还原性假设。我们假设即使是最小工作集的日志数据,对单个节点的存储可能也是太大的。因此,消费者将必须通过查询多个节点获取结果。这意味着最终的时间分区的数据集将是分布式的,并且是复制的。


producers --> forwarders --> ingester ---> **storage** <--- querying  <--- consumer

+---+ +---+
P -> F -> | I | | Q | --.
P -> F -> | | +---+ |
+---+ +---+ '->
+---+ ? | Q | ----> C
P -> F -> | I | +---+ .->
P -> F -> | | +---+ |
P -> F -> | | | Q | --'
+---+ +---+

现在我们引入分布式,这意味着我们必须解决协同问题。


协同


协同是分布式系统的死亡之吻。(协同主要是解决分布式数据的一致性问题)。我们的日志系统是无协同的。让我们看看每个阶段需要什么。


生产者,更准确地说,forwarders,需要能够连接任何一个ingest节点,并且发送日志记录。这些日志记录直接持久化到ingester所在的磁盘上,并尽可能地减少中间处理过程。如果ingester节点挂掉了,它的forwarders应该非常简单地连接其他ingester节点和恢复日志传输。(根据系统配置,在传输期间,它们可以提供背压,缓冲和丢弃日志记录)言外之意,forwarders节点不需要知道哪个ingest是ok的。任何ingester节点也必须是这样。


有一个优化点是,高负载的ingesters节点可以把负载(连接数)转移到其他的ingesters节点。有三种方式:、



  • ingesters节点通过gossip协议传递负载信息给其他的ingesters节点,这些负载信息包括:连接数、IOps(I/O per second)等。

  • 然后高负载ingesters节点可以拒绝新连接请求,这样forwarders会重定向到其他比较轻量级负载的ingesters节点上。

  • 满负载的ingesters节点,如果需要的话,甚至可以中断已经存在的连接。但是这个要十分注意,避免错误的拒绝合理的服务请求。


例如:在一个特定时间内,不应该有许多ingesters节点拒绝连接。也就是说日志系统不能同时有N个节点拒绝forwarders节点日志传输请求。这个可以在系统中进行参数配置。


consumers需要能够在没有任何时间分区和副本分配等条件的情况下进行查询。没有这些已知条件,这意味着用户的一个查询总是要分散到每个query节点上,然后聚合和去重。query节点可能会在任何时刻挂掉,启动或者所在磁盘数据空。因此查询操作必须优雅地管理部分结果。


另一个优化点是,consumers能够执行读修复。一个查询应该返回每一个匹配的N个备份数据记录,这个N是复制因子。任何日志记录少于N个备份都是需要读修复的。一个新的日志记录段会被创建并且会复制到集群中。更进一步地优化,独立的进程能够执行时空范围内的顺序查询,如果发现查询结果存在不一致,可以立即进行读修复。


在ingest层和query层之间的数据传输也需要注意。理想情况下,任何ingest节点应该能够把段传送到任何查询节点上。我们必须优雅地从传输失败中恢复。例如:在事务任何阶段的网络分区。


让我们现在观察怎么样从ingest层把数据安全地传送到query层。


ingest段


ingesters节点从N个forwarders节点接收了N个独立的日志记录流。每个日志记录以带有ULID的字符串开头。每个日志记录有一个合理精度的时间错是非常重要的,它创建了一个全局有序,且唯一的ID。但是时钟全局同步是不重要的,或者说记录是严格线性增长的。如果在一个很小的时间窗口内日志记录同时到达出现了ID乱序,只要这个顺序是稳定的,也没有什么大问题。


到达的日志记录被写到一个活跃段中,在磁盘上这个活跃段是一个文件。


          +---+
P -> F -> | I | -> Active: R R R...
P -> F -> | |
P -> F -> | |
+---+

一旦这个段文件达到了B个字节,或者这个段活跃了S秒,那么这个活跃段就会被flush到磁盘上。(ps: 时间限制或者size大小)


          +---+
P -> F -> | I | -> Active: R R R...
P -> F -> | | Flushed: R R R R R R R R R
P -> F -> | | Flushed: R R R R R R R R
+---+

这个ingester从每个forwarder连接中顺序消费日志记录。当当前的日志记录成功写入到活跃的段中后,下一个日志记录将会被消费。并且这个活跃段在flush后立即同步复制备份。这是默认的持久化模式,暂定为fast。


Producers选择性地连接一个独立的端口上,其处理程序将在写入每个记录后同步活跃的段。者提供了更强的持久化,但是以牺牲吞吐量为代价。这是一个独立的耐用模式,暂时定为持久化。(ps: 这段话翻译有点怪怪的,下面是原文)


Producers can optionally connect to a separate port, whose handler will sync the active segment after each record is written. This provides stronger durability, at the expense of throughput. This is a separate durability mode, tentatively called durable.


第三个更高级的持久化模式,暂定为混合模式。forwarders一次写入整个段文件到ingester节点中。每一个段文件只有在存储节点成功复制后才能被确认。然后这个forwarder节点才可以发送下一个完整的段。


ingesters节点提供了一个api,用于服务已flushed的段文件。



  • Get /next ---- 返回最老的flushed段,并将其标记为挂起

  • POST /commit?id=ID ---- 删除一个挂起的段

  • POST /failed?id=ID ---- 返回一个已flushed的挂起段


ps: 上面的ID是指:ingest节点的ID


段状态由文件的扩展名控制,我们利用文件系统进行原子重命名操作。这些状态包括:.active、.flushed或者.pending, 并且每个连接的forwarder节点每次只有一个活跃段。


          +---+                     
P -> F -> | I | Active +---+
P -> F -> | | Active | Q | --.
| | Flushed +---+ |
+---+ +---+ '->
+---+ ? | Q | ----> C
P -> F -> | I | Active +---+ .->
P -> F -> | | Active +---+ |
P -> F -> | | Active | Q | --'
| | Flushed +---+
| | Flushed
+---+

观察到,ingester节点是有状态的,因此它们需要一个优雅地关闭进程。有三点:



  • 首先,它们应该中断链接和关闭监听者

  • 然后,它们应该等待所有flushed段被消费

  • 最后,它们才可以完成关闭操作


消费段


这个ingesters节点充当一个队列,将记录缓冲到称为段的组中。虽然这些段有缓冲区保护,但是如果发生断电故障,这内存中的段数据没有写入到磁盘文件中。所以我们需要尽快地将段数据传送到query层,存储到磁盘文件中。在这里,我们从Prometheus的手册中看到,我们使用了拉模式。query节点从ingester节点中拉取已经flushed段,而不是ingester节点把flushed段推送到query节点上。这能够使这个设计模型提高其吞吐量。为了接受一个更高的ingest速率,更加更多的ingest节点,用更快的磁盘。如果ingest节点正在备份,增加更多的查询节点一共它们使用。


query节点消费分为三个阶段:



  • 第一个阶段是读阶段。每一个query节点定期地通过GET /next, 从每一个intest节点获取最老的flushed段。(算法可以是随机选取、轮询或者更复杂的算法,目前方案采用的是随机选取)。query节点接收的段逐条读取,然后再归并到一个新的段文件中。这个过程是重复的,query节点从ingest层消费多个活跃段,然后归并它们到一个新的段中。一旦这个新段达到B个字节或者S秒,这个活跃段将被写入到磁盘文件上然后关闭。

  • 第二个阶段是复制阶段。复制意味着写这个新的段到N个独立的query节点上。(N是复制因子)。这是我们仅仅通过POST方法发送这个段到N个随机存储节点的复制端点。一旦我们把新段复制到了N个节点后,这个段就被确认复制完成。

  • 第三个阶段是提交阶段。这个query节点通过POST /commit方法,提交来自所有ingest节点的原始段。如果这个新的段因为任何原因复制失败,这个query节点通过POST /failed方法,把所有的原始段全部改为失败状态。无论哪种情况,这三个阶段都完成了,这个query节点又可以开始循环随机获取ingest节点的活跃段了。


下面是query节点三个阶段的事务图:


Q1        I1  I2  I3
-- -- -- --
|-Next--->| | |
|-Next------->| |
|-Next----------->|
|<-S1-----| | |
|<-S2---------| |
|<-S3-------------|
|
|--.
| | S1∪S2∪S3 = S4 Q2 Q3
|<-' -- --
|-S4------------------>| |
|-S4---------------------->|
|<-OK------------------| |
|<-OK----------------------|
|
| I1 I2 I3
| -- -- --
|-Commit->| | |
|-Commit----->| |
|-Commit--------->|
|<-OK-----| | |
|<-OK---------| |
|<-OK-------------|

让我们现在考虑每一个阶段的失败处理



  • 对于第一个阶段:读阶段失败。挂起的段一直到超时都处于闲置状态。对于另一个query节点,ingest节点的活跃段是可以获取的。如果原来的query节点永远挂掉了,这是没有任何问题的。如果原始的query节点又活过来了,它有可能仍然会消费已经被其他query节点消费和复制的段。在这种情况下,重复的记录将会写入到query层,并且一个或者多个会提交失败。如果这个发生了 ,这也ok:记录超过了复制因子,但是它会在读时刻去重,并且最终会重新合并。因此提交失败应该被注意,但是也能够被安全地忽略。

  • 对于第二个阶段:复制阶段。错误的处理流程也是相似的。假设这个query节点没有活过来,挂起的ingest段将会超时并且被其他query节点重试。如果这个query节点活过来了,复制将会继续进行而不会失败,并且一个或者多个最终提交将将失败

  • 对于第三个阶段:commit阶段。如果ingest节点等待query节点commit发生超时,则处在pending阶段的一个或者多个ingest节点,会再次flushed到段中。和上面一样,记录将会重复,在读取时进行数据去重,然后合并。


节点失败


如果一个ingest节点永久挂掉,在其上的所有段记录都会丢失。为了防止这种事情的发生,客户端应该使用混合模式。在段文件被复制到存储层之前,ingest节点都不会继续写操作。


如果一个存储节点永久挂掉,只要有N-1个其他节点存在都是安全的。但是必须要进行读修复,把该节点丢失的所有段文件全部重新写入到新的存储节点上。一个特别的时空追踪进行会执行这个修复操作。它理论上可以从最开始进行读修复,但是这是不必要的,它只需要修复挂掉的段文件就ok了。


查询索引


所有的查询都是带时间边界的,所有段都是按照时间顺序写入。但是增加一个索引对找个时间范围内的匹配段也是非常必要的。不管查询节点以任何理由写入一个段,它都需要首先读取这个段的第一个ULID和最后一个ULID。然后更新内存索引,使这个段携带时间边界。在这里,一个线段树是一个非常好的数据结构。


另一种方法是,把每一个段文件命名为FROM-TO,FROM表示该段中ULID的最小值,TO表示该段中ULID的最大值。然后给定一个带时间边界的查询,返回所有与时间边界有叠加的段文件列表。给定两个范围(A, B)和(C, D),如果A<=B, C<=D以及A<=C的话。(A, B)是查询的时间边界条件,(C, D)是一个给定的段文件。然后进行范围叠加,如果B>=C的话,结果就是FROM C TO B的段结果


A--B         B >= C?
C--D yes

A--B B >= C?
C--D no

A-----B B >= C?
C-D yes

A-B B >= C?
C----D yes

这就给了我们两种方法带时间边界的查询设计方法


合并


合并有两个目的:



  • 记录去重

  • 段去叠加


在上面三个阶段出现有失败的情况,例如:网络故障(在分布式协同里,叫脑裂),会出现日志记录重复。但是段会定期透明地叠加。


在一个给定的查询节点,考虑到三个段文件的叠加。如下图所示:


t0             t1
+-------+ |
| A | |
+-------+ |
| +---------+ |
| | B | |
| +---------+ |
| +---------+
| | C |
| +---------+

合并分为三步:



  • 首先在内存中把这些重叠的段归并成一个新的聚合段。

  • 在归并期间,通过ULID来进行日志记录去重和丢弃。

  • 最后,合并再把新的聚合段分割成多个size的段,生成新的不重叠的段文件列表


t0             t1
+-------+-------+
| | |
| D | E |
| | |
+-------+-------+

合并减少了查询搜索段的数量。在理想情况下,每次都会且只映射到一个段。这是通过减少读数量来提高查询性能。


观察到合并能改善查询性能,而且也不会影响正确性和空间利用率。在上述合并处理过程中同时使用压缩算法进行合并后的数据压缩。合适的压缩可以使得日志记录段能够在磁盘保留更长的时间(ps: 因为可以使用的空间更多了,磁盘也没那么快达到设置的上限),但是会消耗衡更多的CPU。它也可能会使UNIX/LINUX上的grep服务无法使用,但是这可能是不重要的。


由于日志记录是可以单独寻址的,因此查询过程中的日志记录去重会在每个记录上进行。映射到段的记录可以在每个节点完全独立优化,无需协同。


合并的调度和耦合性也是一个非常重要的性能考虑点。在合并期间,单个合并groutine会按照顺序执行每个合并任务。它每秒最多进行一次合并。更多的性能分析和实际研究是非常必要的。


查询


每个查询节点提供一个GET /query的api服务。用户可以使用任意的query节点提供的查询服务。系统受到用户的查询请求后,会在query层的每一个节点上进行查询。然后每个节点返回响应的数据,在query层进行数据归并和去重,并最终返回给用户。


真正的查询工作是由每个查询节点独立完成的。这里分为三步:



  • 首先匹配查询时间边界条件的段文件被标记。(时间边界条件匹配)

  • 对于第一步获取的所有段,都有一个reader进行段文件查找匹配的日志记录,获取日志记录列表

  • 最后对获取到的日志记录列表通过归并Reader进行归并,排序,并返回给查询节点。


这个pipeline是由很多的io.ReaderClosers构建的,主要开销在读取操作。这个HTTP响应会返回给查询节点,最后返回给用户。


注意一点,这里的每个段reader都是一个goroutine,并且reading/filtering是并发的。当前读取段文件列表还进行goroutine数量的限制。(ps: 有多少个段文件,就会生成相应数量的goroutine)。这个是应该要优化的。


用户查询请求包括四个字段:



  • FROM, TO time.Time - 查询的时间边界

  • Q字符串 - 对于grep来说,空字符串是匹配所有的记录

  • Regex布尔值 - 如果是true,则进行正则表达式匹配

  • StatsOnly布尔值 - 如果是true,只返回统计结果


用户查询结果响应有以下几个字段:



  • NodeCount整型 - 查询节点参与的数量

  • SegmentCount整型 - 参与读的段文件数量

  • Size整型 - 响应结果中段文件的size

  • io.Reader的数据对象 - 归并且排序后的数据流


StatsOnly可以用来探索和迭代查询,直到它被缩小到一个可用的结果集


组件模型


下面是日志管理系统的各个组件设计草案


进程


forward


  • ./my_application | forward ingest.mycorp.local:7651

  • 应该接受多个ingest节点host:ports的段拉取

  • 应该包含DNS解析到单个实例的特性

  • 应该包含在连接断掉后进行容错的特性

  • 能够有选择fast, durable和chunked写的特性

  • Post-MVP: 更复杂的HTTP? forward/ingest协议;


ingest


  • 可以接收来自多个forwarders节点的写请求

  • 每条日志记录以\n符号分割

  • 每条日志记录的前缀必须是ULID开头

  • 把日志记录追加到活跃段中

  • 当活跃段达到时间限制或者size时,需要flush到磁盘上

  • 为存储层的所有节点提供轮询的段api服务

  • ingest节点之间通过Gossip协议共享负载统计数据

  • Post-MVP: 负载扩展/脱落;分段到存储层的流传输


store


  • 轮询ingest层的所有flush段

  • 把ingest段归并到一起

  • 复制归并后的段到其他存储节点上

  • 为客户端提供查询API服务

  • 在某个时刻执行合并操作

  • Post-MVP:来自ingest层的流式段合并;提供更高级的查询条件


Libraries


Ingest日志


  • 在ingest层的段Abstraction

  • 主要操作包括:创建活跃段,flush、pending标记,和提交

  • (I've got a reasonable prototype for this one) (ps: 不明白)

  • 请注意,这实际上是一个磁盘备份队列,有时间期限的持久化存储


Store日志


  • 在storage层的段Abstraction

  • 操作包括段收集、归并、复制和合并

  • 注意这个是长期持久化存储


集群



  • 来之各个节点之间的信息的Abstraction

  • 大量的数据共享通信是不必要的,只需要获取节点身份和健康检查信息就足够了

  • HashiCorp's memberlist fits the bill (ps:不明白)

结合 Go 读 APUE-基本文件I/O

suc 回复了问题 • 4 人关注 • 3 个回复 • 400 次浏览 • 3 天前 • 来自相关话题

OK Log

cdh0805010118 发表了文章 • 2 个评论 • 210 次浏览 • 4 天前 • 来自相关话题

OK Log姊妹篇


OK Log


OK Log是一个用于大规模集群的分布式且无协的日志管理系统。我是从一些最基本的原则考虑这个系统的设计的。下面介绍的就是这个原型的思路来源。


绪论


过去的一两年时间,我受邀参加很多关于微服务、Go和Go kit的演讲和研讨会议。选一个微服务架构,意味着要对很多考虑点进行技术选型。如果可能的话,对一些新兴的中等规模系统,我愿意给出一些技术指南。开源社区的项目是非常丰富的。



  • 服务编排? 有Kubernetesnomad、DC/OS、ECS等,有很多服务编排工具,都是很好的选择。(ps:目前docker和Kubernetes深度合作了,Mesos可能要被边缘化了)。

  • 服务发现? ConsulEtcdZookeeper动态服务发现等工具,也有静态注册和服务发现工具Linkerd

  • 分布式调用链跟踪?ZipkinJaeger、Appdash、Lightstep等,它还在爆发式增长.

  • 监控工具?Prometheus, 它是目前最强的监控工具、InfluxDB等,它们结合Grafana工具使用

  • 日志?我陷入了沉思....


很明确的答案似乎是Elastic和ELK技术栈。确实,它很有特点、且入手很容易。但是Elastic被很多人认为,对于中等规模的集群,都很难操作。同时我相信,在全文、基于文档搜索时,Lucene或许不是最好的的数据存储格式。最终,我了解了很多使用Elastic的朋友,由于操作的难度很高,他们中的大多数都不怎么乐意使用它。几乎很少有人使用更高级的特性。


更美好的事物


我认为,对于日志管理系统,应该有一个更好的答案。我问了一些同事,他们正在着手的解决方案。一些同事实际上采用了Kafka消息队列解决日志系统管理,特别是对于高QOS和持久化日志要求。但是它的操作也相当难,且最终设计成的日志管理系统,和我感兴趣要解决的问题也不相同。其他人通过数据仓库HBase来解决。但是管理一个Hadoop集群需要更加专业化的只是和非凡的努力。对于这些方案的选择,我认为具体化的或者比较重的系统设计都是一个好的建议。


我还在Twitter上提出了这个问题。Heka似乎是最接近我需要的,但是因为作者前期设计错误,导致了16年年底遇到了无法修复的性能问题,已经放弃了Heka的维护,这是一件非常糟糕的事情。Ekanite提供了端到端的解决方案,但是它的系统日志协议与微服务的工作负载有很明显的不匹配。对于日志传送和注解有非常好的工具,例如:FluentdLogstash,但是它们只能解决部分问题;它们不能处理存储和日志查询。委托解决方案的工具,有SplunkLoggly,如果你的日志是低容量,且不介意把日志上传到云端,这两个工具都是很好的选择,但是它们很快变得昂贵,且无法再本地和开放源代码框中打勾。(ps: 这句话不是很明白)。


Prometheus日志


我意识到我需要的是Prometheus日志的设计原则。什么意思呢?Prometheus好的地方有什么呢?我的观点:



  • 独立运行:它既是开源的、又可以在本地部署

  • 云原生的工作负载:动态的、容器化的和微服务的水平扩展. (ps: 链接中的解释我是非常满意的,是不是就是Serverless)

  • 容易操作:本地存储、没有集群、拉模式

  • 完善的系统:不需要独立的TSDB(时间序列数据库)、web UI等,容易使用

  • 系统扩容:90%的用户承认使用很小的成本,就可以获取比较高的满意度


那Prometheus日志是什么样子的呢?我希望冬天把这个日志管理系统设计完成,我认为这是非常有趣的,同时我也可以学到很多的知识。首先我需要思考得更加深入。


设计


高层次目标


首先,像Prometheus一样,系统应该是开源的,且支持本地部署。更重要的是,它应该很容易部署和水平扩展。它应该更加关注容器化的微服务工作负载。同时他应该是一个完善的,端到端的系统,有forwarders、ingesters、storages和query四个特性。


这个日志管理系统关注点:



  • 微服务的应用程序日志,包括:debug、info、warn等各种级别日志。这个是典型的高容量、低QOS日志,但是对延时(查询时间)有较高的要求。

  • 我们也想服务于事件日志,包括:审计跟踪和点击跟踪等等。这是典型的低容量,搞QOS,但是对延时(查询时间)没有较高的要求。

  • 最后,它应该有一个统一的日志消费者,管理来自黑盒的日志输出,例如:mysql服务。也就是说,我们不会控制日志格式。
    我相信这样的系统可以服务于所有的需求,同时扩展性也非常好。


心里有了这些目标,我们就需要开始增加一些约束,有了边界才能使问题更加容易处理,关注点更加集中。


问题约束


宝贵的经验告诉我,数据系统应该更多地关注数据传输,同时增加数据的价值。这就是说:



  • 它是一个数据运输系统,解决更多的机械问题,黑盒运输

  • 它也应该是一个应用系统,提供商业价值,对拓扑和性能要求不需要参与


如果尝试用一个方案解决这两个问题,会造成竞争和一定的妥协。所以我比较感兴趣数据传输系统,旨在解决低吞吐率和延时问题。我们可以使用其他的工具,在系统外部增加数据的商业价值。例如:上下文context可以在ingest之前发生。或者,解析日志再聚合可以在ETLs(数据仓库技术)中完成。然后再使用更加丰富的查询功能的数据系统将其结果视图化。


考虑到这一点,基于时间边界的grep查询接口是完全可接受的。对于新用户,他们经常想要一个熟悉的接口来帮助他们调试-“我想要grep我的日志”,这是非常有用的。构建ETLs(数据仓库技术)到更复杂的系统中是完全足够的。总之,这个日志管理系统是一个基本的、底层系统,它可以和其他工具搭配使用,至于搭配什么样的工具,主要看你自己的需求。(ps: 类似于系统插件化)


分布式系统


去年在旧金山的Prometheus见面会上,Julius Volz观察到日志数据比监控数据要大几个数量级。Prometheus安装的大多数节点日志已经超出了ingest和单节点容量限制。因此,与Prometheus相同的日志系统必须是一个分布式系统。这个复杂度是根本性的、不可避免的。那好,我们就着手解决它。


无协同


到目前为止,我们最重要的目标是系统容易操作。并且从Prometheus中,我们学习到它应该是平滑水平扩展的,从测试环境到生产环境,没有重大结构变化。在简单和复杂的系统设计中做出合适的权衡是非常痛苦的。但是我强烈建议无协同系统。无协同意味着放弃了其他软件系统的一些优秀特性,如:Elastic、Kafka和Canssandra等。没有master选举、没有节点分配、没有分区表、没有分布式索引、没有vnodes。承认暂停、分袂和死节点是这个设计的一部分。某种意义上来说,这些会使系统设计更加困难,我们使用很少的技术做支撑,所以需要花更多的时间做前期设计。但是另一方面,它更容易实现,因为无协同组件往往更简单,更容易实现。


我们可以看到,如果我们能够设计一个组件模型是无协同的,那么我会充分思考系统的设计


写的重要性


在我们开始设计之前,一个重要的观察:在日志管理系统中,写需求是强烈的,读需求可以等待。(ps: 因为写阻塞,会影响业务系统的性能)。所以,我认为最重要的运行时挑战是写高吞吐量。理想情况下,它无限接近到硬件速度————这也有助于日志管理系统的节点容量扩展。


首先日志系统的总体设计。Agents把日志记录从容器forward到ingesters中。这个ingesters应该执行快速的序列化写操作,把日志记录写入到一个活跃的段文件中,这个ingesters的任务就完成了。让存储节点更关心读操作的优化。
ingesters


Ingestion


由于我们对ingestion和querying有不同的性能要求,那么分离这些组件是非常有意义的。它们是细粒度的节点安装。我们也可以通过一个编译过的二进制文件安装,这样更加容易。


在写操作的时候,每条日志记录被ingester赋予一个全局唯一的ID。这使得很多特性的实现变成了可能,例如:多次相同日志记录的去重。唯一的ID生成有很多方式,例如:UUID或者ULID, 还有twitter提出的64位byte[毫秒数:业务线:机房:机器:预留:毫秒内序列号],这些都非常好。有一点非常重要,每条日志记录有个合理精度的时间戳去创建一个全局唯一的ID;但是有一点不重要的,时钟是全局同步的,或者这些日志记录是严格的线性递增。同时,我认为如果在同一个相同的时间窗口内出现了乱序,只要顺序是稳定的,这也是ok的。ULIDs曾明能够在50ns内生成ID有序,它可以很好的工作。


为了满足日志持久化要求,这里有不同的持久化模式。



  • 如果我们主要关心吞吐量,例如:应用程序的日志,我们可以使用fast模式。写入一个文件描述符而不需要直接同步;

  • 对于事件日志,有一个持久化模式,我们定期地同步活跃段到磁盘上;

  • 最后一种批处理模式:许多客户端同时写整个段文件,只有当它完全复制到其他节点时才被确认。(这个是从Kafka获取的灵感)


这样,我们的日志管理系统的组件模型慢慢的变成了下面这幅图:
Ingesters


我们可以对协同的讨论思考得更多一点。如果我们编排它,以至于任何ingest节点都能够服务于任何forwarder传送的日志记录,这样我们避免了forwarders需要知道超出ingesters的地址信息。forwarders是在ingesters池中任意拿一个进行连接,并且能够实现反压(backpressure)、缓冲和重连逻辑等。依赖于其他下游服务。有了这些限制,ingesters可以不受约束。到目前为止这种方式还挺好的。


日志复制


日志记录写入磁盘是不安全的,一旦落到磁盘,日志记录就需要被复制备份。我们知道数据需要多节点存储。


来自Prometheus的日志设计思路,我们把典型的日志复制有push模式改为了pull模式。准确地说,一个集群中的所有ingest和存储节点,需要通过gossip协议通信。所有的存储节点定期地和随机地消费来自所有ingest节点的段文件。消费后的段需要合并,同时合并之后的文件达到一定的size后,就会复制到其他存储节点上进行备份。只有成功复制之后,原始段从ingest节点中确认和清除。


replication图解


实际上,我们把每一个ingest节点变成了一个后端磁盘队列。并且每一个存储节点获得了整个日志的子集,并且密度由复制因子决定。


为什么要把ingest节点数据移动到存储节点上呢?



  • 在较小的场景中,低频的读操作和写操作负载可能这是没有什么作用。例如:本地测试,我们会提供ingest+存储的混合节点

  • 在一些较大的场景中,I/O可能是最主要的瓶颈,并且ingest工作负载(顺序写)与存储工作负载(半随机读和写)是竞态的。隔离是很聪明的做法。


在复制事务期间,任何失败(或者超时)都会造成事务的中断,并且ingester段将会在后面重新消费。这回造成重复日志记录,但是这是ok的。因为查询时结果通过ULIDs是去重的。最终,我们至少交付了一次。这种复制形式是事务的,但是没有协同。


弹性


注意到,ingest层实际上是一个分布式的,磁盘存储的日志记录队列。我们能够扩展ingesters来处理我们的写容量。同时我们也能扩展存储层来处理我们的复制因子,设置日志有效期,以及读容量要求。


增加节点到每一层,就像让他们加入集群并开始工作一样简单。有一个优化,ingest节点能够通过gossip协议扩散负载信息,并通过增减来平衡节点负载。存储节点自动地开始消费来自多个ingest节点的段共同平分的份额。只要ingest段size小于存储段size,就可以立即平衡写负载。磁盘利用率在保留时间范围内保持平衡。所有这些都没有明确的成员注册,芈月空间声明或者任何形式的公式。集群的增长或者削减都是无协同的。


合并


存储节点最终累计不同size和时间范围的段文件。合并是对日志记录的清洗、合并和重新分割的过程,目的是统一数据格式存储和优化查询。


compaction


合并能够merge段的叠加,如上图所示,合并小的、序列化的段。在每一个阶段,它可以炒作统一的段文件数据格式一步步进化,这就是我们想要的时间边界查询。同时,合并agent能够用于强制保留期。观察数据集保持不变,只有磁盘上的布局优化。合并的影响是透明的,本地的,所以无协同。


查询


查询字面上是时间边界grep。我们分散查询到所有的查询节点上,然后聚集数据,返回合并后且日志去重的记录给用户。每一个ULIDs日志记录为去重的日志记录提供可排序的身份ID。通过从较少的节点读取,可以提交效率吗?Yes。but that would involve prior knowledge about segment location/allocation, which requires some form of coördination. We deliberately make the read path dumb, and pay some costs of inefficiency, to keep it coördination-free., (ps: 这段话不明白)


原型设计


实现


在几个朋友的帮助下,我逐渐详细地描述了系统设计。这给我带来了很大的乐趣。设计无协同的分布式系统是人生中一个非常大的乐趣。经过几周时间的努力,我开始说服自己,设计方案是可行的。经过整个假期,我开始了一个设计代码实现。经过一周或者更久,我有了一个看似正确且有用的原型。然后开始花时间进行压力测试。


验证


现在我将会描述验证的过程,并且通过连续的系统负载测试来分析系统性能。这个测试环境是由DigitalOcean提供的,在此感谢他们!


我创建了8个forwarder节点集合,3个ingester节点和3个存储节点。我开始从一些基本的正确性和crash测试入手,很快就被来自每个组件的垃圾日志所淹没。重现状态时非常困难的,或者从日志垃圾邮件中得出有意义的结论。我最终删除了一大堆日志语句,并添加了很多指标。构建Prometheus表达式和图表是建立洞察力更有效的方法。最后,我仅仅在启动时记录一些运行参数,并清除错误,如:写入文件失败。我非常清楚地意识到这里的坑。


吞吐量


我想要第一件要优化的是吞吐量。为了满足我自己的好奇心,我再Twitter上做了一个调查。我对集群中的每个节点日志吞吐量非常感兴趣。这个测试结果范围值很大,从1KBps到25MBps之间变化。5MB/sec/node对于80%~90%的方案是一个比较好的目标。让我们看看典型的测试用例。


DigitalOcean磁盘显然可以达到250MBps的持续写入,在云服务中这是表现非常好的。在我自己的测试环境中,磁盘写入测试要少一些,它在150MBps上下浮动。如果我们系统设计得正确,那么150MBps就是我们的I/O性能瓶颈。因为我们每个节点的写速度控制在5MB/sec/node,则单个ingest节点能够处理写操作不阻塞的集群节点大小范围:150/5=30 ~ 250/5=50个节点。这个范围间的集群因子都是合理的。因为我们有3个ingest节点,所以写操作的速度是150MBps*3个节点=450MBps的聚合速度。


优化forwarding


这个forwarder不过是netcat而已。基本地


    // client connection, forwarder就类似于下面的作用,传送数据
conn, _ := net.Dial("tcp", ingesterAddress)
s:=bufio.NewScanner(os.Stdin)
for s.Scan() {
fmt.Fprintf(conn, "%s\n", s.Text()) // 往tcp链路中向服务端发送终端产生的数据
}

Go's bufio.Scanner在这里非常形象;产生数据后,通过tcp链路传送数据。~~whatever limits I hit, they weren’t imposed by the scanner. ~~,(ps: 不明白)。我用一些低效率地方式来生成日志记录。我观察到CPU一路飘高,吞吐率远低于预期。性能分析暴露了两个问题:




  1. 我在热循环中使用了一个time.Ticker。每一个日志行带有一个ticker


    hz:=time.Second / recordsPerSecond
    for range time.Tick(hz) {
    // 传送一条日志记录
    }

    这里有一个问题,如果你想要每秒记录1000条日志时,则每1ms阻塞一次是资源浪费的。我推荐采用批量传送日志记录,如下所示:


    var (
    recordsPerCycle = 1
    timePerCycle = time.Second / recordsPerSecond
    )
    for timePerCycle < 50*time.Millisecond {
    recordsPerCycle *= 2
    timePerCycle *= 2
    }
    for range time.Tick(timePerCycle) {
    // 每次记录recordsPerCycle条日志记录
    }


  2. 我利用随机数据在热循环中构建每一行日志,大量消耗CPU。在程序开始时,预先计算一大组固定的随机日志可以解决这个问题。通过这些变化,我可以很轻松地从每个进程推送大量的MBps,而且负载可以忽略不计。这个第二点翻译感觉很有问题, 原文如下:


Also, I was building each log line, of random data, within the hot loop, and burning lots of CPU in math.Rand to do it. Precomputing a large, fixed set of random log lines at program start solved that one. With those changes, I could easily push plenty of MBps second from each process with negligible load.


我为每个forward节点创建了1-8个forwarders,共为8个ingest节点设置了8-64个forwarders。每个进程将每秒处理100-1000条日志记录,每条日志记录有100-8000个bytes,每秒生成能力高达512MB。非常高的性能。


优化ingestion


在开头我很担心每一条记录一个全局唯一ULID造成的性能问题。但是多亏Tomás Senart写的优秀库ULID,这些代价实际上非常低,每ULID消耗50ns,则1s可以生成1000000000/50=2千万个ULID。因为我们不需要任何数据加密协议。下面是测试性能数据:


BenchmarkNew/WithoutEntropy-8    30.0 ns/op  534.06 MB/s  1 alloc/op
BenchmarkNew/WithEntropy-8 65.8 ns/op 243.01 MB/s 1 alloc/op
BenchmarkNew/WithCryptoEntropy-8 771 ns/op 20.73 MB/s 1 alloc/op

我最初能够将每个ingest实例推送到30MBps,但是事情变得很棘手。初始化性能分析揭露了在bytes.FieldsFunc和系统调用方法中,CPU不成比例的过度消耗。



  • 对于bytes.FieldFunc方法,我对比了ULIDs,前者表现令人意想不到的差劲。切换到固定大小偏移分割-ULID给了我们这个能力,改进并提升性能,但不是ingest率。

  • 对于系统调用syscalls,我怀疑是文件系统的竞争导致的。为了验证我的怀疑,我抽象了文件系统,并提供了一个无操作的实现。即使扩展到几百兆的ingest节点写入,这性能表现也非常好。事实证明,在最初的设计中,我将所有传入的连接多路复用到相同的活跃段文件中。假设将部分段传送到存储节点会是一个瓶颈,所以我的想法是优化(最小化)每个ingest节点产生的活跃部分数量


但是性能分析揭示,多路复用会花费很大的代价。因此,我分离它,把每个连接写入到自己的段文件中。这是最主要的改善。


活跃的段文件达到一定的时间或者size就会写入到存储节点,并关闭段文件。这个时间限制保证了系统日志的实时性,文件大小保持复制可管理特性。



  • 如果根据时间窗口关闭段文件,日志记录大小可能会比较低,而且事情也简单。我们能够基于想要的日志记录延时,选择你想要的时间窗口。(ps:但是这可能会造成日志持久化不实时,当日志产生后查询不到结果)。3s是默认值

  • 如果根据size大小关闭段文件,我们会有高容量,但是如果日志记录产生速度过慢,则会造成段文件在一定时间内达不到size阈值,造成上面一样的结果,不实时。


所以我们必须要进行时间和size的权衡,文件应该足够大,以便分摊复制开销,并有效利用典型的SSD,但又足够小,可以快速传输,并有效地组成存储层段文件。我选择16MB作为默认的ingest段文件大小,同时选择128MB作为存储节点段文件的默认大小。但是这些选择没有必要是统一的。他们可能会极大地影响系统吞吐量,应该要在特定的环境中进行验证。


随着ingest节点积极地消费和保存记录,下一步我们讨论存储节点。


优化复制


日志复制工作起来要像这样:一旦一个ingest节点的段,由于时间窗口或者size关闭了,那么存储节点则可以存储该段了。存储节点随机轮询ingest节点,并且获取最老可用的段(ps: 偏移位置)。它们会在内存中就合并好这些段到一个聚合的段中。这都是在一个循环中做的,直到这个聚合段达到了设置的时间窗口或者size。然后它们会复制这个聚合段到N个随机的存储节点上,N是由你的复制因子决定的。一旦复制确认完成,这个ingest节点的段将被确认和删除。


在实践中这个复制机制工作得很好。甚至当ingestion真的不平衡时,存储节点也会消化掉要复制的日志段。


ingest&store nodes


存储节点通过状态机consume段文件:gather、replicate和commit。



  • gather状态要求尽快地调整并消费这些段文件(当然,我明白轮询很糟糕,订阅方式比较好。未来我会实现这个功能)

  • replicate状态主要是采用POST方式发送聚合的段到其他存储节点。用足够大的段来分摊这个POST的成本也是很重要的。(我们也可以通过一直选择本节点作为目标节点之一来进行优化。这也是未来的工作)


这里成功的标准是看ingest节点数在队列上的深度。。换句话说,就是存储节点的消费日志的速率与ingest节点接收日志的比率。


store nodes/ingest nodes


这个比率保持在1附近上下浮动是最好的。既不会饥饿,也不会撑死。大于1,则表示存储节点的消费速度大于ingest节点的接收速度;~~maybe chewing through a backlog? ~~ (ps: 不明白)。小于1,则表示存储节点消费得不够快,最好是添加存储节点。


最初的这些设计证明是稳定可靠的。当然还有很多可以优化的地方,但是还没有发现重大问题。我很满意整个系统处理日志的速率。但是它能查询吗?


优化querying


这个是真正的考验。如果查询速度太慢,大多数用户将无法忍受。初始的查询性能测试完全达不到我的预期。现在这部分仍然还有很多可以改善的地方。


初始的grep设计师非常简单的。我们通过读取多条日志记录,然后通过多路归并算法找到匹配的段日志记录。并且我们在这个归并的输出上附加上了时间范围和查询表达式过滤的条件查询。整体查询框架图如下:


querying


Tomás注意到,在一个完美紧凑的系统中,不会有重叠的段。在这种情况下,不需要过多消耗CPU来完成全局的归并。我们可以首先读取不重叠的段,然后序列化地读取重叠段先进行一次归并。(ps: 可以这样理解,先把不重叠的段不合并,重叠的段先局部合并,然后再一次次的局部合并,最后在做一次整体合并)因此,我实现了一个MultiReader,它由普通的文件reader或者归并reader组成。具体取决于段的重叠。


优化后的querying


利用这种方式,在某些例子中可以提高50%的速率。


然后我们再收集了一些性能分析数据,显示在系统调用syscalls、日志记录的过滤管道和正则表达式匹配时的CPU消耗。我们认为显示的文件mmaping可能会提高读性能,所以我们设计的一个mmaping文件系统抽象的原型。但是和直接的文件系统做性能对比,我们没有获取显著的性能改善。


direct


mmaping


同时我也对比了过滤日志记录从磁盘读取日志的时间开销。对比图如下所示:



  • 在直接的文件系统中,读取时间开销主要在系统调用;

  • 在mmap文件系统中,读取时间开销主要在memmove上。


direct


mmaping


我发现页面缓存是非常有效的


经过一两天的思考,我意识到操作的顺序是归并然后过滤


merge


上面这幅图,有一部分CPU消耗在了不相关的段归并上。由于归并建立在一个全局有序的事务上,所以它被约束在一个CPU核上来完成。如果我们先过滤,然后再归并过滤后的段。这使得前者可以并发执行,充分利用多核特性。


优化后的查询


做了这些改变后,查询速度比之前高两倍了。CPU利用率更高了。但是阻塞分析揭示了在每个段的goroutine写入到io.Pipe中需要花费大量的等待时间。如果我们可以缓冲pipe,我们会有更好的性能提升。但不幸的是,io.Pipe没有给你一个可以缓冲的内存空间。同时,幸运的是,我们找到了djherbis/nio包,它提供了一个具有相同功能的缓冲Pipe。用一个适中的1MB缓冲区与直接用io.Pipe进行速度对比,提高了2倍多,太惊人了!!!


满意后,我们开始把注意力放在过滤时附加的正则表达式时间消耗。切换到bytes.Contains后有了合理的改善。事实证明各个点优化也是非常nice的。因此,我给查询时间定义了这个flag,只有在需要的时候才选择加入正则表达式匹配。(进一步优化,可能会使用PCRE(ps: 这个perl写的正则表达式库比其他库的速度要提高3倍)正则表达式,这是未来一段时间的工作)。


此时,我们意识到,我们当前的工作集(大约21G)超过了存储节点(8G)。如果我们可以获取更多的内存,我可以一次性加载整个工作集到页面缓存中,并希望借此解决其他的低效率问题。我们启动了一个32G DigitalOcean droplet,包括更多的CPU核来协助并发过滤。没有其他的变化就给了我们两倍的提速。


之前调优Cassandra的经验给了我们更多的想法。我们调整了I/O调度和readahead设置,这给了我们另外20%的改进。尽管在这一点上,我们已经非常接近仅仅基于节点上的内存总线。我们在4.6s内可靠的查询了20G日志记录。读取吞吐量为4.47GBps。这里可能还有额外的工作来优化磁盘访问,但是这似乎完全可以达到初始设置的标准。


版本一


现在大家使用的就是版本一!OK Log。哪还有什么工作要留给未来呢?


未来工作


我们能够,而且应该做类似于Cassandra的一些读修复。即查询结果在所有存储节点的数据都是相同的。存储节点数据不一致的日志记录目前还不能被检测、批处理和重写到存储节点中。未来合并会把他们最终存储到合适的位置。这个是issue 6


```Cassandra读修复
Cassandra读修复


客户端读取某个对象的时候,触发对该对象的一致性检查:


读取Key A的数据时,系统会读取Key A的所有数据副本,如果发现有不一致,则进行一致性修复。



  1. 如果读一致性要求为ONE,会立即返回离客户端最近的一份数据副本。然后会在后台执行Read Repair。这意味着第一次读取到的数据可能不是最新的数据;

  2. 如果读一致性要求为QUORUM,则会在读取超过半数的一致性的副本后返回一份副本给客户端,剩余节点的一致性检查和修复则在后台执行;
    3.如果读一致性要求高(ALL),则只有Read Repair完成后才能返回一致性的一份数据副本给客户端。可见,该机制有利于减少最终一致的时间窗口。


相关地,在数据丢失之前,整个系统能够容忍N-1个存储节点挂掉,但是如果一个存储节点挂掉了,没有其他修复进程拉起和修复数据的话,我们会进行服务降级。一个修复进程监听每个存储节点,当节点挂掉后会立即启动进程修复,它会顺序地查询每条日志记录,触发对尚未完全复制的日志记录进行读取修复。这个是issue 11


添加一个ingest节点对系统来说是无感知的。仅仅只有新的客户端会连接它,存储节点消费它,小事情。但是ingest节点会通过gossip协议传送各自的瞬时负荷给通信的对方。例如:连接的客户端数量,吞吐量等。这里有三种处理情况:



  • 有些高负载的服务器可以在连接时向新客户端提供一些轻量级的服务。客户端,例如:forward节点也会根据ingest节点的负载情况合理的传送日志记录。

  • 高负荷的ingest节点会在一段时间内拒绝新连接请求。

  • 如果需要的话,再高负载的ingest节点,也可以既拒绝新连接请求,也可以启动现有连接。


以上三种策略都是可以的,考虑到forward节点会重连集群中的其他ingest节点。有一点需要关心的是:响应慢,绝不要超过X%的ingest节点拒绝连接。这个ingest节点的负载均衡在issue 2有讨论。


相似地,增加一个新的存储节点也工作得很好。它会开始consume来之ingest节点的共享成比例段文件。在保留时间的窗口内,这将是平等的。但在该窗口过去之前,新的存储节点和其他存储节点相比,前者获取的共享比例段比较小。作为一种优化,它会通过gossip协议告诉其他存储节点它的当前数据集。当复制时,存储节点偏向于具有较小总体数据集大小的节点。新的存储节点将会获得较多的数据复制请求;老的存储节点将会接收较少的数据复制请求。这种逐渐重新平衡策略反应了我的观点,即尽可能地不要移动数据,静态存放。在发生大的拓扑变化之后,我观察到由分段/分片/节点重新平衡引起的大量中断。我认为这是可行的,但是它是否真的可行还有待观察。存储层的负载均衡在issue 3有讨论。


失败模式是经过深思熟虑的,但是没有经验证实它。我有兴趣构建一个带有故障注入的分布式验证框架,类似于简化的Jepsen风格测试工具。另外,建立一种方法来验证总体系统的吞吐量(MBps)和延时(ingest to query)。这个测试工作在issue 14中讨论。


在早期的设计过程中,我观察到队列理论在这里比较适用。我真的很喜欢用Adrian Cockcroft的微服务响应时间分布式的分析方法对系统进行建模。我开始着手这个工作,但是我没有太多时间去跟这个事情。这个模型在issue 9中讨论。


这真的只是一个开头;还有很多其他工作量比较小的事情要做。问题清单可能还需要一两个月才能列出来。


总结


这个系统对一些人是有用的吗?我不知道,或许吧。如果有人多听说,请试一试,或者可以邀请我和你一起在线讨论。如果没有,也好,这个系统设计对我也是一种很好的锻炼,是一个很享受的过程体验,并从中学到了很多知识。对我来说这就足够了。

Teleport v2.5发布,支持限制包大小与自定义包协议

henrylee2cn 发表了文章 • 0 个评论 • 232 次浏览 • 5 天前 • 来自相关话题

Teleport v2.5(简称tp v2.5)今日发布啦!它是一个通用、高效、灵活的TCP Socket框架。可用于Peer-Peer对等通信、... 查看全部

Teleport v2.5(简称tp v2.5)今日发布啦!它是一个通用、高效、灵活的TCP Socket框架。可用于Peer-Peer对等通信、RPC、长连接网关、微服务、推送服务,游戏服务等领域。这次升级新增了自定义通信协议、包大小限制等一些新特性,并作了一系列深度优化。


tp v2.5 特性变化:



  • 【新增】支持设置读取包的大小限制(如果超出则断开连接)

  • 【新增】支持定制通信协议

  • 【升级】支持插件机制,可以自定义认证、心跳、微服务注册中心、统计信息插件等

  • 【优化】无论服务器或客户端,均支持优雅重启、优雅关闭

  • 支持实现反向代理功能

  • 【优化】日志信息详尽,支持打印输入、输出消息的详细信息(状态码、消息头、消息体)

  • 服务器和客户端之间对等通信,两者API方法基本一致

  • 底层通信数据包包含HeaderBody两部分

  • 支持单独定制HeaderBody编码类型,例如JSON Protobuf string

  • Body支持gzip压缩

  • Header包含状态码及其描述文本

  • 支持推、拉、回复等通信模式

  • 支持设置慢操作报警阈值

  • 底层连接使用I/O缓冲区

  • 端点间通信使用I/O多路复用技术


teleport




tp v2.5 升级详情:


一、增加对自定义通信协议的支持,通过实现socket.Protocol接口来定制:


// Protocol socket communication protocol
type Protocol interface {
// WritePacket writes header and body to the connection.
WritePacket(
packet *Packet,
destWriter *utils.BufioWriter,
tmpCodecWriterGetter func(string) (*TmpCodecWriter, error),
isActiveClosed func() bool,
) error

// ReadPacket reads header and body from the connection.
ReadPacket(
packet *Packet,
bodyAdapter func() interface{},
srcReader *utils.BufioReader,
codecReaderGetter func(byte) (*CodecReader, error),
isActiveClosed func() bool,
checkReadLimit func(int64) error,
) error
}

然后,可以通过以下任意方法指定自己的通信协议:


func SetDefaultProtocol(socket.Protocol)
func (*Peer) ServeConn(conn net.Conn, protocol ...socket.Protocol) Session
func (*Peer) DialContext(ctx context.Context, addr string, protocol ...socket.Protocol) (Session, error)
func (*Peer) Dial(addr string, protocol ...socket.Protocol) (Session, error)
func (*Peer) Listen(protocol ...socket.Protocol) error

二、新增限制通信包大小


在读取包时可以限制包的大小,如果超出最大值则会主动断开连接。全局设置函数:


func SetReadLimit(maxPacketSize int64)

三、升级插件接口



  1. 插件返回值由以前的error改为tp.Xerror,从而用户可以灵活地在插件中定义错误码和错误描述;

  2. 增加更多、更细、更合理的插件位置

  3. 插件执行出错时的日志格式更加清晰整洁


// Interfaces about plugin.
type (
Plugin interface {
Name() string
}
PostRegPlugin interface {
Plugin
PostReg(*Handler) Xerror
}
PostDialPlugin interface {
Plugin
PostDial(PreSession) Xerror
}
PostAcceptPlugin interface {
Plugin
PostAccept(PreSession) Xerror
}
PreWritePullPlugin interface {
Plugin
PreWritePull(WriteCtx) Xerror
}
PostWritePullPlugin interface {
Plugin
PostWritePull(WriteCtx) Xerror
}
PreWriteReplyPlugin interface {
Plugin
PreWriteReply(WriteCtx) Xerror
}
PostWriteReplyPlugin interface {
Plugin
PostWriteReply(WriteCtx) Xerror
}
PreWritePushPlugin interface {
Plugin
PreWritePush(WriteCtx) Xerror
}
PostWritePushPlugin interface {
Plugin
PostWritePush(WriteCtx) Xerror
}
PreReadHeaderPlugin interface {
Plugin
PreReadHeader(ReadCtx) Xerror
}

PostReadPullHeaderPlugin interface {
Plugin
PostReadPullHeader(ReadCtx) Xerror
}
PreReadPullBodyPlugin interface {
Plugin
PreReadPullBody(ReadCtx) Xerror
}
PostReadPullBodyPlugin interface {
Plugin
PostReadPullBody(ReadCtx) Xerror
}

PostReadPushHeaderPlugin interface {
Plugin
PostReadPushHeader(ReadCtx) Xerror
}
PreReadPushBodyPlugin interface {
Plugin
PreReadPushBody(ReadCtx) Xerror
}
PostReadPushBodyPlugin interface {
Plugin
PostReadPushBody(ReadCtx) Xerror
}

PostReadReplyHeaderPlugin interface {
Plugin
PostReadReplyHeader(ReadCtx) Xerror
}
PreReadReplyBodyPlugin interface {
Plugin
PreReadReplyBody(ReadCtx) Xerror
}
PostReadReplyBodyPlugin interface {
Plugin
PostReadReplyBody(ReadCtx) Xerror
}

PostDisconnectPlugin interface {
Plugin
PostDisconnect(PostSession) Xerror
}
)

四、更多细节优化



  1. 运行日志中打印增加包序号seq,便于debug

  2. 当收到不支持的包类型时,断开连接并打印包详情

  3. tp.PullCmd增加func (c *PullCmd) Result() (interface{}, Xerror)方法,便于使用Session.GoPull方法进行并发请求

  4. 升级平滑重启与关闭功能

  5. 增加对并发资源的控制,防止内存资源耗尽

  6. 一些代码块的细节优化


Teleport项目地址:
https://github.com/henrylee2cn/teleport

结合 Go 读 APUE-文件共享

zhaohu 发表了文章 • 0 个评论 • 163 次浏览 • 2017-11-10 21:26 • 来自相关话题

在公众号 "别捉急" 上 同步了文章,并且可以点击原文链接阅读:查看全部


在公众号 "别捉急" 上 同步了文章,并且可以点击原文链接阅读:传送门



文件共享


UNIX 系统支持在不同进程间共享打开文件, 知识点:内核用于所有 I/O 的数据结构、原子操作。


概念性的 I/O 数据结构


内核用于所有 I/O 的数据结构,只是个概念性的,不一定适用,有个大体的轮廓就 OK。



  • 进程表 (process table entry) 中的记录

  • 文件表项 (file table entry)

  • v节点表项 (v-node table entry)


打开文件的内核数据结构


这是一个 打开文件的内核数据结构 图。打开文件 这个操作是一个进程, 每个进程在进程表中都有一个记录,而 打开文件进程记录 中包含一张打开文件描述符表, 包括:



  • 文件描述符标志

  • 指向一个文件表项的指针


文件描述符表用 Go 抽象如下表示:


type fd struct {
flags int
pointer *FileTableEntry
}

代码中 flags 的类型是随便定义的(实际我没查),由图中看出 pointer 指向文件表项 (file table entry), 内核为所有打开文件维持一张文件表, 每个文件表项包括:



  • 文件状态标志 (读、写、添写、同步和非阻塞)

  • 当前文件偏移量

  • 指向该文件 v 节点表项的指针


文件表项用 Go 抽象如下表示:


type FileTableEntry struct {
status int
offset int
pointer *VNodeTableEntry
}

由图中看出 pointer 指向v节点表项 (v-node table entry), 每个打开文件都有一个 v 节点结构如下所示:



  • 文件类型和对此文件进行各种操作函数的指针,统称为 v节点信息

  • 该文件的 i 节点: 文件所有者、文件长度、指向文件实际数据块在磁盘上所在位置的指针等


V 节点表项和 i 节点用 Go 抽象如下:


type VNodeTableEntry struct {
information *Information
vData *INode
}

type INode struct {
owner *Owner
length int
vNodeTableEntry *VNodeTableEntry
}

通过这种方式,来加深对 内核通用 I/O 数据结构 的理解。


如果两个独立进程各自打开同一个文件,则三者关系如下所示:


两个独立进程打开同一个文件


原子操作


一般而言,原子操作 (atomic operation) 指的是由多步组成的一个操作。如果该操作原子地执行,则要么执行完所有步骤,要么一步也不执行,不可能只执行所有步骤的一个子集。


函数 dup 和 dup2


下面两个函数都可用来复制一个现有的文件描述符。


#include <unistd.h>

int dup(int fd);

int dup2(int fd, int fd2);

上面函数中的参数:



  • fd 表示要复制的文件描述符

  • fd2 表示复制后的文件描述符


dup2 函数是可以指定复制后的文件描述符,而 dup 是返回当前可用文件描述符中的最小数值。


调用 dup(1) 函数后,进程表项,文件表,v 节点表,之间的关系图如下:
dup(1) 后的内核数据结构


对于传入的参数 fd2, 已经被打开了,会先关闭。知道了这个点,就明白了,下面的操作中会先调用 close(fd2)


很不爽了,没找到相应的 Go 的源码。复制一个描述符的另一种方法是使用 fcntl函数, dup2(fd, fd2) 等效于 close(fd2)fcntl(fd, F_DUPFD, fd2), 但不完全等效,因为 dup2(fd, fd2) 是个原子操作。


目前我先知道 fcntl函数 可以改变已经打开文件的属性,就可以啦。

不再傻傻分不清:atoi, itoa, iota

Xanthus 发表了文章 • 0 个评论 • 175 次浏览 • 2017-11-09 21:01 • 来自相关话题

atoi

Array to Integer 字符数组(字符串)转化为整数。golang标准库与C++标准库均有

itoa

Integer to Array 整数转化为字符串。golang标准库与C++标准... 查看全部

atoi


Array to Integer
字符数组(字符串)转化为整数。golang标准库与C++标准库均有


itoa


Integer to Array
整数转化为字符串。golang标准库与C++标准库均有


iota


希腊字母。golang中定义常量经常用的iota关键字,C艹中用于Store increasing sequence

emmmmm, 都是递增

记录一个拷贝文件到GlusterFS卡住的解决过程

houyy668 回复了问题 • 3 人关注 • 2 个回复 • 328 次浏览 • 2017-11-03 16:32 • 来自相关话题

Gorush 輕量級手機訊息發送服務

appleboy 发表了文章 • 6 个评论 • 256 次浏览 • 2017-11-01 10:21 • 来自相关话题

文章轉錄自: Gorush 輕量級手機訊息發送服務


68747470733a2f2f7261772e6769746875622e636f6d2f676f6c616e672d73616d706c65732f676f706865722d766563746f722f6d61737465722f676f706865722e706e67


今年第一次參加濁水溪以南最大研討會 Mopcon,給了一場議程叫『用 Go 語言打造輕量級 Push Notification 服務』,身為南部人一定要參加 Mopcon,剛好透過此議程順便發佈新版 Gorush,其實今年投稿 Mopcon 最主要是回家鄉宣傳 Google 所推出的 Go 語言,藉由實際案例來跟大家分享如何入門 Go 語言,以及用 Go 語言最大好的好處有哪些。底下是此議程大綱:



  • 為什麼建立 Gorush 專案

  • 如何用 Go 語言實作

  • Drone 自動化測試及部署

  • Kubernetes 上跑 Gorush


什麼是 Gorush


Gorush 是一套輕量級的 Push Notification 服務,此服務只做一件事情,就是發送訊息給 Google Andriod 或 Apple iOS 手機,啟動時預設跑在 8088 port,可以透過 Docker 或直接用 Command line 執行,對於 App 開發者,可以直接下載執行檔,在自己電腦端發送訊息給手機測試。藉由這次投稿順便發佈了新版本。底下來說明新版本多了哪些功能。


支援 RPC 協定


在此之前 Gorush 只有支援 REST API 方式,也就是透過 JSON 方式來通知伺服器來發送訊息,新版了多了 RPC 方式,這邊我是使用 gRPC 來實作,Gorush 預設是不啟動 gRPC 服務,你必須要透過參數方式才可以啟動,詳細可以參考此文件,底下是 Go 語言客戶端範例:


package main

import (
"log"

pb "github.com/appleboy/gorush/rpc/proto"
"golang.org/x/net/context"
"google.golang.org/grpc"
)

const (
address = "localhost:9000"
)

func main() {
// Set up a connection to the server.
conn, err := grpc.Dial(address, grpc.WithInsecure())
if err != nil {
log.Fatalf("did not connect: %v", err)
}
defer conn.Close()
c := pb.NewGorushClient(conn)

r, err := c.Send(context.Background(), &pb.NotificationRequest{
Platform: 2,
Tokens: []string{"1234567890"},
Message: "test message",
})
if err != nil {
log.Fatalf("could not greet: %v", err)
}
log.Printf("Success: %t\n", r.Success)
log.Printf("Count: %d\n", r.Counts)
}

支援 ARM64 Docker 映像檔


目前已經支援 ARM64 的 Docker 版本,所以可以在 ARM64 板子內用 Docker 來執行,可以直接在 Docker Hub 找到相對應的標籤


支援全域變數


Gorush 本身支援 Yaml 設定檔,但是每次想要改設定,都要重新修改檔案,這不是很方便,所以我透過 Viper 套件來讓 Gorush 同時支援 Yaml 設定,或 Global 變數,也就是以後都可以透過變數方式來動態調整,有了這方式就可以讓 Docker 透過環境變數來設定。底下是範例讓開發者動態調整 HTTP 服務 Port。請注意所有變數的前置符號為 GORUSH_


$ GORUSH_CORE_PORT=8089 gorush

支援 Kubernetes


此版增加了 Kubernetes 設定方式,有了上述的全域變數支援,這時候設定 Kubernetes 就更方便了,請直接參考 k8s 目錄,詳細安裝步驟請參考此說明,底下是透過 ENV 動態設定 Gorush


env:
- name: GORUSH_STAT_ENGINE
valueFrom:
configMapKeyRef:
name: gorush-config
key: stat.engine
- name: GORUSH_STAT_REDIS_ADDR
valueFrom:
configMapKeyRef:
name: gorush-config
key: stat.redis.host

iOS 支援動態發送到開發或正式環境


在此之前發送訊息到 iOS 手機,都必須在啟動伺服器前將 iOS 環境設定好,現在可以動態調整 JSON 參數。


{
"notifications": [
{
"tokens": ["token_a", "token_b"],
"platform": 1,
"message": "Hello World iOS!"
}
]
}

可以加上 developmentproduction 布林參數,底下是將訊息傳給 iOS 開發伺服器


{
"notifications": [
{
"tokens": ["token_a", "token_b"],
"platform": 1,
"development": true,
"message": "Hello World iOS!"
}
]
}

投影片


底下是議程投影片,有興趣的參考看看


https://www.slideshare.net/appleboy/gorush-a-push-notification-server-written-in-go


最後有講到如何部署及測試 Go 語言,這邊講了一下 Drone 這套自動化測試工具。如果大家有興趣可以參考我在 Udemy 開設的課程,目前特價 1600 元。Drone 幫忙開發者自動化測試,部署到 Docker Hub 或編譯出執行檔,這些在 Drone 裡面都可以透過 YAML 來設定,開發者只需要專注於寫程式就可以了。

不得不知道的golang知识点之nil

qiangmzsx 发表了文章 • 2 个评论 • 874 次浏览 • 2017-10-30 17:43 • 来自相关话题

golang中的nil,很多人都误以为与Java、PHP等编程语言中的null一样。但是实际上Golang的niu复杂得多了,如果不信,那我们继续往下阅读。

nil 为预声明的标示符,... 查看全部

golang中的nil,很多人都误以为与Java、PHP等编程语言中的null一样。但是实际上Golang的niu复杂得多了,如果不信,那我们继续往下阅读。


nil 为预声明的标示符,定义在builtin/builtin.go


// nil is a predeclared identifier representing the zero value for a
// pointer, channel, func, interface, map, or slice type.
// Type must be a pointer, channel, func, interface, map, or slice type
var nil Type

// Type is here for the purposes of documentation only. It is a stand-in
// for any Go type, but represents the same type for any given function
// invocation.
type Type int

nil的零值


按照Go语言规范,任何类型在未初始化时都对应一个零值:布尔类型是false,整型是0,字符串是"",而指针、函数、interface、slice、channel和map的零值都是nil。


PS:这里没有说结构体struct的零值为nil,因为struct的零值与其属性有关


nil没有默认的类型,尽管它是多个类型的零值,必须显式或隐式指定每个nil用法的明确类型。


package main

func main() {

// 明确.
_ = (*struct{})(nil)
_ = []int(nil)
_ = map[int]bool(nil)
_ = chan string(nil)
_ = (func())(nil)
_ = interface{}(nil)

// 隐式.
var _ *struct{} = nil
var _ []int = nil
var _ map[int]bool = nil
var _ chan string = nil
var _ func() = nil
var _ interface{} = nil
}

如果关注过golang关键字的同学就会发现,里面并没有nil,也就是说nil并不是关键字,那么就可以在代码中定义nil,那么nil就会被隐藏。


package main

import "fmt"

func main() {
nil := 123
fmt.Println(nil) // 123
var _ map[string]int = nil //cannot use nil (type int) as type map[string]int in assignment
}

nil类型的地址和值大小


nil类型的所有值的内存布局始终相同,换一句话说就是:不同类型nil的内存地址是一样的。


package main
import (
"fmt"
)
func main() {
var m map[int]string
var ptr *int
var sl []int
fmt.Printf("%p\n", m) //0x0
fmt.Printf("%p\n", ptr ) //0x0
fmt.Printf("%p\n", sl ) //0x0
}

业务中一般将nil值表示为异常。nil值的大小始终与其类型与nil值相同的non-nil值大小相同。因此, 表示不同零值的nil标识符可能具有不同的大小。


package main

import (
"fmt"
"unsafe"
)

func main() {
var p *struct{} = nil
fmt.Println( unsafe.Sizeof( p ) ) // 8

var s []int = nil
fmt.Println( unsafe.Sizeof( s ) ) // 24

var m map[int]bool = nil
fmt.Println( unsafe.Sizeof( m ) ) // 8

var c chan string = nil
fmt.Println( unsafe.Sizeof( c ) ) // 8

var f func() = nil
fmt.Println( unsafe.Sizeof( f ) ) // 8

var i interface{} = nil
fmt.Println( unsafe.Sizeof( i ) ) // 16
}

大小是编译器和体系结构所依赖的。以上打印结果为64位体系结构和正式 Go 编译器。对于32位体系结构, 打印的大小将是一半。


对于正式 Go 编译器, 同一种类的不同类型的两个nil值的大小始终相同。例如, 两个不同的切片类型 ( []int和[]string) 的两个nil值始终相同。


nil值比较


1.不同类型的nil是不能比较的。


package main
import (
"fmt"
)
func main() {
var m map[int]string
var ptr *int
fmt.Printf(m == ptr) //invalid operation: m == ptr (mismatched types map[int]string and *int)
}

在 Go 中, 两个不同可比较类型的两个值只能在一个值可以隐式转换为另一种类型的情况下进行比较。具体来说, 有两个案例两个不同的值可以比较:



  • 两个值之一的类型是另一个的基础类型。

  • 两个值之一的类型实现了另一个值的类型 (必须是接口类型)。


nil值比较并没有脱离上述规则。


package main
import (
"fmt"
)
func main() {
type IntPtr *int
fmt.Println(IntPtr(nil) == (*int)(nil)) //true
fmt.Println((interface{})(nil) == (*int)(nil)) //false
}

2.同一类型的两个nil值可能无法比较
因为golang中存在map、slice和函数类型是不可比较类型,它们有一个别称为不可比拟的类型,所以比较它们的nil亦是非法的。


package main
import (
"fmt"
)
func main() {
var v1 []int = nil
var v2 []int = nil
fmt.Println(v1 == v2)
fmt.Println((map[string]int)(nil) == (map[string]int)(nil))
fmt.Println((func())(nil) == (func())(nil))
}

不可比拟的类型的值缺是可以与“纯nil”进行比较。


package main
import (
"fmt"
)
func main() {
fmt.Println((map[string]int)(nil) == nil) //true
fmt.Println((func())(nil) == nil) //true
}

3.两nil值可能不相等


如果两个比较的nil值之一是一个接口值, 而另一个不是, 假设它们是可比较的, 则比较结果总是 false。原因是在进行比较之前, 接口值将转换为接口值的类型。转换后的接口值具有具体的动态类型, 但其他接口值没有。这就是为什么比较结果总是错误的。


package main
import (
"fmt"
)
func main() {
fmt.Println( (interface{})(nil) == (*int)(nil) ) // false
}

常见问题


1.函数返回


func nilReturn() (string,error)  {

return nil,nil //cannot use nil as type string in return argument
}

因为error是接口类型所以error类型没有报错。


2.map的nil key
map的key为指针、函数、interface、slice、channel和map,则key可以为nil。


package main
import (
"fmt"
)
func main() {
mmap := make(map[*string]int,4)
a:="a"
mmap[&a] = 1
mmap[nil] = 99
fmt.Println(mmap) //map[0xc042008220:1 <nil>:99]
}

总结


nil之所以比较难以理解因为我们经常混淆了nil值和nil类型,希望各位同学细细品味其中区别。

软件开发过程中值不值得写单元测试?

voidint 发表了文章 • 2 个评论 • 282 次浏览 • 2017-10-27 23:14 • 来自相关话题


原文链接: https://voidint.github.io/post/2017/10/24/ut/



一、不写单元测试的理由


工作几年,遇到过不少抗拒写单元测试的人,总结一下大致有以下这么几个理由:


首先,写单元测试所支出的时间可能比实现功能本身所花费的时间还多。言外之意,在实现完所有功能之前不值得写单元测试。如果现阶段功能开发大致完毕,可能也不会去补上亏欠的单元测试,理由大致有这么几点:



  • 需求总是无穷尽的,还有下阶段功能需求要实现,没空补单元测试。

  • 要补的单元测试太多,无从下手,主观上抗拒。

  • 单元测试编写难度大。一方面原因可能是功能函数实现上不够合理,另一方面是没有(或者不知道)好用的单元测试框架和mock框架。

  • 单元测试不算入工作量内。


其次,功能需求还不稳定,写单元测试的性价比不高。换句话说,万一明天需求一变,那不光功能代码废了,单元测试也废了。如果不写单元测试,那这部分工夫就不会白费。


二、写单元测试的投入产出比


投入产出比作为这个问题的判断依据,我觉得是所有理性人都会做出的选择。而既然引入了投入产出比这个经济学上的概念,那么应该也能接受另一个经济学上的普世规律——边际收益递减。以及资源(主要指时间和精力)具有稀缺性这一规律。


下面的分析都将是建立在以上这些共识的基础之上,如果你并不认同这些规律同样可以用来分析软件开发过程中值不值得写单元测试这个问题,那么阅读以下内容仅仅是在浪费你宝贵的时间。如果你继续往下阅读,那么,我将假设你已经和我达成了这些共识。


虽然要为这个问题做精确的投入产出比定量分析比较困难,但并不妨碍我们通过定性分析来得出自己心中的答案。


成本(投入)



  • 编写单元测试用例所额外付出的时间,短期内会拖慢项目进度。


收益(产出)




  • 提升代码质量。监督开发人员写出更加易于测试和可维护的代码。




  • 提升开发团队内部的协作效率。其他开发人员可以通过阅读单元测试用例来理解代码原作者的意图。




  • 保证功能实现的长期稳定。代码一旦发生与原功能意图不相符的变化,通过跑单元测试可以体现出来,即可以防止功能被无意识地破坏。



  • 提高自动化测试占比,降低其他测试方式上的投入。


从上面我所罗列的成本/收益数量上说,收益的数量远大于成本。但是,分析投入产出比并不是掰手指数数量就能得出答案的,特此说明。


首先,短期项目不写单元测试是划算的选择。至于到底多长时间算作短期,并无定论。我选择将一个月作为划分的界限,一月以内为短期,多于一月是中期或者长期,至于中长期的界限在哪儿,这个无关紧要,暂且不论。


短期项目的典型代表就是演示用demo项目。这类项目的共性是时间紧迫,项目生命周期短,无需后续的维护,使用一次性(或者接近一次性)。如果说中长期项目的使用周期(区别于开发周期)是时间段的话,那么短期项目的使用周期就是一个时间点或者几个时间点。这种情况下将有限的资源投入到单元测试中,所获得的收益并不明显。如果继续追加投入,由于收益增幅平缓,投入产出比极低,甚至为负。索性零投入零产出,虽然略显保守,但也不失为一个好的选择。


其次,中长期项目不写单元测试绝对不是划算的选择。请注意,这里我并没有说中长期项目写单元测试是划算的选择这样的话。因为写单元测试这个说法太宽泛。很明显,单元测试覆盖率1%99%是有巨大区别的,而这两者都属于写单元测试这个范畴。


对于中长期项目,将资源投入到单元测试上所获得的收益曲线会是这样(不太会画图,暂且用文字描述):




  • 横坐标为投入,纵坐标为产出。




  • 开始阶段,随着投入的增加,收益(产出)增幅明显,曲线比较陡峭。




  • 当投入到达某一临界值,单位投入所获得的收益最大。



  • 当投入继续追加并超过临界值,收益的增幅明显放缓,曲线开始变得平缓,单位投入所获得的收益越来越少,即边际收益递减。


如果你也认可以上的投入产出比曲线,那么不难推出以下几个结论:




  • 不写单元测试一定不是最佳选择。不写单元测试可以理解为是零投入/零成本,那么必然的也会是零收益。




  • 写单元测试并且单元测试的覆盖率接近100%也一定不是最佳选择。由于边际收益递减,投入一旦越过临界值,那么继续追加投入所带来的收益将越来越少。



  • 临界值才是理论上的最佳选择。


上面提到的临界值是属于写单元测试的范畴,并且单元测试覆盖率在0%~100%之间的某个点。所以说,软件开发过程中值不值得写单元测试这个问题的答案应该无需再多言了。


文章一开始提到的那些不写单元测试的理由,往往是抱着一个非黑即白的观点,认为不写单元测试的反面就是写单元测试且覆盖率近100%。因而会过分夸大写单元测试的成本(单元覆盖率100%的代价当然大),选择短期利益。


三、个人的做法


下面以个人的小开源项目gbb为例,啰嗦几句我自己写单元测试的习惯。




  • 开源项目一定要写单元测试!开源项目一定要写单元测试!开源项目一定要写单元测试!按照规定,重要的事情得说三遍。除了上文提到的单元测试的好处外,单元测试也是开源项目负责任的一种体现。我花了时间去构思各种情况下的测试用例,我能为开源项目质量负责。




  • 不是非得写完一个函数就必须立马写单元测试或者TDD。关于写单元测试的时机,我一般选择在完成一个基本功能后写单元测试,优先覆盖功能的主体逻辑,一些边界条件逻辑不会体现在这个阶段的单元测试当中。我觉得这个阶段也是投入产出比最大的阶段。




  • 等功能相对稳定后,再把一些边界条件逻辑纳入到单元测试当中。这是单元测试覆盖率提升较大的阶段。



  • 最后一个阶段是刷数据阶段(gbb项目单元测试覆盖率曲线)。所谓的刷数据阶段的主要目标就是为了让覆盖率尽可能提高。刷数据并不是在某个功能成熟稳定后开始,我是选择在开源项目进入成熟阶段后开始刷单元测试覆盖率,为的就是使其更加好看。对于非开源项目,建议选择跳过这个抠细节的阶段。


四、参考


在写这篇博客时,我并没有查阅相关的这类文章。为的就是防止他人的观点在不知不觉中掺入其中,影响了我的原始观点。

从Go标准库strings看字符串匹配算法

redstorm 发表了文章 • 1 个评论 • 443 次浏览 • 2017-10-18 20:23 • 来自相关话题

Go的标准库本身质量非常高,本文主要深入strings库,从源代码中探查字符串匹配常用算法的具体实现

我们先看一个简单的例子开始。

在目标字符串中检查是否有子串等于匹配文本,这是非常常见的操作. 最容易让人想到的算法就是从目标... 查看全部

Go的标准库本身质量非常高,本文主要深入strings库,从源代码中探查字符串匹配常用算法的具体实现


我们先看一个简单的例子开始。


在目标字符串中检查是否有子串等于匹配文本,这是非常常见的操作. 最容易让人想到的算法就是从目标字符串第一位开始,逐个字符与待匹配文本比较.匹配成功则指针右移继续比较,要不然从目标文本的第二位开始和待匹配文本继续逐个比较。如果指针先到达待匹配文本末尾,则匹配成功,要不然匹配失败。该算法称之为朴素算法,非常容易理解,但效率也比较慢.具体实现如下:


#include<stdio.h>
#include<string.h>
void search(char *pat, char *txt)
{
int M = strlen(pat);
int N = strlen(txt);
int i;
// 只需要遍历目标文本 N-M 次, 因为从目标文本的 N-M 位开始的子串,长度永远小于 M, 所以不会匹配成功
for (i = 0; i <= N - M; i++)
{
int j;
for (j = 0; j < M; j++)
{
if (txt[i+j] != pat[j])
break;
}
if (j == M)
{
printf("Pattern found at index %d \n", i);
}
}
}

int main()
{
char *txt = "AABAACAADAABAAABAA";
char *pat = "AABA";
search(pat, txt);
return 0;
}

Go 标准库中的strings.Contains函数使用了Rabin-Karp算法, 主要思想如下:


假设匹配文本的长度为M,目标文本的长度为N



  1. 计算匹配文本的hash值

  2. 计算目标字符串中每个长度为M的子串的hash值(需要计算N-M+1次)

  3. 比较hash值, 如果hash值不同,字符串必然不匹配,如果hash值相同,还需要使用朴素算法再次判断


步骤2中每次都要重新计算hash, Rabin-Karp算法的优点在于设计了一个特别的hash算法,使其在计算下一个子串的hash时可以利用之前的hash结果, 以达到加速计算的效果。将每一个字节看作数字, 选择一个比较大的质数作为base. 字节的值是包含在基数之内的


举例说明:

文本为"abracadabra",base为101,那么 hash("abr") = 97 101的2次方 + 98 101的1次方 + 114 101的0次方= 999509

下一个子串 "bra"的hash值为 98
101的2次方 + 114 101的1次方 + 97 101的0次方. 我们可以利用之前"abr"的hash值, 写成:



//       base  old hash      new 'a'    old 'a' base

hash("bra") = 1011
hash("abr") + (97 × 101的0次方) - (97 × 101的3次方)



可以看出hash算法里要点是确立一个非常大的数字作为base,同时根据子串长度得到乘数因子(上述的 101的3次方,其实就是base的len(待匹配文本)次方).


src/strings/strings_amd64.go相关代码注释


// 选择非常大的一个质数16777619 作为 base 
const primeRK = 16777619

// hashStr 返回子串的hash值和乘数因子
func hashStr(sep string) (uint32, uint32) {
hash := uint32(0)
for i := 0; i < len(sep); i++ {
hash = hash*primeRK + uint32(sep[i]) //计算hash值
}
// 计算(最高位 + 1)位的乘数因子, 使用位移, 没有使用 i--, 可以有效减少循环次数. i >>=1 相当于遍历二进制的每一位
var pow, sq uint32 = 1, primeRK
for i := len(sep); i > 0; i >>= 1 {
if i&1 != 0 {
pow *= sq
}
sq *= sq
}
return hash, pow
}

// Index 返回sep在s里第一次匹配时的index, 无法匹配则返回-1.
func Index(s, sep string) int {
n := len(sep)
// 先分析一些常见情况, 起到进一步加速的效果
switch {
case n == 0:
return 0
case n == 1: //如果为一个字节,则调用IndixByte(汇编语言)
return IndexByte(s, sep[0])
case n <= shortStringLen: //如果sep的长度小于31且大于1, 则使用汇编代码(也是一种优化).
return indexShortStr(s, sep)
case n == len(s):
if sep == s {
return 0
}
return -1
case n > len(s):
return -1
}
// 使用Rabin-Karp算法匹配
// 步骤1 初始计算待匹配的文本的hash值和乘数因子,
hashsep, pow := hashStr(sep)
var h uint32
for i := 0; i < n; i++ {
h = h*primeRK + uint32(s[i]) // 步骤2 计算长度跟sep一样的s子串的hash值
}
if h == hashsep && s[:n] == sep {
return 0
}
for i := n; i < len(s); {
// 利用先前的hash值, 计算新的hash值
h *= primeRK // 乘以base
h += uint32(s[i]) // 加上下一个字符的 hash 值
h -= pow * uint32(s[i-n]) // 减去先前子串的第一个字符的hash值
i++
// 如果hash相等则继续使用朴素算法比较, 如果hash不一致,则直接用下一个匹配
if h == hashsep && s[i-n:i] == sep {
return i - n
}
}
return -1
}

strings库里还实现了BM算法, 在这之前,我们先来看另一个非常经典的KMP算法


假设检查bacbababaabcbab是否包含abababca, 此时发现第6位不一样


bacbababaabcbab   
abababca
|
第六位

朴素算法:
bacbababaabcbab
abababca
|
移动一位后开始重新比较

KMP算法:
bacbababaabcbab
abababca
|
直接移动两位后开始重新比较

如果按朴素算法则按上面所示需要搜索词移一位后重新从第一位开始匹配。仔细想想, 前5个字符ababa已经匹配成功,也就是我们已经知道双方的文本, 通过提前的计算,可以多移几位, 而不仅仅移一位. 这样可以加快搜索


KMP算法的主要原理如下:

s为目标文本, 长度为m

p为搜索词,长度为n

假设p[i]与s[x]匹配失败,那么p[i-1]与s[x-1]是匹配成功的, 则试图找到一个索引 j, 使得p[0:j] = p[i-j-1:i-1] (p[0:j] 包含p[j])

如果有则s[x]继续与p[j+1]进行比较, 相当于搜索词移动i-j-1位

无则s[x]与p[0]比较. (具体代码实现时无可以表示为-1, 这样+1 后正好为0) 相当于搜索词移动i位


void cal(char *p, int *next)
{
int i;
int k;
/*第一次字符前面没有索引了, 算corner case, 直接赋值为-1*/
next[0] = -1;
/* 循环每一个索引, 并计算next值 */
for (i = 1; p[i] != '\0'; i++) {
/* 获取前一个索引的next值 */
k = next[i - 1];
/* 当p[i] != p[k + 1]时, 则令 k = next[k], 直到相等或者k == -1 退出*/
while (p[i] != p[k + 1]) {
if (k == -1) {
k = -2;
break;
}
k = next[k];
}
/* 1. p[i] == p[k + 1] 则 i对应的next值为 ++k
2. 无索引时, k= -2, 则++k正好为-1
*/
next[i] = ++k;
}
}

int kmp(char *p, char*t)
{
/*next为数组, 存储搜索词里每一个索引对应的next值, 使得 p[0:next[i]] == p[i-j-1:i-1]*/
int next[strlen(p)];
cal(p, next);
int i, j;
i = 0;
j = 0;
while (p[i] != '\0' && t[j] != '\0') {
if (p[i] == t[j]) {
/* 值相等, 则指针 i, j 都递增 */
i++;
j++;
} else {
if (i == 0) {
j++;
continue;
}
i = next[i - 1] + 1;
}
}
if (p[i] == '\0') {
return 0;
} else {
return 1;
}
}

Go语言里在 strings/search.go 里使用了Boyer-Moore字符串搜索算法, 这个思想和KMP类似,都是根据Pattern自身计算出移动的步数. 有两个优化点:



  1. BM算法是从后向前逐渐匹配.

  2. kmp里的通过已匹配的文本增加移动步数的叫做好规则,那么BM里同时还增加了坏规则


假定Text为"HERE IS A SIMPLE EXAMPLE",Pattern为"EXAMPLE"。

当T[i] != P[j], P[j]右边都匹配时时, 具体的移动规则如下:

坏字符规则: 此时T[i]定义为坏字符, 如果P[0..j-1]中包含T[i]这个字符, 则移动T使坏字符与相等的字符对齐, 如果不包含,则直接移动len(P)


HERE IS A SIMPLE EXAMPLE
|
EXAMPLE

此时P为坏字符, 因EXAMPLE包含P, 则T的i指针右移二位使之对齐,然后重新开始从P的末端继续匹配(下面打X处).

HERE IS A SIMPLE EXAMPLE
| X
EXAMPLE

如下场景,T中的M与P中的E不匹配, 按Go的代码实现,是移动两位(取该字符到P末尾的最短距离),没完全按上面的规则实现
大家是不是发现没有跳跃前进,反而匹配又倒回到之前已完成的匹配过程。 Go代码这么做是为了实现简单。
因为还有好规则可以保证最终的移动步数是正确的
ABCADADEEFXYZ
|
AYEDADE
移动为
ABCADADEEFXYZ
| X
AYEDADE

好后缀规则: 当发生不匹配时,之前已经匹配成功的,称之为好字符. 如下I和A不匹配, 后面的MPLE就是好后缀. 首先检查P里是否好后缀只出现过一次: 比如此时的MPLE作为好后缀在整个字符串EXAMPLE中只出现过一次



  • 不是, 则移动P使T中的好后缀与P中长度相等的字符串对齐

  • 是, 则继续检查好后缀的所有后缀(比如PLE,PL,E)是否和同等长度的P前缀相等, 如果相等则移动P使之对齐, 不相等则移动 len(P).

    这里相当于要求后缀必须出现在P的首部, 如果非首部, 因前缀的前一个字符必然不相等,则整个字符串肯定无法匹配


HERE IS A SIMPLE EXAMPLE
||||
EXAMPLE
MPLE, PLE,LE没法和首部匹配,但后缀E和P前缀相等, 则移动T使其对齐,从打X出继续从后向前比较
HERE IS A SIMPLE EXAMPLE
| X
EXAMPLE

具体的代码注释如下:


func makeStringFinder(pattern string) *stringFinder {
f := &stringFinder{
pattern: pattern,
goodSuffixSkip: make([]int, len(pattern)),
}
// last 是pattern最后一个字符的索引
last := len(pattern) - 1

// 创建坏字符表,记录不匹配时T的i指针移动步数
// 第一阶段,初始化256个字符全部移动 len(pattern) 步
for i := range f.badCharSkip {
f.badCharSkip[i] = len(pattern)
}

// 第二阶段:从左到右遍历pattern,更新其索引与P末尾的距离,结果就是该字符到末尾的最小距离
// 没有计算last byte的距离, 因为移动至少要一步。 没有0步。
for i := 0; i < last; i++ {
f.badCharSkip[pattern[i]] = last - i
}

// 创建好后缀表
// 第一阶段: 此时pattern[i+1:]都是已经匹配的,且好后缀只出现了一次
// 计算T中的指针要移动的步数
lastPrefix := last
for i := last; i >= 0; i-- {
if HasPrefix(pattern, pattern[i+1:]) {
lastPrefix = i + 1
}
// 好后缀时T的指针移动分两步,首先移动到与 pattern的末尾对齐,即 last - i
// lastPrefix 用来记录 pattern[i+1:]中所有后缀与同等长度的前缀相等时的最大索引
// 然后移动 lastPrefix步
f.goodSuffixSkip[i] = lastPrefix + last - i
}
// 第二阶段: 好后缀在pattern前面部分还出现过, 如下计算相应的移动步数
// 会覆盖之前第一阶段的部分值。但好后缀出现过移动步数比没出现的小。所以最终值是正确的
// 举例: "mississi" 中好后缀是issi, 在pattern[1]处出现过,所以移动步数为 last-i + lenSuffix
for i := 0; i < last; i++ {
lenSuffix := longestCommonSuffix(pattern, pattern[1:i+1])
if pattern[i-lenSuffix] != pattern[last-lenSuffix] {
// (last-i) is the shift, and lenSuffix is len(suffix).
f.goodSuffixSkip[last-lenSuffix] = lenSuffix + last - i
}
}
return f
}
// longestCommonSuffix 仅仅比较两个字符串的共同后缀的长度, 没有则为0
func longestCommonSuffix(a, b string) (i int) {
for ; i < len(a) && i < len(b); i++ {
if a[len(a)-1-i] != b[len(b)-1-i] {
break
}
}
return
}

// next 主要返回p在text里第一次匹配时的索引, 不匹配则返回-1
func (f *stringFinder) next(text string) int {
// i 是T(即变量text)中要检查的字符索引, j为P中要检查的字符索引

// 因从后向前比较, 所以i初始化为P的最后一位索引
i := len(f.pattern) - 1
for i < len(text) {
// 每次比较时都从p的最后一位开始比较
j := len(f.pattern) - 1
for j >= 0 && text[i] == f.pattern[j] {
i--
j--
}
// j为负数,说明匹配成功, 则直接返回 i+ 1
if j < 0 {
return i + 1
}
// j为非负, 表明text[i] != f.pattern[j], 则从坏字符表和好后缀表中获取分别获取i需要移动的步数, 取最大值并使移动到新位置
i += max(f.badCharSkip[text[i]], f.goodSuffixSkip[j])
}
return -1
}

链表以及golang介入式链表的实现

sheepbao 发表了文章 • 4 个评论 • 411 次浏览 • 2017-10-14 21:47 • 来自相关话题

链表以及golang介入式链表的实现

今天看tcp/ip协议栈的代码时看到一个双向链表,链表吗?听过它的顶顶大名,知道它是由节点构成的,每个节点还有个指针指向下一个节点,但是从来没自己实现过一个,没有实践就不能深刻理解,遂有此文。查看全部

链表以及golang介入式链表的实现


今天看tcp/ip协议栈的代码时看到一个双向链表,链表吗?听过它的顶顶大名,知道它是由节点构成的,每个节点还有个指针指向下一个节点,但是从来没自己实现过一个,没有实践就不能深刻理解,遂有此文。

以下所有观点都是个人愚见,有不同建议或补充的的欢迎emial我aboutme


何为链表?


链表(Linked list)是一种常见的基础数据结构,是一种线性表,但是并不会按线性的顺序存储数据,而是在每一个节点里存到下一个节点的指针(Pointer)。由于不必须按顺序存储,链表在插入的时候可以达到O(1)的复杂度,比另一种线性表顺序表快得多,但是查找一个节点或者访问特定编号的节点则需要O(n)的时间,而顺序表相应的时间复杂度分别是O(logn)和O(1)。

简单的说链表是一个具有逻辑顺序的线性表,每一个节点里存到下一个节点的指针。


图示


单链表


list1


双向链表


list2


链表有啥用?


因为链表插入很快,而且具有动态性,想添加几个元素就添加几个(内存空间足够),不像数组一样那么死板,正因为链表的灵活性,所有链表的用处是大大的有啊。

链表最适合用于频繁更新变化的数据,比如一个需要异步执行并且不可丢失的命令序列、一个需要进行实时加载与卸载的驱动,无序且数量未知,这个时候就需要链表结构来协助完成数据的管理。如果不需要过度关注数据的顺序,还可以用链表方便快捷地在任意一个地方插入或删除一个元素,并且不会影响到其它的元素。

又或者我在今天看tcp/ip源码中,链表用来构造队列,作为数据段的队列。我想链表用于队列应该是最多的。如果你看过linux内核源码,应该会发现linux内核中多处使用链表这种结构。


go标准库的双向链表


golang的标准库中实现了一个双向链表,该链表可以存储任何数据,先看看使用标准库链表的例子:


package list_test

import (
"container/list"
"fmt"
"testing"
)

func TestList(t *testing.T) {
// Create a new list and put some numbers in it.
l := list.New()
e4 := l.PushBack(4)
e1 := l.PushFront(1)
l.InsertBefore(3, e4)
l.InsertAfter(2, e1)

// Iterate through list and print its contents.
for e := l.Front(); e != nil; e = e.Next() {
fmt.Println(e.Value)
}
}
// output
// 1
// 2
// 3
// 4

该链表实现了链表所有的功能,链表的增删查改。


实现该链表的数据结构


// List represents a doubly linked list.
// The zero value for List is an empty list ready to use.
type List struct {
root Element // sentinel list element, only &root, root.prev, and root.next are used
len int // current list length excluding (this) sentinel element
}

// Element is an element of a linked list.
type Element struct {
// Next and previous pointers in the doubly-linked list of elements.
// To simplify the implementation, internally a list l is implemented
// as a ring, such that &l.root is both the next element of the last
// list element (l.Back()) and the previous element of the first list
// element (l.Front()).
next, prev *Element

// The list to which this element belongs.
list *List

// The value stored with this element.
Value interface{}
}

可以看到Element结构体看到了链表的结构,next,prev分别指向下一个和前一个元素的指针。Value就是链表中的数据段,可以理解为上图中的object。


介入式链表(intrusive list)


前面的链表都是普通链表,记得<<c语言程序设计>>上讲的链表也是一样,就是链表的节点指针和数据段是放在同一个struct,每实现一个不同的struct就得重新实现一遍链表的功能,这对于“懒惰”的程序员来说是不可忍受的,所以就出来了介入式链表,将数据段和链表的功能区别开来。最经典的例子莫过于linux内核的list_head,详情请看链接klist or Linked List in Linux Kernel,linux中是用c实现的,我想用go实现一个介入式链表。


实现代码


package list

type Intrusive interface {
Next() Intrusive
Prev() Intrusive
AddNext(Intrusive)
AddPrev(Intrusive)
}

// List provides the implementation of intrusive linked lists
type List struct {
prev Intrusive
next Intrusive
}

func (l *List) Next() Intrusive {
return l.next
}

func (l *List) Prev() Intrusive {
return l.prev
}

func (l *List) AddNext(i Intrusive) {
l.next = i
}

func (l *List) AddPrev(i Intrusive) {
l.prev = i
}

func (l *List) Reset() {
l.prev = nil
l.next = nil
}

func (l *List) Empty() bool {
return l.prev == nil
}

// Front returns the first element of list l or nil.
func (l *List) Front() Intrusive {
return l.prev
}

// Back returns the last element of list l or nil.
func (l *List) Back() Intrusive {
return l.next
}

// PushFront inserts the element e at the front of list l.
func (l *List) PushFront(e Intrusive) {
e.AddPrev(nil)
e.AddNext(l.prev)

if l.prev != nil {
l.prev.AddPrev(e)
} else {
l.next = e
}
l.prev = e
}

// PushBack inserts the element e at the back of list l.
func (l *List) PushBack(e Intrusive) {
e.AddNext(nil)
e.AddPrev(l.next)

if l.next != nil {
l.next.AddNext(e)
} else {
l.prev = e
}
l.next = e
}

// InsertAfter inserts e after b.
func (l *List) InsertAfter(e, b Intrusive) {
a := b.Next()
e.AddNext(a)
e.AddPrev(b)
b.AddNext(e)

if a != nil {
a.AddPrev(e)
} else {
l.next = e
}
}

// InsertBefore inserts e before a.
func (l *List) InsertBefore(e, a Intrusive) {
b := a.Prev()
e.AddNext(a)
e.AddPrev(b)
a.AddPrev(e)

if b != nil {
b.AddNext(e)
} else {
l.prev = e
}
}

// Remove removes e from l.
func (l *List) Remove(e Intrusive) {
prev := e.Prev()
next := e.Next()

if prev != nil {
prev.AddNext(next)
} else {
l.prev = next
}

if next != nil {
next.AddPrev(prev)
} else {
l.next = prev
}
}

我们这里用List表示实现了Intrusive接口,也实现了链表的基本功能,所以任何实现了Intrusive接口的对象都是可以作为链表的节点,利用这个介入式链表就很简单了,只要在现有的struct嵌入List这个结构体即可,在举个例子:


package list

import (
"container/list"
"fmt"
"testing"
)

func TestIntrusiveList(t *testing.T) {
type E struct {
List
data int
}
// Create a new list and put some numbers in it.
l := List{}
e4 := &E{data: 4}
e1 := &E{data: 1}
l.PushBack(e4)
l.PushFront(e1)
l.InsertBefore(&E{data: 3}, e4)
l.InsertAfter(&E{data: 2}, e1)

for e := l.Front(); e != nil; e = e.Next() {
fmt.Printf("e: %#v\n", e)
fmt.Printf("data: %#v\n", e.(*E).data)
}
}

E里嵌入List即可作为链表的节点,是不是很方便,其实当我写完介入式链表的栗子后,发现其实标准库的链表更方便,哈哈。。因为golang有interface{}


参考


https://blog.goquxiao.com/posts/2013/07/06/intrusive-list/

http://blog.nlogn.cn/linked-list-in-linux-kernel/

【翻译】使用 sync.ErrGroup 实现并发搜索文件

crazyvv 回复了问题 • 6 人关注 • 2 个回复 • 1708 次浏览 • 2017-10-11 15:11 • 来自相关话题

Go并发机制

simple 发表了文章 • 0 个评论 • 557 次浏览 • 2017-10-08 22:31 • 来自相关话题

不知道为什么有三张图在这里无法显示,有点郁闷!!! ->查看全部

不知道为什么有三张图在这里无法显示,有点郁闷!!! ->原文链接


1. C/C++ 与 Go语言的“价值观”对照


之前看过 白明老师 在GopherChina2017的一篇演讲文章《Go coding in go way》,里面提到C/C++/Go三门语言价值观,感觉很有意思,分享给大家感受一下:


C的价值观摘录



  • 相信程序员:提供指针和指针运算,让C程序员天马行空的发挥

  • 自己动手,丰衣足食:提供一个很小的标准库,其余的让程序员自造

  • 保持语言的短小和简单

  • 性能优先


C++价值观摘录



  • 支持多范式,不强迫程序员使用某个特定的范式

  • 不求完美,但求实用(并且立即可用)


Go价值观



  • Overall Simplicity 全面的简单

  • Orthogonal Composition 正交组合

  • Preference in Concurrency 偏好并发


用一句话概括Go的价值观:
Go is about orthogonal composition of simple concepts with preference in concurrency(Go是在偏好并发的环境下的简单概念/事物的正交组合).


从Go的价值观介绍可以看出 Go很适合并发编程,可以说其是为并发而生的一门语言,那它的并发机制如何?这正是这篇文章想要介绍的。


2. 从线程实现模型说起


线程的实现模型主要有3种:内核级线程模型、用户级线程模型和混合型线程模型。它们之间最大的区别在于线程与内核调度实体KSE(Kernel Scheduling Entity)之间的对应关系上。所谓的内核调度实体KSE 就是指可以被操作系统内核调度器调度的对象实体,有些地方也称其为内核级线程,是操作系统内核的最小调度单元。


2.1 内核级线程模型


用户线程与KSE是1对1关系(1:1)。大部分编程语言的线程库(如linux的pthread,Java的java.lang.Thread,C++11的std::thread等等)都是对操作系统的线程(内核级线程)的一层封装,创建出来的每个线程与一个不同的KSE静态关联,因此其调度完全由OS调度器来做。这种方式实现简单,直接借助OS提供的线程能力,并且不同用户线程之间一般也不会相互影响。但其创建,销毁以及多个线程之间的上下文切换等操作都是直接由OS层面亲自来做,在需要使用大量线程的场景下对OS的性能影响会很大。


2.2 用户级线程模型


用户线程与KSE是多对1关系(M:1),这种线程的创建,销毁以及多个线程之间的协调等操作都是由用户自己实现的线程库来负责,对OS内核透明,一个进程中所有创建的线程都与同一个KSE在运行时动态关联。现在有许多语言实现的 协程 基本上都属于这种方式。这种实现方式相比内核级线程可以做的很轻量级,对系统资源的消耗会小很多,因此可以创建的数量与上下文切换所花费的代价也会小得多。但该模型有个致命的缺点,如果我们在某个用户线程上调用阻塞式系统调用(如用阻塞方式read网络IO),那么一旦KSE因阻塞被内核调度出CPU的话,剩下的所有对应的用户线程全都会变为阻塞状态(整个进程挂起)。

所以这些语言的协程库会把自己一些阻塞的操作重新封装为完全的非阻塞形式,然后在以前要阻塞的点上,主动让出自己,并通过某种方式通知或唤醒其他待执行的用户线程在该KSE上运行,从而避免了内核调度器由于KSE阻塞而做上下文切换,这样整个进程也不会被阻塞了。


2.3 混合型线程模型


用户线程与KSE是多对多关系(M:N), 这种实现综合了前两种模型的优点,为一个进程中创建多个KSE,并且线程可以与不同的KSE在运行时进行动态关联,当某个KSE由于其上工作的线程的阻塞操作被内核调度出CPU时,当前与其关联的其余用户线程可以重新与其他KSE建立关联关系。当然这种动态关联机制的实现很复杂,也需要用户自己去实现,这算是它的一个缺点吧。Go语言中的并发就是使用的这种实现方式,Go为了实现该模型自己实现了一个运行时调度器来负责Go中的"线程"与KSE的动态关联。此模型有时也被称为 两级线程模型即用户调度器实现用户线程到KSE的“调度”,内核调度器实现KSE到CPU上的调度


三种模型的示意图如下:
图片1


3. Go并发调度: G-P-M模型


3.1 G-P-M模型


有了上面的认识,我们可以开始真正的介绍Go的并发机制了,先用一段代码展示一下在Go语言中新建一个“线程”(Go语言中称为Goroutine)的样子:


// 用go关键字加上一个函数(这里用了匿名函数)
// 调用就做到了在一个新的“线程”并发执行任务
go func() {
// do something in one new goroutine
}()

功能上等价于Java8的代码:


new java.lang.Thread(() -> { 
// do something in one new thread
}).start();

可以看到Go的并发用起来非常简单,用了一个语法糖将内部复杂的实现结结实实的包装了起来。其内部可以用下面这张图来概述:
图片2
其图中的G, P和M都是Go语言运行时系统(其中包括内存分配器,并发调度器,垃圾收集器等组件,可以想象为Java中的JVM)抽象出来概念和数据结构对象:

G:Goroutine的简称,上面用go关键字加函数调用的代码就是创建了一个G对象,是对一个要并发执行的任务的封装,也可以称作用户态线程。属于用户级资源,对OS透明,具备轻量级,可以大量创建,上下文切换成本低等特点。

M:Machine的简称,在linux平台上是用clone系统调用创建的,其与用linux pthread库创建出来的线程本质上是一样的,都是利用系统调用创建出来的OS线程实体。M的作用就是执行G中包装的并发任务。Go运行时系统中的调度器的主要职责就是将G公平合理的安排到多个M上去执行。其属于OS资源,可创建的数量上也受限了OS,通常情况下G的数量都多于活跃的M的。

P:Processor的简称,逻辑处理器,主要作用是管理G对象(每个P都有一个G队列),并为G在M上的运行提供本地化资源。

从2.3节介绍的两级线程模型来看,似乎并不需要P的参与,有G和M就可以了,那为什么要加入P这个东东呢?

其实Go语言运行时系统早期(Go1.0)的实现中并没有P的概念,Go中的调度器直接将G分配到合适的M上运行。但这样带来了很多问题,例如,不同的G在不同的M上并发运行时可能都需向系统申请资源(如堆内存),由于资源是全局的,将会由于资源竞争造成很多系统性能损耗,为了解决类似的问题,后面的Go(Go1.1)运行时系统加入了P,让P去管理G对象,M要想运行G必须先与一个P绑定,然后才能运行该P管理的G。这样带来的好处是,我们可以在P对象中预先申请一些系统资源(本地资源),G需要的时候先向自己的本地P申请(无需锁保护),如果不够用或没有再向全局申请,而且从全局拿的时候会多拿一部分,以供后面高效的使用。就像现在我们去政府办事情一样,先去本地政府看能搞定不,如果搞不定再去中央,从而提供办事效率。

而且由于P解耦了G和M对象,这样即使M由于被其上正在运行的G阻塞住,其余与该M关联的G也可以随着P一起迁移到别的活跃的M上继续运行,从而让G总能及时找到M并运行自己,从而提高系统的并发能力。

Go运行时系统通过构造G-P-M对象模型实现了一套用户态的并发调度系统,可以自己管理和调度自己的并发任务,所以可以说Go语言原生支持并发自己实现的调度器负责将并发任务分配到不同的内核线程上运行,然后内核调度器接管内核线程在CPU上的执行与调度。


3.2 调度过程


Go运行时完整的调度系统是很复杂,很难用一篇文章描述的清楚,这里只能从宏观上介绍一下,让大家有个整体的认识。


// Goroutine1
func task1() {
go task2()
go task3()
}

假如我们有一个G(Goroutine1)已经通过P被安排到了一个M上正在执行,在Goroutine1执行的过程中我们又创建两个G,这两个G会被马上放入与Goroutine1相同的P的本地G任务队列中,排队等待与该P绑定的M的执行,这是最基本的结构,很好理解。 关键问题是:

a.如何在一个多核心系统上尽量合理分配G到多个M上运行,充分利用多核,提高并发能力呢?
如果我们在一个Goroutine中通过go关键字创建了大量G,这些G虽然暂时会被放在同一个队列, 但如果这时还有空闲P(系统内P的数量默认等于系统cpu核心数),Go运行时系统始终能保证至少有一个(通常也只有一个)活跃的M与空闲P绑定去各种G队列去寻找可运行的G任务,该种M称为自旋的M。一般寻找顺序为:自己绑定的P的队列,全局队列,然后其他P队列。如果自己P队列找到就拿出来开始运行,否则去全局队列看看,由于全局队列需要锁保护,如果里面有很多任务,会转移一批到本地P队列中,避免每次都去竞争锁。如果全局队列还是没有,就要开始玩狠的了,直接从其他P队列偷任务了(偷一半任务回来)。这样就保证了在还有可运行的G任务的情况下,总有与CPU核心数相等的M+P组合 在执行G任务或在执行G的路上(寻找G任务)。

b. 如果某个M在执行G的过程中被G中的系统调用阻塞了,怎么办?

在这种情况下,这个M将会被内核调度器调度出CPU并处于阻塞状态,与该M关联的其他G就没有办法继续执行了,但Go运行时系统的一个监控线程(sysmon线程)能探测到这样的M,并把与该M绑定的P剥离,寻找其他空闲或新建M接管该P,然后继续运行其中的G,大致过程如下图所示。然后等到该M从阻塞状态恢复,需要重新找一个空闲P来继续执行原来的G,如果这时系统正好没有空闲的P,就把原来的G放到全局队列当中,等待其他M+P组合发掘并执行。

图片3

c. 如果某一个G在M运行时间过长,有没有办法做抢占式调度,让该M上的其他G获得一定的运行时间,以保证调度系统的公平性?

我们知道linux的内核调度器主要是基于时间片和优先级做调度的。对于相同优先级的线程,内核调度器会尽量保证每个线程都能获得一定的执行时间。为了防止有些线程"饿死"的情况,内核调度器会发起抢占式调度将长期运行的线程中断并让出CPU资源,让其他线程获得执行机会。当然在Go的运行时调度器中也有类似的抢占机制,但并不能保证抢占能成功,因为Go运行时系统并没有内核调度器的中断能力,它只能通过向运行时间过长的G中设置抢占flag的方法温柔的让运行的G自己主动让出M的执行权。

说到这里就不得不提一下Goroutine在运行过程中可以动态扩展自己线程栈的能力,可以从初始的2KB大小扩展到最大1G(64bit系统上),因此在每次调用函数之前需要先计算该函数调用需要的栈空间大小,然后按需扩展(超过最大值将导致运行时异常)。Go抢占式调度的机制就是利用在判断要不要扩栈的时候顺便查看以下自己的抢占flag,决定是否继续执行,还是让出自己。

运行时系统的监控线程会计时并设置抢占flag到运行时间过长的G,然后G在有函数调用的时候会检查该抢占flag,如果已设置就将自己放入全局队列,这样该M上关联的其他G就有机会执行了。但如果正在执行的G是个很耗时的操作且没有任何函数调用(如只是for循环中的计算操作),即使抢占flag已经被设置,该G还是将一直霸占着当前M直到执行完自己的任务。


4. Goroutine与Channel: 锁之外的另一种同步机制


在主流的编程语言中为了保证多线程之间共享数据安全性和一致性,都会提供一套基本的同步工具集,如锁,条件变量,原子操作等等。Go语言标准库也毫不意外的提供了这些同步机制,使用方式也和其他语言也差不多。

除了这些基本的同步手段,Go语言还提供了一种新的同步机制: Channel,它在Go语言中是一个像int, float32等的基本类型,一个channel可以认为是一个能够在多个Goroutine之间传递某一类型的数据的管道。Go中的channel无论是实现机制还是使用场景都和Java中的BlockingQueue很接近。


使用方式


// 声明channel变量
var syncChan = make(chan int) // 无缓冲channel,主要用于两个Goroutine之间建立同步点
var cacheChan = make(chan int, 10) // 缓冲channel
// 向channel中写入数据
syncChan <- 1
cacheChan <- 1
// 从channel读取数据
var i = <-syncChan
var j = <-cacheChan

几乎等价于的Java中的操作:


TransferQueue<Integer> syncQueue = new LinkedTransferQueue<Integer>();
BlockingQueue<Integer> cacheQueue = new ArrayBlockingQueue<Integer>(10);

syncQueue.transfer(1);
cacheQueue.put(1);

int i = syncQueue.take();
int j = cacheQueu.take();

使用场景

a. 与Java的BlockingQueue一样用在需要生产者消费者模型的并发环境中。

b. 锁同步场景下一种替代方案。在Go的并发编程中有一句很经典的话:不要以共享内存的方式去通信,而要以通信的方式去共享内存。在Go语言中并不鼓励用锁保护共享状态的方式在不同的Goroutine中分享信息(以共享内存的方式去通信)。而是鼓励通过channel将共享状态或共享状态的变化在各个Goroutine之间传递(以通信的方式去共享内存),这样同样能像用锁一样保证在同一的时间只有一个Goroutine访问共享状态。但这的确需要转换以前用锁做并发同步的思维方式,大家觉得那种适合自己和自己的使用场景就用哪种好了,并不能很简单、绝对地说哪种方式更好,更高效。


5. Go语言对网络IO的优化


在谈论高性能网络IO编程时,我们几乎都离不开epoll/kqueue/iocp等技术术语了,如Java最新的的NIO,Node等等的高性能网络IO模型都是基于这些技术实现的。诞生于21世纪,有互联网时代的C语言之称的Go语言,这么重视高并发,当然不会放过对网络的优化。且Go语言中对网络IO的优化很巧妙,让你可以用和以前一样的(同步的)思维方式去编程的同时(而不是反人类的异步方式),还能享受到与异步方式几乎同等高效的运行性能。那Go语言中是如何做的呢?主要是从两方面下手的:

a. 将标准库中的网络库全部封装为非阻塞形式,防止其阻塞底层的M并导致内核调度器切换上下文带来的系统开销。

b. 运行时系统加入epoll机制(针对Linux系统),当某一个Goroutine在进行网络IO操作时,如果网络IO未就绪,就将其该Goroutine封装一下,放入epoll的等待队列中,当前G挂起,与其关联的M可以继续运行其他G。当相应的网络IO就绪后,Go运行时系统会将等待网络IO就绪的G从epoll就绪队列中取出(主要在两个地方从epoll中获取已网络IO就绪的G列表,一是sysmon监控线程中,二是自旋的M中),再由调度器将它们像普通的G一样分配给各个M去执行。

Go语言将高性能网络IO的实现方式直接集成到了Go本身的运行时系统中,与Go的并发调度系统协同高效的工作,让开发人员可以简单,高效地进行网络编程。


6. 总结


Go语言并不完美,它是以软件工程为目的的语言设计。其实现的并发机制也并不是什么革新的技术,只是将这些经典的理论和技术以一种简洁高效的方式组合了起来,并用简单抽象的API或语法糖开放给开发人员,着实减轻了开发人员编程的心智负担。而且其通过引入channel机制,将另一种并发编程模型(CSP: 通信顺序进程)带给了我们,给我们提供了使用其他并发编程思维方式的机会(有关CSP模型建议大家看看《七周七并发模型》这本书的第六章),Goroutine与Channel的组合是一对很有powerful的并发工具,相信其可以给你带了更好的并发编程体验。


7. 参考


《Go并发编程实战》 第2版

《Go语言学习笔记》

也谈goroutine调度器

Go coding in go way