题记

考虑到本系列文章有部分新的读者，所以关于本系列文章名字的起源就不再赘述了，见这里《"星霜荏苒"名字诞生记》

当你看到这篇年终总结的时候，距离我上一篇年终总结整整过去了 500 多天。你一定很好奇这 500 天我干了哪些事情，为什么这篇文章的标题叫这个名字？

---

学生等级

为了解释文章标题，就要先解释 10 个词语。笔者将学生等级从高到低排序是：

等级	名词	解释
1	学魔	对学习走火入魔，癫狂状态，不做题会死掉。
2	学霸	隐匿在人间有头脑的高智商人物，社交范围广泛，融合契合度高，琴棋书画样样精通，高端大气上档次。
3	学神（学帝、学仙、学圣）	高大帅气，青春靓丽，不食人间烟火，天天游走在高难度的练习册当中却依然风华正茂。
4	学痞	他们上课睡觉，下课玩闹，但他们的成绩仍然很好。
5	学婊	每天都在玩，几乎不学习，每场考试结束后第一时间宣布自己要挂科了。但是考试成绩出来后，门门都拿第一。
6	学民	智商均衡，膜拜学霸，却瞧不起学渣等人物。他们只有一个信念，总有一天超越学霸，因此艰苦奋斗。
7	学弱	他们因为没日没夜地熬油点灯，已经身体虚弱，不堪重负。
8	学渣（学灰）	智商处于半疯癫状态，兢兢业业，刻苦学习，却总是不得志。
9	学残	智商处于全疯癫状态。他们已经被学习折磨得痛苦不堪，没有人样。
10	学沫	智商不够用，却也不是很努力，每天在混着日子。总是觉得能够不劳而获。
11	学水	已经不能用智商与努力来评判他们了，他们已经自甘堕落，自暴自弃好多年。

笔者这一年和学魔，学霸一起学习，经过多次考试的锤炼，最终认定自己是学民。有时候人与人之间学习能力的差距，你不得不认。笔者给自己的定义是，认真学习但也非刻苦的程度。分数不低但也不是高分。

---

好了，至此我关于标题的解释都述说完了。接下来聊聊今年学习和生活中遇到的一些所见所想。还有读者好奇的我这一年在忙什么。本文都会一一讲清。至于每个人对自己的人生都有自己的选择，笔者的选择不一定对。在强者面前，我的选择可能就如小丑表演一般，滑稽而可笑。那本文就献丑了。

工作的变化

今年组里一个大佬离职了。于是我有机会带领 5-6 名工程师一起向前冲。当真正 owner 一个超大项目以后才能感受到皇冠的重。各种事情都需要亲力亲为的开会，沟通，拿决策。每个组都有各自的利益，会议桌上各自都怀着自己的目的，所有人都想要自己的利益最大化。如果是因为我的“谈判”失误，导致影响到了组里全年的绩效，我会非常自责。不过结果来看，马马虎虎。我们吃掉了好几个组的业务。回收了他们部分 KPI 绩效，也成功拿到了他们的机器资源。接下来打算分享几件个人觉得比较成功的案例：

（这部分的分享本来有 3000 字左右，但是由于读者看到这篇文章的时候，已经是过去快 2 年的事情了，有旧事重提“炫耀”的嫌疑，加上笔者后来也从这家公司离职了。故笔者把这段删除了。）

职业生涯的思考

去欧洲旅行，遇到德国人讲英语，我实在不习惯他们的口音，基本都听不懂。双方无法交流。特别尴尬。但是这次我还自己给自己找理由，毕竟是有德国口音，听不懂也正常。回国以后我也忘记了这段尴尬。

去澳洲旅行，遇到意大利人说英语，能听懂他说什么，但是当对方笑起来的时候，我并不明白他的笑点梗在哪里。这一次我又安慰自己，文化差异导致我们无法交流。回国以后我有意识的去了解了一下当地的风俗文化，不过一段时间以后，我又忘记了这段“痛”。

去迪拜旅行，和中东人聊天零星听得懂，但是自己说话对方听不懂，冲沙项目中有一个人英语可以和对方交流，于是她成为了团队的中心。我心里暗暗记住，我也要成为队伍的中心，这次经历彻底击碎我的底线。学习英语迫在眉睫。回国以后便开始制定学习计划。不然下次旅行还会被英语卡。

职业生涯如果以 5 年为界限，笔者即将到达第一个 5 年了。那么第一个 5 年做了哪些成就呢？这 5 年我对我交出的答卷并不满意。兜兜转转把客户端，前端，后端都摸了一遍。如果 5 年都专注一个领域，可能早就可以升到某个领域的技术专家了。既然开局稀烂了，那第二个 5 年必须做出一点改变，“扭转乾坤”。

在中国改革开放的大环境下，中国的市场涌入了大量的外企，给市场注入了很多新的思想。在这个地球村的时代，全球旅行不再是梦想。我已经打破了旅游的地理限制，我已经走遍了 30 多个国家，5 大洲遍布了我的足迹。接下来我在考虑，我能否打破工作上的地理限制呢？比如我能否在全球任意一个国家，通过我的本领找到一份能赚钱养活自己的工作？语言是必须优先解决的。工作中沟通交流必不可少。在我脑海中这个想法还没有成型的时候，突然某一天看到脉脉上双非本科毕业生晒 Google 的工作日常。确认过信息真实性以后，我也有了去硅谷打拼几年的想法，我要打破工作上的地理限制。我不是 985 毕业，但是依旧有自己的梦想。自己实力很菜，去不了 Google，去一个 startup 应该还可以。于是我萌生了去海外工作的想法。但是此时这个想法不是很强烈，还在徘徊中。

慢慢的，我又经历了一些事情：

人是一个社会属性的生物，他的一些决定是参考了社会因素的。例如，家长们在一起聊天，A 家长对你爸妈说，“你看你们家的孩子这么没出息，你们怎么教育的？”，或者“你看我们家孩子，年薪 5000 多W，你们家孩子年薪才 10W，干一年顶你们家孩子干 500 年。”一般这些话爸妈听了，脸面上通常会一笑而过，也许不会和你说。但是这话要是自己听到了，肯定不是滋味。由于自己的不努力，或者不够成功，导致了自己的父母在外面被其他家长“踩”。凡是有上进心的人，一定会采取一些“绝地反击”的措施吧。“我是全村的希望”，这句话看上去那么的骄傲，背后其实反映了一种自豪，一种努力，不愧对父母的养育，自己的成功，自己的出类拔萃，也让父母在别人面前出人头地。当然不少父母也是低调的，自己孩子多么成功也不会在外人面前吹。但至少，这个孩子的成功没有给其他家长“踩”自己爸妈的机会！

我通常在同行面前都会说自己是菜鸟。久而久之，大家觉得我带坏了一些风气，装弱。有些人也会觉得我是谦虚。我的花名是，霜菜，提交系统的时候，我赋予了这个花名一个含义，菜的含义是，谦逊为人，低调做事，山外有山，人外有人。读书读的越多，就会发现身边的同学都是优秀且不带优越感的人，他们明亮不刺眼，自信又懂得收敛，让你仰慕的同时又能给你能量。仔细反思，我觉得自己也许根上并不是谦虚，更多的可能是自卑。大学毕业以后，我因为不是 985 的学历，被某些独角兽公司扔简历到了垃圾桶，没有给面试机会；因为不是 985 的学历，被牛人鄙视是垃圾。考研报考了 985 学校，也因为种种原因最终失败。社会一次次的实打实的挫败着我，985 已然成为了我心头上一道深深的伤疤。这道伤什么时候会揭掉，我不知道，这道沟壑什么时候会翻篇，我也不知道，我唯一明白的是，我俯下身子，在地上爬，不想让大家看见我破损不堪的心灵，既然是垃圾了，还有什么资格平起平坐或者高人一等？

我遇到了一个 985 大佬来自学历的“鄙视”，我仔细反省以后，我觉得必须证明一下自己，打败内心不自信的“心魔”。这个心魔算是我职业生涯第一个 5 年最后的一个大 boss，也是职业生涯第二个 5 年的第一个 boss，所以我下定决心必须过了这一关。想证明自己不输于 985 学生，想证明自己的实力，需要从某些方面来证明。正好那段时间看见了一段话，“看程序员是否勤奋就看他的英语好不好，智商高不高就看他算法好不好”。那么我就觉得在这 2 方面上证明自己的实力。心里默默下定决心：等我下次面试，我一定也要去面“鄙视”我的那位 985 大佬所在的厂，当我作为他的同事，职级还要和他平起平坐。希望到了那个时候，他能不鄙视我非 985 的学历。人活着有时候就是为了一口气，你若不认输，不服输，不愿意承认自己是 loser，不愿当学渣，那就用行动证明给嘲笑过你的人看，去证明你也有“过人之处”~我就想用成绩来证明给天下人看，“虽然我不是学霸，但是我决不是学渣！”，我想撕掉心灵上学渣的这块疤。

以上就是我从迪拜回来到 7 月份这期间的一些心路历程。从萌生去海外工作的想法，到自我反省，自我认知，最后到下定决心证明自己。所以从 7 月开始，我决定开始了英语雅思和托福的备考。从今年 3 月 18 日开始，我开始刷 LeetCode。英语 + 算法两手抓。

在英语备考期间也有来自各个大厂大佬的面试邀请，有来自滴滴，腾讯，头条，阿里巴巴，百度，拼多多，美团等等大厂的面试邀请。（此处对 @孙源 大佬说声对不起🙏🏻，邀请了我好几次，我都没有说明原因，有些话当时实在不方便说出来，现在如果你看到这里，还请谅解我啊）也有来自大厂 HR 的面试邀请。我在这里和你们说一声对不起了，当时回复你们的都是“我有我自己的安排，对不起”。其实我是想去海外大厂干一段时间，多成长成长，和世界上最优秀的工程师切磋切磋，再以最好的状态入职大厂。看到这里，就是我一直隐藏着的答案。

此处也需要 @子奇 和 @萧玉，这两个大佬也邀请我面试很久了。我一直委婉拒绝，可能由于和你们太熟了，导致我没法去你们那入职。🙏🏻还请 2 位大佬原谅我的当初的拒绝。还有 @阿里云大佬，@淘敏，@闲鱼大佬，@宗心的面试邀请，我实在非常不好意思。还有很多大佬也私信邀请过我，此处没法一一@，如果你们看到这里了，就请原谅小弟我吧。有不少人说我走了一步错棋，因为没有加入你们的团队。在此我也一并说声抱歉吧，小弟不加入只因为我觉得我还不配加入你们，你们都非常优秀，我还太菜了。人生还长，当我修炼好自己以后，未来再加入你们的机会很多！

在提升英语的期间，也有太多的“诱惑”，有来自猎头的“嘲讽”：大概 19 年 7 月的时候，有一个猎头加我，说我工作快 5 年了，还没有到 P7，技术有点垃圾 ，在阿里也呆了 2 年了，可以出去看看机会。我当时心里好“憋屈”，我当时就想证明一下自己的实力。不过有“心魔”的我，当时也只能忍了，宛拒了，“我的技术实力还不行，面不上头条，我还需要在这里再磨炼几年”。朋友圈也有来自旅友的美景照片，有来自日本旅友的旅行邀请，有来自欧洲旅友的旅行邀请。有来自美洲朋友的旅行邀请。在没有战胜我内心心魔的时候，实在没心思去玩，但是每天刷到美景照片，心里实在痒痒。于是，我还是选择关闭朋友圈，减少诱惑。但是又发现旅游群里面还有诱惑，于是屏蔽群消息，又发现还会有一些私聊的诱惑，最终选择关闭微信了。这和某学习 app 每日推送说的一样。“微信不必每条都看，但是单词不能一日不背”。（不过好像不用微信，对生活好像影响不太大，工作中重要的消息都在钉钉上，和家人聊天都用的 iMessage，逢年过节上微信和朋友问候一下，发发红包送祝福）

我一度删掉了微信好几个月，屏蔽了所有外部消息。这也是为什么好多人发我消息，我都没回复的原因。并不是因为忙，并不是因为不想回，而是因为我没用微信了。微信上诱惑太多，加上我自己心理调节能力弱，我直接断舍离。微信群里经常会有人晒百万高薪，晒千万跑车，晒万亿豪宅，看了以后要么自己会酸柠檬，要么就恨自己有多没出息。多多少少心理上会有一些波动。这些心理波动对备考时期没有任何好处，至少对我来说。删掉微信以后，我每天只想我自己的事情。心理上至少做到了不以物喜不以己悲的境界。（每个人备考状态和自我调节能力不同，我的心路历程也许不能复制，只是写出来给大家参考，也可以当做是“笑料”，给大家笑笑。）

人需要沉淀，需要“静养”，弹簧的姿态压的越低，之后弹的越高。有些人说我是一个很自律的人。不过恰恰相反，我自己认为我是一个不太自律的人。一个大师和说了三个字，解决我不自律的问题，“戒，定，慧”。让我戒掉一些东西，节约时间，心定下来，产生一些学习的定力，最终会产生智慧。在群里聊天我会很不自律呢。群友的问题我看着就想回答。也可能工作 5 分钟，聊天 3 小时😂🙏所以我决定克制一下。

就这样，我朝着去海外工作的目标狂奔着。

决定留学

在准备学英语之前，我去某英语培训机构裸测了一次雅思水平，居然才 5.5 分。扎心的是，老师还安慰我，“你快工作 5 年了，可能平时用英语不多吧，来上我们的课，带你提高英语水平”。我平时工作上每天打代码都是英语啊，这心里落差把我打至谷底。现在回头想想，这个英语培训机构给我的评分是真实的么？会不会故意打低分，“骗”我报班的套路？毕竟我没有参加过一次真实的雅思考试。再后来我就报班了。报了托福的培训班。😂由于雅思裸测 5.5 分，老师在课上叫我五分选手，（难道四舍五入不应该叫我六分选手么？）我经过一年的努力，我已经成功变成了八分选手。为了自嘲自己的超低水准以及提醒自己记住这段痛苦奋斗的日子，我现在仍然自称自己是五分选手。（这也是公众号名字的来源）

也因为雅思考的分数不高，使得我“怒转”复习备考托福。托福的口语是人机对话。你面对的是没有感情的机器，它不会因为你说错而产生尴尬气氛。雅思是人人对话，和真人对话时间，你说的如果对方听不懂，那会非常的尴尬。而且雅思听力里面也有我不擅长的听数字，填空等等题目形式。

这时候我也发现，一步出国工作可能有点难，可以分为两步走。先在国内的外企工作一段时间，再转去外国本土工作。国内的有名气的外企有 FLAG，开始投简历咯。我开始刷题，准备英文简历，冲击外企。现实给我泼了很大一身凉水。收到了 HR 给我的反馈，“同学，对不起，这个岗位你的学历背景差了一点，我们的候选人目前都是清北交复的硕士博士。希望未来你还有机会加入我们，谢谢”。我不会在这里公开这是哪家企业，面试岗位不是 leader 岗，只是高级 SDE 岗位而已。我接到这个回复以后，脑袋飞速运转，“可能是我的技术实力不行，用学历差了这个理由婉拒我”“有可能真的是学历不定，看 HR 的语气，要招硕士起步的人才”。不管是哪个原因，我的世界都乌云密布。既然先在国内外企工作一段时间再办签证去海外的这条路已经不通了。那么只剩下通过留学这条路去海外了。到此时，笔者才确定了留学的这条路了。

今年 7 月 13 号，滑滑鸡即将前往 CMU 读 Master 了。我和瓜神一起和他吃了自助餐，送他去美国。餐桌上他向我传授了托福裸考 108 分的经验。这次吃饭是经典的“西瓜霜”聚餐。滑滑鸡的名字中带有“西”的同音字，瓜神就是“瓜”，而我的名字中带有“霜”。餐桌上我也许下了诺言，2020 年会去美国找滑滑鸡一起吃饭。如果有缘的话，更希望能成为 CMU 校友。希望我能早日兑现男人的承诺。

写书

很多人发现我从 2019 年就开始刷 LeetCode，其中有一小部分人怀疑我要跳槽，我的同事起初也怀疑过我要离职，但是看到我刷了半年以后都没动静，也就慢慢相信我刷题是因为爱。这一小部分人猜对了真相，不过我笑而不语。同事也都不明真相，真实情况是因为我没有通过面试，为了“掩盖”这个事实，我干脆一不做二不休继续刷。刷满一年以后，我就没有日日不间断了。现在我还隔三差五的刷 LeetCode 真的是用爱发电。刷算法是热爱，刷到世界充满爱！

我写书也是受到了 @欧长坤 大佬的影响，他的两本经典开源书令我收获颇丰。于是我也想写点开源书，回馈社区或者单纯分享知识。由于今年全年我都在复习英语和刷 LeetCode，于是决定写这两本书。

这两个本书我都放在 github 上迭代更新。至于什么时候会印制成纸质实体书，还不确定。众所周知我目前还是一个 gopher，所以这两本书的网站必然要用 hugo 搭建。第一本书我用的是 wowchemy 主题，第二本书我用的是 hugo-book 自定义主题。项目代码也都开源在 github 上了。

写开源书（让我姑且称它为书）真的非常花时间，书和一篇博客不同，博客的目录只是单篇文章的主线，但是书的目录就不同了，它是整个知识体系的主线。笔者时常写到某篇文章的时候，突然想到另外一篇文章有问题，又去调整前面写的好几篇文章。为了每天能记住 200-300 个单词，我选择每天早上 6 点早起，先把 LeetCode 每日一题写完，然后把解题思路，代码，测试文件 push 到 github。如果快的话，6 点 30 分左右都能搞完。接下来吃早餐，7 点到 9 点是刷托福 TPO 和背单词的时间。我会把阅读的文章翻译一遍，记录和分析错题，精听听力文章等等。做完练习再把心得和方法等等内容 push 到 github。9 点半左右出门去公司上班。

常看 O’Reilly 动物书的同学一看这个封面就知道是向他们致敬。确实是这个目的。O’Reilly 的封面动物都是稀缺动物，并且画风都是黑白素描风。第一本书的封面上的动物是 Coyote。经常听托福听力的同学一看到这个单词就会觉得特别亲切了。之所以选这个动物作为封面也是这个原因。Coyote 在托福听力生物类中出现的频率比较高。既然此书是动物书，又和托福有关，那么选 Coyote 是理所应当了。

Coyote 翻译过来是土狼，或者郊狼。郊狼（学名：Canis latrans），也叫草原狼、丛林狼、北美小狼，是犬科犬属的一种，与狼是近亲。郊狼产于北美大陆的广大地区，北起阿拉斯加、南到巴拿马。欧洲探险家最初是在美国西南部发现这种动物。郊狼一般单独猎食，偶尔也会组成小型的群体。平均寿命为 6-10 年。郊狼在其大小、颜色和头部形状都十分相似濒危的红狼。 其英文 coyote 一词来自中美洲阿兹特克等部族所用的纳瓦特尔语单词 coyōtl，后经西班牙语传入到英语。

第二本 LeetCode 这本书的封面动物是孔雀。孔雀开屏的意义是希望大家刷完 LeetCode 以后，提高了自身的算法能力，在人生的舞台上开出自己的“屏”。全书配色也都是绿色，因为这是 AC 的颜色。

这两本书我会一直保持更新，直到我的托福考到一个理想的分数，LeetCode 刷到 500 题。当满足这两个条件的时候，便是你看到书的时刻了。

当你在读这篇文章的时候，第二本书应该已经开源了，所有代码都在 github repo 中，并且也是 public 的。但是第一本书可能难产了。并非笔者不想开源分享，而是笔者的托福分数没有考到满分。有一天去知乎上看了一眼，个个都是托福 120 分满分选手。我这种英语垃圾选手还是低调的找个地缝藏起来了。所以笔者这辈子都不打算开源第一本书了。

既然第一本书不打算公开了。那在这里放一些它的截图吧。记录一下今年我的一些努力时光。下图是 github private repo。

接下来几张图是用 wowchemy 主题做的这本书配套的网站。

艰辛的托福备考

实话说，备战托福对我来说，是比较花时间的。我妹子全程目睹了我“无比艰难”的备考过程，尤其是在职那段时间，备考太艰难了，加班到半夜到家以后还要开始读英语，睡觉都是听着 TPO 睡的。那段时间我还经常被公司安排 on call，很多次都有离职的冲动，换一个闲一点的公司。（最后还是咬牙坚持把一年干完，拿完年终奖走人了。）我妹子也从来不鼓励我，反而她一直都在劝我放弃，（这难道是反向操作的鼓励？），她经常说：“你看看人家滑滑鸡，不到 6 个月解决 GT 2 门，还都是高分，你呢？6 个月 GT 还没考到人家分数的 80%，还努力啥？坚持啥？”，我确实有过放弃的念头，但是一看我的学托福学雅思的时间投入，我就不能放弃了。走路都走一半了，这个时候放弃，过去的好几千小时岂不是白费了？大家都说，凡事要享受过程，不要在乎结果。“功利”的我偏偏非常在意结果，人一辈子的时间是短暂的，投入产出比不高的事情，优先级可以降低。于是我这个垃圾靠着不能放弃宝贵时间的前期投入，一直咬牙坚持。

开始我还不承认我是学渣，我觉得自己还算努力，算学弱应该不为过（学弱：学习很努力，但是分数很低）。经过几次 GT 考试的摧残以后，我自觉的把自己的身段从学弱放回了学渣。在我妹子眼里，滑滑鸡就是100%准学霸。这点我也心服口服。在被托福和 GRE 考试无情摧残了 4 个月以上时间的我，每次看到滑滑鸡的朋友圈，或者和他微信聊天，我内心真的充满了对学霸的敬佩。（如果有不服气的，可以立即报考托福和 GRE，3 个月内能达到托福 110+，GRE 330+ 都是学霸。）滑滑鸡这种学霸型，自己高分杀 G 砍 T，期间还有额外精力去实习赚钱，刷 GPA，四线操作切换自如。不仅无开销，还赚了几十万。有时候我一个人在家备考的下午，我就会翻翻自己的留学时间线和花销记录小本本，我和他的差距已经不是学渣和学霸这两级的差距，我觉得说我们相差四级都不为过😭这可能就是双非大学毕业的学渣中的学渣和顶级 985 毕业的学霸中的学霸的巨大差距吧。在托福和 GRE 这两门考试上，我是输的心服口服，我被虐的体无完肤，毫无脾气。

大家也都别参考我，我英语太垃圾，学习能力也很差。在职的时候公司加班比较忙，每天下班以后还要刷 LeetCode，写博客，写书，背 500 个托福单词，刷托福 TPO，背 GRE 单词，刷 GRE 题目。我实在是 hold 不住了。如果每天都要完成这些任务，每天我的睡眠时间只有一个小时。最后决定托福 + GRE + LeetCode 是每天必须完成，写博客和写书放在周末完成。由于在职复习和学习进度有点慢，所以我到今年年底也没有提交申请学校的申请。

虽然我的托福没有考到满分，但是依旧可以有一些经验可以和读者分享。

考试心态

托福考试是一种能力的测试，不是想办法出难题刁题（TPO只遇到过一次）为难大家。同时，考试也没有大家想象的那么难，我和考友的交流发现，考试改革以后，容错率仍然是挺高的。所以大家在对待考试的时候，不要对它产生惧怕心理，要藐视它，相信自己的付出一定能得到相应的分数。同时，是考试就有应试的技巧，无论是报课程，大家的攻略，还是每个人自己复习时候的心得，都是大家找到考试诀窍的方法。一定要对托福考试有一种熟悉感，一种我知道你想出什么题目，你在哪里出题的熟悉感，预判出题者的预判。这需要复习的时候有一颗热诚的心，要像对待女孩子一样了解她，学会去预判她的想法，做她希望你的事，而不是浪费精力做无关的事，或者作死。所以请大家复习的时候要有安排，有技巧，把精力放在最容易有效果的地方，多拿一分是一分。

托福考试分四个科目，读听说写四个方向，其实每一个方向都有一个共同的基础，那就是词汇。词汇不过关，100 分就是虚无缥缈的幻想。所以第一个月一定要沉下心来好好背单词。单词就用 KMF 词汇 app 和默默背单词，我是工作党，利用闲暇时光，目标是一天 300 个单词，白天争取在午休和吃饭的时候背好，不求背了就永远不忘，因为你会要背 3-4 遍。背单词其实是一种加速的过程，到后面几遍的时候，可能需要背的单词量也就一半不到，所以不用担心。

而听力是从 80 分到 100 分的最终助力，其实除了阅读，每个科目都和听力息息相关，每门课都要拿到高分，那需要能够精准的反映出听力的信息。我首考 74，我觉得很大的原因，除了第一次考试紧张和一些突发意外，最重要的就是没有重视听力，没有做好仔细听的心理准备，到时口语和写作综合题没有踩到所有的得分点，因此，大家一定要心理上要重视听力，考试的时候提醒自己保持听力注意力。无时无刻保持警惕感，去找到出题的点。

单词

我背单词都是用零碎时间背的，例如，上下班等车，坐地铁，午饭后散步，晚饭后闲逛。这些零碎时间都可以用来背单词。背单词不要一个单词记 5 分钟。正确的做法应该是多重复。一个单词看 20 秒就过，一天多重复 3-4 次。重复次数越多记的越牢。试想，情景一，一个陌生人和你打个招呼聊天 10 分钟，一年以后你还能记住他么？情景二，一个陌生人和你打招呼，每天都聊 1 秒钟，连续 365 天天天和你相见，一年以后你还能记住他么？很明显，每天都重复一定会让你记住他。以下是我的单词 app 的刷词记录：

上图是笔者重新更新的截图。到 2019 年年底，连续打卡天数只有 190 天左右。笔者先用考满分单词 app 刷完所有托福词汇，并开启第二遍刷单词，差不多 10 月份的时候，开始用小站单词 app 开始刷它。当我把小站 app 所有单词都刷完一遍，我又开始用默默背单词了。如果你问我哪个 app 最好用？这个问题我其实回答不上来。因为我所有都用过，记忆单词有累加效果，并非某单一 app 促使我记住所有单词。其实这些单词 app 都差不多的。建议读者都试试，找一个适合自己的，刷起来吧。

时间规划

托福备考最好能速战速决，拖的时间越长，备考效率也不高。一般以 3-4 个月为周期最佳。

• 第一个月：夯实基础阶段。

这个阶段是一个准备期，就是背单词，每天争取 300 个词。背单词的同时，可以开始做阅读 TPO 了，一天一套的练习。

• 第二个月是复习高峰阶段。

除了写作和口语综合题以外，所有科目都要每天分配一定的时间去完成每日目标，单词继续 300 一天，阅读一天一个 TPO，听力一天一个 TPO，口语独立一天一题。这段时间是最痛苦的时间，因为程序员工作比较忙，大家一般都 23 点左右，如果加班，到家就 1-2 点了。所以睡前一定要合理安排好工作生活与学习。每天看着时间流逝，还是很抓狂的，但是大家一定要坚持下来，而对于学生党，这些工作量应该是很轻松能完成的。

• 第三个月是冲刺阶段。

全题型都需要冲刺的阶段，重点可能会放到口语综合题和写作上面，一天的时间单词继续 250 一天，阅读一天一个 TPO，听力一天一个 TPO，口语独立一天一题，口语综合题争取一天一套（我的实际情况是考前大概做了 30 套口语），写作两天一套（考前做了 5 套）。至于每天的具体时间安排，因人而异，我可以说说我的周末安排。

​ 上午无论几点起床，一套阅读一套听力的 TPO 的模考和错题回顾。

​ 下午午休，醒来后一套写作。

​ 睡前完成口语 TPO 一套。

​ 其中找时间穿插背一背单词和背一下语料（口语和写作通用）。

有人觉得一天需要做两套 TPO，我觉得很难实现，毕竟做错题还要检查为什么错，基本上一天一套 TPO 差不多了，何况 TPO 资源有限，不能浪费，必须做一套反思一套。接下来单独说说 4 科的复习方法。

阅读

阅读是拿分重点，词汇是基础，所以单词一定要背几遍。

复习的时候，学会去找同义转换，要相信考试的时候，绝大多数题目还是在做原文的同义转换，快速定位关键词 KEYWORD，将原文和题目选项对照，所有题都做排除法，基本能保证很高的分数。我有两次考试，最后三题剩的时间都不多，最后分数 28，一方面说明同义转换的正确定，也说明了考试的容错率。

每篇 TPO 阅读做完以后可以写一写反思。错题错在哪里了。然后精读一下文章，把长难句和不懂的单词查一查。笔者大概精读了 30 篇左右。考前 TPO 阅读都做完，因为有可能出现阅读题目变成听力题目，所以 TPO 刷完，最后问题不大。

听力

听力是从 80 分到 100 分最重要的助力。听力题是有容错率的。拿我的考试经验看，每次到时会遇到七八题需要思考一下二选一，两次考试分数都是 23 分，比自己预想的要高。听力复习最重要的是多听。

1. 通过多听，培养英语的声音结构。

听力不是一个个单词听，每句话都有自己的主谓宾，考试考点无非就是考察主语或者宾语你有没有听到，少量题目是对整体或者段落的态度理解。

2. 通过多听，熟悉段落结构，养成对出题点，问题，语气词，转折词的敏感性，在考试中助力自己能在出题点前提高自己的耳朵注意力。

每天一套 TPO，加考后精听，精听是边听边思考上面的两个要点，一遍通过。靠前两周，做到 1.2 倍速度做题没问题，复习平均分基本就是考试分数。做题的时候，坚持到底，不要因为一句或者一段没听清楚就放弃，很多情况下，听力题目在不理解听力素材的条件下也能回答。但是要做到有连续听六题的忍耐力和注意力。TPO 题目现在有可能成为你阅读的题目，所以大家一定要做完。

关于笔记，我的历程是一文章五六个词，到一文章记得密密麻麻，到最后考试前一文章不超过 10 个词，其实最重要的内容是记下文章观点和逻辑。关键词，术语，不重要。考前听力 TPO 都做完，问题不大。分享几个听力的笔记。有些同学不做笔记也可以拿到高分，所以笔记并不是必须的，因人而异。

口语

考前一个月时间准备口语。如果你的口语真的说的不溜，可以考虑报个口语班，多练练，我只能说，语料是真的好用，2 个月的考试，独立口语会命中相似的语料。班上对每种题型的理解也比较清晰，也提供了答题需要注意的结构和模版，挺好用。

具体复习的时候，口语只能靠自己多说了，而且要厚着脸皮说，因为考试的时候，自信心非常重要，你身边可能有比你说的不好的，可能说得和老外一样好，这时候你如果心态疲软，会出现答题问题，越说越不自信，越说声音越小，越说语言越枯燥。

独立口语就是语料的积累，我为自己准备了多种语料，能面对交流，文化，科技，环保等内容的时候，有话可说。同时积累逻辑词，因果关系，递进关系，转折关系每种关系准备两三个词，通过结构和语料，共同搭建起自己一分钟的语言轰炸。考试的时候，保证自己能说完整完一个观点，包含态度，原因，举例，反方，总结。

综合口语更重要的其实是听力，能把考点都记下来。所以这种题型相对简单，考题结构也比较稳定，比听力题简单点。最后通过模版或者自己准备的逻辑词把重要考点说出来就可以。

关于发音练习，笔者这里推荐一个免费课程，lisa美语音标发音教程中文字幕（4.0高级技巧）

写作

写作我也不知道要不要建议大家准备模版，因为我第一次用了模版是 21，第二次没用模版是 24。当然，这也是因为我综合写作听力有一定专门的训练，所以可能是综合写作分数往上提升。

有模版，写起来会比较简单，字数，高级词汇也会有一点。

不用模版，也不是不行，因为考试最重要的是看你的大逻辑和提供的论点是否切题，是否符合你的态度。

另一个关于打字数度，我可以给大家一个参考，独立写作 380 字，综合写作 220 字，分数也是能拿到 24 了。有些大神就 600 起步的，最猛的有打到 700 字的，（这手速要多快？）我真的很佩服，他们拿 30 分合理。

关于拼写，我感觉这个对分数影响是很大的，考前要做到不超过 5 个拼写错误。平时不要用带拼写检查的软件练习写作！语料同口语一起准备，考前两周，做到两天一套 TPO，保证自己写作评分能在 4 分稳定，问题不大。写在最后，首考的同学要做好心理建设工作，因为真实考试和模考是会有不同的。听力有时会出现英式英语，口语第一题会有一种突然开始的感觉。

最后，复习到今年年底，笔者身边很多同学说自己撑不住了，是的，这确实是备考托福的过程中，最难熬的一段时光，就像在一条，狭窄的隧道中，前面只有一丝光亮。而后面，漆黑一片，早已没有了退路。放弃是不可能了。但想拼却又觉得就那点光亮，值得么？你会觉得别人都是在阳光下奔跑，十足的胜算，你的拼搏，都变得卑微，于是你开始犹豫，你每天也在看书，但是却没有了对于美好未来的一丝憧憬，你只是在想这段路，早点结束吧。“放我出去”，是你唯一的期待。但是你却忘了，这世界上没有人是容易的。哪有什么阳光下的奔跑，都是在这条隧道中，艰难前行。未来有一天，那些所谓成功的人，回忆走过的路时，他们一定会提到，当年他们也在一条隧道里，艰难地拼搏。你才突然想起，这个地方我也去过啊，狭路相逢勇者胜，不要妄自菲薄，未来成功投射到当下，只会是一丝光亮，每个人都一样，没有阳光明媚，只有微光一点，但你的努力，本就光芒万丈，你忘记了，这一路走来，不是未来给了你希望，而是你一直在给自己力量，你要去拥抱的，不是什么狗屁成功，成功就是一个贱人，他只会依附于强大的人，你要去遇见的是那个更好的自己，那个绝不服输，决不放弃的更强大的自己。其实，你看到的那一点光亮，也是那个自己给你的，不要让在隧道尽头等着你的那个自己失望，因为你要是放弃了，她就等不到你了，而成功这个小人，也会离她而去，拼搏吧！燃烧吧！去看见，去遇见，去拥抱，然后有一天你带着成功一起讲述，你就是从隧道里寻着一束光，找到她的。然后终有一天，你可以笑着去讲述那些曾经让你哭的瞬间。💪

关于旅行

相比 2018 年去了 5 个国家来说，今年大幅减少了。利用五一假期去了迪拜，本来十一打算去美国常青藤学校看看，但是由于笔者托福备考进度慢了，所以十一没有出去，旅行的时间都用来上复习托福考试了。这是 2018 年年终总结里面写的新年愿望，现在看看，只完成了一部分，以后还是少立 flag，计划赶不上变化，脸疼。

2019 年的梦想是去迪拜完成 15000 米跳伞，去沙漠坐骆驼。2019 年旅行计划主要就是迪拜，欧洲和美国，去德法意瑞，看看欧洲列强们如今过的还好么；去美国体验体验常青藤学校浓厚的学术氛围；去迪拜看看白袍们有多么奢侈的生活，捡一捡丢在马路边的“垃圾”，兰博基尼，住一住七星级酒店。

一切源于年初的时候，我在世界地图上选择了 9 处比较浪漫的地方作为今年女友 18 岁的生日礼物，打算 2-3 年内完成这 9 处的打卡。我是一个不懂浪漫的穷人家的孩子，送不起房子，送不起车，只能送回忆了。既然作为 18 岁生日的礼物，那主题就叫 “勇敢者的游戏”吧。

最终定下来是去迪拜，完成棕榈岛 15000 米跳伞和沙迦沙漠深处冲沙。强烈推荐跳伞项目，真的太好玩了。笔者跳伞的长视频发在 @ halfrost 抖音号上了，欢迎读者去观看。关于迪拜的酒店，强烈推荐全世界唯一的七星级酒店帆船酒店 Burj Al Arab，和六星级酒店亚特兰蒂斯 Atlantis The Palm，当然还推荐全世界唯一的八星级酒店，只不过不在迪拜，在阿布扎比，阿布扎比皇宫酒店(Emirates Palace)。下面 2 张图分别是亚特兰蒂斯和帆船酒店。

可能有读者会问上面两个图的拍摄角度为什么这么特殊。笔者是在直升机上拍的。在亚特兰蒂斯酒店旁边有一个直升机场，可以买一张票，笔者买的是 25 分钟的票，直升机会带你飞到市中心转一圈再飞回来。沿途会经过帆船酒店，黄金相框，世界岛，哈利法塔，所有经典景观都会让你从上空看一遍。最后围绕棕榈岛半圈，回到亚特兰蒂斯旁边的停机坪。

迪拜的亚特兰蒂斯酒店可以去玩它的水上公园。是免费的。它的水上公园真的非常好玩。住在亚特兰蒂斯的话，旁边也没有什么小店可以吃饭，吃喝都在酒店里面了。（住亚特兰蒂斯的人还考虑消费么？花钱就是快乐）亚特兰蒂斯里面很多自助餐厅，至于价位嘛，消费上不封顶。带多少钱都能在这里挥霍完。笔者非常“省吃俭用”的在这里住了几天。

帆船酒店就不用说了。纯金的马桶，每晚 13-20W 的高级套房，爱马仕香皂。一切都是奢华的顶配。出门可以预约劳斯莱斯幻影。帆船酒店的房间里面的私人管家服务周到，只要你有钱，不太过分的要求都能尽量满足。比如你想把 F1 赛车运到帆船酒店的顶楼直升机停机坪，玩漂移，都是可以的。（并非玩笑，是真的可以）

比帆船酒店还要再高一星级的皇宫酒店，笔者没有体验，只去吃了自助餐。皇宫酒店在阿布扎比，它的内部不一定有帆船酒店奢华，但是它的社会地位比帆船酒店高。它本来是专门招待各国顶级领导人的。当领导人来访问的时候，是不开放给普通游客的。如果你不住在皇宫酒店，还是可以单独约这里的自助餐的。自助餐全部都是米其林星级厨师纯手工制作。所以中午吃一个自助餐，吃 2-3 小时是很正常的。

迪拜其他的打卡地有，民俗村，运河，博物馆，黄金相框等等。

上图是迪拜的民俗村，是迪拜最老的当地人住的地方。房间上面那些横着的柱子和风洞，是用来调节房间温度的，是一个天然空调。

上图是黄金相框，个人觉得还是在直升机上看这个相框更有感觉。站在它的面前，你能看到相框里面的景色就只有天空了。

迪拜的朱美拉清真寺（Jumeirah Mosque）是不得不提的打卡地。笔者是五一去玩的，正好遇到斋戒期。去清真寺一定要符合宗教的服装要求，男女服装都有要求。男士和女士需穿着保守、宽松、不透明的衣服，最好选择长袖 (腕长)，长裙子 (脚踝长度) 或裤子，进入清真寺前，有一个商店可以租这些衣服。

清真寺建筑群的内外墙壁用来自希腊和马其顿的汉白玉包裹而成，内部装饰金碧辉煌，错综复杂的雕刻和壁画令人赞叹！雪白的大理石圆顶及墙面，在阳光下隐隐发亮，清真寺前湛蓝的一池清水，不禁被这片圣洁之地所吸引。黄金柱头，简洁的柱子底座，大有彩色图腾花纹的柱身，加上具有标志性的拱形洞口，整个色调的把握，把阿拉伯文化淋漓尽致演绎到建筑之中，整个清真寺就是一个奢华艺术品！在这里，你还可以亲眼目睹世界上最大的大理石马赛克装饰和纯手工地毯，给你前所未有的视觉震撼。上两张图只有你亲临现场，才能被彻底的震撼到。相关的视频也记录在笔者的抖音里面了。

上图是在直升机上拍的棕榈岛，这个就不用说了，世人都知道。还有一个没有完工的世界岛，是另外一个人造岛。笔者在直升机上没有拍到，当飞行员解说到世界岛了，我看了半天才意识到哪里是世界岛，最终错过拍照了。

最后一个打卡地就是世界最高的塔，哈利法塔。可以去 128 层的观光层往下看风景，有迪拜城市的全貌。笔者建议下午 3-4 点去，这样可以一直呆到 6 点日落的时候，和心爱的人在世界之巅一起看日落。值得一提的是哈利法塔的电梯也是世界最快的。笔者拍摄了它上升的速度，非常震撼，视频都在抖音号上。

其他的活动就是消费活动咯。The Dubai Mall 肯定是必去的。它由 10 到 15 个 Mall 中 Mall 组成，一共将有大约 1200 个商店，有 16000 个停车位。此外，它还将有世界最大的水族馆，最大的黄金市场，奥运比赛规模的冰场，6 层楼高的巨幅屏幕影院，探险公园，沙漠喷泉等等。

迪拜购物中心单独占地 500 万平方英尺（约 46 万 5000 平方米），相当于 56 个足球场的面积，连同其所有辅助设施、附属建筑在内、总共占地 900 万平方英尺（约 83 万 6000 平方米），这些数据都刷新了世界纪录，超过了加拿大埃德蒙多市的得西埃商业中心和美国明尼苏达州布卢明顿市的美国购物中心。也许你对这些数字没有感受，那笔者这样描述吧。从早上商场开门，一直逛到晚上商场关门，逛整整一天，只能逛完其中一层，连续逛一周才能把整个购物中心逛完。

迪拜的茶余饭后的娱乐活动非常匮乏，没有中国的棋牌，麻将，广场舞等等活动。女性唯一的娱乐活动就是逛商场，消费。所以 The Dubai Mall 拥有全世界最新款的 LV 包包，拥有全世界最新款和最贵的奢侈品。只有这样才能满足迪拜女性饭后的娱乐需求。（这段不是开玩笑，说的是真实的）所以当你在购物中心看见一个白袍领着他的老婆们一顿买买买，一口气买 16 个 LV，4 辆劳斯莱斯的时候，别认为是人家败家，其实人家是在娱乐呢。（4 个老婆，每个老婆一辆劳斯莱斯，一年四季，每个季节都要一个 LV 包包。所以需要 16 个 LV，4 辆劳斯莱斯）在迪拜允许一夫多妻，但是必须对每个老婆都公平对待，你的爱要平均分给每个人。

好了，今年的旅行就说到这里了，大多数迪拜的旅行视频都在 @halfrost 的抖音号上。读者有空感兴趣的话可以去看看。希望明年笔者的托福和 GRE 可以考到满意的分数。考完了想去南美或者冰岛转一转。（立 flag 要打脸）

最后

一位大佬朋友圈写道：看程序员是否勤奋就看他的英语好不好，智商高不高就看他算法好不好。这句话我当时看到了很触动，默默的记在了心底。2019 年一年我就只做了 2 件事情，刷算法，学英语。我现在还不敢说我是优秀的程序员，但是我至少努力过。不辜负时光，无愧于自己。以上就是你们想要的答案，这就是我 2019 年的年终总结，里面揭秘了 98% 的人都不了解的事情。很多猎头迷惑的内容也都在里面了。感谢周围亲戚朋友的这 2 年的关心。这篇总结不是原子弹💥别太惊讶。

回过头来看，好像也没有做出什么牛逼的事情。就是下了一步小棋，布了一个小局。也不是什么传奇经历，无非是奋斗路上立 flag，达到 flag，再立 flag 的一滴滴汗水罢了。过去 2 年，我一直隐身于网络，也没人知道我立的这些 flag。大家的年终总结都是对自己今年 flag 完成度的复盘，而我没有勇气去直面被人嘲笑的梦想，如果最终梦想没有实现，flag 无非就是人们饭后的谈资，打肿我脸的笑柄。这也是近 2 年都没有看到我的年终总结公开出来的原因。最终的梦想总算完成了，也是时候可以把年终总结公开出来了，给关心我的读者和朋友一些交代了。（本文写于 2019 年年末，最后这一段修改于 2021年 3 月）

最后一些“只言片语”的感受分享一下作为年终总结的结尾吧。

1. 不要向任何人诉苦。因为 20% 的人不关心，剩下 80% 的人听了会开心。
2. 狮子从来不会关心一只羊的看法，不要在意别人怎么看你，你应该努力强大成为一只狮子。
3. 要学会拒绝，没有人会感激你的善良，很多人只会得寸进尺。
4. 不需要解释的不要解释，从你张嘴的那一刻你就已经输了。
5. 当有人侮辱你的时候你要记住，狮子不会因为狗叫而回头。

好了，2019 年的【星霜荏苒】就到这里了。如有任何异议或者想讨论的地方，欢迎和我交流。

2019 年 5 月 5 日，于迪拜 Dubai

---

GitHub Repo：Halfrost-Field

Follow: halfrost · GitHub

Source: https://halfrost.com/halfrost_2019/

作为 Go 并发原语的第一篇文章，一定绕不开 Go 的并发哲学。从 Tony Hoare 写的 Communicating Sequential Processes 这篇文章说起，这篇经典论文算是 Go 语言并发原语的根基。

一. What is CSP

CSP 的全程是 Communicating Sequential Processes，直译，通信顺序进程。这一概念起源自 1978 年 ACM 期刊中 Charles Antony Richard Hoare 写的经典同名论文。感兴趣的读者可以看 Reference 中的第一个链接看原文。在这篇文章中，Hoare 在文中用 CSP 来描述通信顺序进程能力，姑且认为这是一个虚构的编程语言。该语言描述了并发过程之间的交互作用。从历史上看，软件的进步主要依靠硬件的改进，这些改进可以使 CPU 更快，内存更大。Hoare 认识到，想通过硬件提高使得代码运行速度快 10 倍，需要付出 10 倍以上的机器资源。这并没有从根本改善问题。

1. 术语和一些例子

尽管并发相对于传统的顺序编程具有许多优势，但由于其会出错的性质，它未能获得广泛的欢迎。Hoare 借助 CSP 引入了一种精确的理论，可以在数学上保证程序摆脱并发的常见问题。Hoare 在他的 Learning CSP（这是计算机科学中引用第三多的神书！）一书中，使用“进程微积分”来表明可以处理死锁和不确定性，就像它们是普通进程中的最终事件一样。进程微积分是一种对并发系统进行数学化建模的方式，并且提供了代数法则来进行这些系统的变换来分析它们的不同属性，并发和效率。

为了防止数据被多线程破坏，Hoare 提出了临界区的概念。进程进入临界区后可以获得对共享数据的访问。在进入临界区之前，所有其他的进程必须验证和更新这一共享变量的值。退出临界区时，进程必须再次验证所有进程具有相同的值。

保持数据完整性的另一种技术是通过使用互斥信号量或互斥量。互斥锁是信号量的特定子类，它仅允许一个进程一次访问该变量。信号量是一个受限制的访问变量，它是防止并发中竞争的经典解决方案。其他尝试访问该互斥锁的进程将被阻止，并且必须等待直到当前进程释放该互斥锁。释放互斥锁后，只有一个等待的进程可以访问该变量，所有其他进程继续等待。

1970年代初期，Hoare 基于互斥量的概念开发了一种称为监视器的概念。根据 IBM 编写的 Java 编程语言 CSP 教程：

“A monitor is a body of code whose access is guarded by a mutex. Any process wishing to execute this code must acquire the associated mutex at the top of the code block and release it at the bottom. Because only one thread can own a mutex at a given time, this effectively ensures that only the owing thread can execute a monitor block of code.”

monitor 可以帮助防止数据被破坏和线程死锁。在 CSP 论文中为了说明清楚进程之间的通信，Hoare 利用 ？和 ！号代表了输入和输出。！代表发送输入到一个进程，？号代表读取一个进程的输出。每个指令需要指定具体是一个输出变量(从一个进程中读取一个变量的情况)，还是目的地(将输入发送到一个进程的情况)。一个进程的输出应该直接流向另一个进程的输入。

上图是从 CSP 文章中截图的一些例子，Hoare 简单的举了下面这个例子：

[c:character; west?c ~ east!c] 

上述代码的意思是读取 west 输出的所有字符，然后把它们一个个的输出到 east 中。这个过程不断的重复，直到 west 终止。从描述上看，这一特性完完全全是 channel 的雏形。

2. 哲学家问题

文章的最后，回到了经典的哲学家问题。

在哲学家问题中，Hoare 将 philosopher 的行为描述如下：

PHIL = *[... during ith lifetime ... --->,
THINK;
room!enter( );
fork(0!pickup( ); fork((/+ 1) rood 5)!pickup( );
EAT;
fork(i)!putdown( ); fork((/+ 1) mod 5)!putdown( );
room!exit( )
]

每个叉子由坐在两边的哲学家使用或者放下：

FORK =
*[phil(0?pickup( )--* phil(0?putdown( )
0phil((i - 1)rood 5)?pickup( ) --* phil((/- l) raod 5)?putdown( )
]

整个哲学家在房间中的行为可以描述为：

ROOM = occupancy:integer; occupancy .--- 0;
,[(i:0..4)phil(0?enter ( ) --* occupancy .--- occupancy + l
11(i:0..4)phil(0?exit ( ) --~ occupancy .--- occupancy - l
] 

决定如何向等待的进程分配资源的任务称为调度。Hoare 将调度分为两个事件：

• processes 请求资源
• 将资源分配给 processes

那么这个哲学家问题可以转换成 PHIL 和 FORK 这两个组件并发的过程：

[room::ROOM I [fork( i:0..4)::FORK I Iphil( i:0..4)::PHIL]. 

从请求到授予资源的时间就是等待时间。在 CSP 中，有几种技术可以防止无限的等待时间。

• 限制资源使用并提高资源可用性。
• 先进先出(FIFO)将资源分配给等待时间最长的进程。
• 面包店算法Carnegie Melon. Bakery Algorithm

3. 缺陷

在确定性程序中，如果环境恒定，结果将是相同的。 由于并发基于非确定性，因此环境不会影响程序。给定所选的路径，程序则可以运行几次并收到不同的结果。为了确保并发程序的准确性，程序员必须能够在整体水平上考虑其程序的执行。

但是，尽管 Hoare 引入了正式的方法，但仍然缺少任何验证正确程序的证明方法。CSP 只能发现已知问题，而不能发现未知问题。虽然基于 CSP 的商业应用程序（例如ConAn）可以检测到错误的存在，但是不能检测没有错误的情况，(无法验证正确性)。尽管 CSP 为我们提供了编写可以避免常见并发错误的程序的工具，但是正确程序的证明仍然是 CSP 中尚未解决的领域。

4. 未来

CSP 在生物学和化学领域具有巨大的潜力，可以对自然界中的复杂系统进行建模。 由于该行业面临许多现存的逻辑问题，因此尚未在行业中广泛使用。在关于 CSP 开发 25 周年的会议上，Hoare 指出，尽管有许多由 Microsoft 资助的研究项目，但比尔·盖茨（Bill Gates）忽略了 Microsoft 何时能够将 CSP 的研究成果商业化的问题。

Hoare 提醒他的听众，动态过程领域仍然需要更多的研究。当前，计算机科学界陷入了顺序思维的范式。随着 Hoare 建立正式的并发方法的基础，科学界已做好准备成为并行编程的下一个革命。

5. Go 并发哲学

在 Go 语言发布之前，很少有语言从底层为并发原语提供支持。大多数语言还是支持共享和内存访问同步到 CSP 的消息传递方法。Go 语言算是最早将 CSP 原则纳入其核心的语言之一。内存访问同步的方式并不是不好，只是在高并发的场景下有时候难以正确的使用，特别是在超大型，巨型的程序中。基于此，并发能力被认为是 Go 语言天生优势之一。追其根本，还是因为 Go 基于 CSP 创造出来的一系列易读，方便编写的并发原语。

Go 语言除了 CSP 并发原语以外，还支持通过内存访问同步。sync 与其他包中的结构体与方法可以让开发者创建 WaitGroup，互斥锁和读写锁，cond，once，sync.Pool。在 Go 语言的官方 FAQ 中，描述了如何选择这些并发原语：

为了尊重 mutex，sync 包实现了 mutex，但是我们希望 Go 语言的编程风格将会激励人们尝试更高等级的技巧。尤其是考虑构建你的程序，以便一次只有一个 goroutine 负责某个特定的数据。

不要通过共享内存进行通信。建议，通过通信来共享内存。（Do not communicate by sharing memory; instead, share memory by communicating）这是 Go 语言并发的哲学座右铭。相对于使用 sync.Mutex 这样的并发原语。虽然大多数锁的问题可以通过 channel 或者传统的锁两种方式之一解决，但是 Go 语言核心团队更加推荐使用 CSP 的方式。

关于如何选择并发原语的问题，本文作为第一篇文章必然需要解释清楚。Go 中的并发原语主要分为 2 大类，一个是 sync 包里面的，另一个是 channel。sync 包里面主要是 WaitGroup，互斥锁和读写锁，cond，once，sync.Pool 这一类。在 2 种情况下推荐使用 sync 包：

• 对性能要求极高的临界区
• 保护某个结构内部状态和完整性

关于保护某个结构内部的状态和完整性。例如 Go 源码中如下代码：

var sum struct {
	sync.Mutex
	i int
}

//export Add
func Add(x int) {
	defer func() {
		recover()
	}()
	sum.Lock()
	sum.i += x
	sum.Unlock()
	var p *int
	*p = 2
}

sum 这个结构体不想将内部的变量暴露在结构体之外，所以使用 sync.Mutex 来保护线程安全。

相对于 sync 包，channel 也有 2 种情况：

• 输出数据给其他使用方
• 组合多个逻辑

输出数据给其他使用方的目的是转移数据的使用权。并发安全的实质是保证同时只有一个并发上下文拥有数据的所有权。channel 可以很方便的将数据所有权转给其他使用方。另一个优势是组合型。如果使用 sync 里面的锁，想实现组合多个逻辑并且保证并发安全，是比较困难的。但是使用 channel + select 实现组合逻辑实在太方便了。以上就是 CSP 的基本概念和何时选择 channel 的时机。下一章从 channel 基本数据结构开始详细分析 channel 底层源码实现。

以下代码基于 Go 1.16

二. 基本数据结构

channel 的底层源码和相关实现在 src/runtime/chan.go 中。

type hchan struct {
	qcount   uint           // 队列中所有数据总数
	dataqsiz uint           // 环形队列的 size
	buf      unsafe.Pointer // 指向 dataqsiz 长度的数组
	elemsize uint16         // 元素大小
	closed   uint32
	elemtype *_type         // 元素类型
	sendx    uint           // 已发送的元素在环形队列中的位置
	recvx    uint           // 已接收的元素在环形队列中的位置
	recvq    waitq          // 接收者的等待队列
	sendq    waitq          // 发送者的等待队列

	lock mutex
}

lock 锁保护 hchan 中的所有字段，以及此通道上被阻塞的 sudogs 中的多个字段。持有 lock 的时候，禁止更改另一个 G 的状态（特别是不要使 G 状态变成ready），因为这会因为堆栈 shrinking 而发生死锁。

recvq 和 sendq 是等待队列，waitq 是一个双向链表：

type waitq struct {
	first *sudog
	last  *sudog
}

channel 最核心的数据结构是 sudog。sudog 代表了一个在等待队列中的 g。sudog 是 Go 中非常重要的数据结构，因为 g 与同步对象关系是多对多的。一个 g 可以出现在许多等待队列上，因此一个 g 可能有很多sudog。并且多个 g 可能正在等待同一个同步对象，因此一个对象可能有许多 sudog。sudog 是从特殊池中分配出来的。使用 acquireSudog 和 releaseSudog 分配和释放它们。

type sudog struct {

	g *g

	next *sudog
	prev *sudog
	elem unsafe.Pointer // 指向数据 (可能指向栈)

	acquiretime int64
	releasetime int64
	ticket      uint32

	isSelect bool
	success bool

	parent   *sudog     // semaRoot 二叉树
	waitlink *sudog     // g.waiting 列表或者 semaRoot
	waittail *sudog     // semaRoot
	c        *hchan     // channel
}

sudog 中所有字段都受 hchan.lock 保护。acquiretime、releasetime、ticket 这三个字段永远不会被同时访问。对 channel 来说，waitlink 只由 g 使用。对 semaphores 来说，只有在持有 semaRoot 锁的时候才能访问这三个字段。isSelect 表示 g 是否被选择，g.selectDone 必须进行 CAS 才能在被唤醒的竞争中胜出。success 表示 channel c 上的通信是否成功。如果 goroutine 在 channel c 上传了一个值而被唤醒，则为 true；如果因为 c 关闭而被唤醒，则为 false。

三. 创建 Channel

创建 channel 常见代码：

ch := make(chan int)

编译器编译上述代码，在检查 ir 节点时，根据节点 op 不同类型，进行不同的检查，如下源码：

func walkExpr1(n ir.Node, init *ir.Nodes) ir.Node {
	switch n.Op() {
	default:
		ir.Dump("walk", n)
		base.Fatalf("walkExpr: switch 1 unknown op %+v", n.Op())
		panic("unreachable")

	case ir.OMAKECHAN:
		n := n.(*ir.MakeExpr)
		return walkMakeChan(n, init)

	......
}

编译器会检查每一种类型，walkExpr1() 的实现就是一个 switch-case，函数末尾没有 return，因为每一个 case 都会 return 或者返回 panic。这样做是为了与存在类型断言的情况中返回的内容做区分。walk 具体处理 OMAKECHAN 类型节点的逻辑如下：

func walkMakeChan(n *ir.MakeExpr, init *ir.Nodes) ir.Node {
	size := n.Len
	fnname := "makechan64"
	argtype := types.Types[types.TINT64]

	if size.Type().IsKind(types.TIDEAL) || size.Type().Size() <= types.Types[types.TUINT].Size() {
		fnname = "makechan"
		argtype = types.Types[types.TINT]
	}

	return mkcall1(chanfn(fnname, 1, n.Type()), n.Type(), init, reflectdata.TypePtr(n.Type()), typecheck.Conv(size, argtype))
}

上述代码默认调用 makechan64() 函数。类型检查时如果 TIDEAL 大小在 int 范围内。将 TUINT 或 TUINTPTR 转换为 TINT 时出现大小溢出的情况，将在运行时在 makechan 中进行检查。如果在 make 函数中传入的 channel size 大小在 int 范围内，推荐使用 makechan()。因为 makechan() 在 32 位的平台上更快，用的内存更少。

makechan64() 和 makechan() 函数方法原型如下：

func makechan64(chanType *byte, size int64) (hchan chan any)
func makechan(chanType *byte, size int) (hchan chan any)

makechan64() 方法只是判断一下传入的入参 size 是否还在 int 范围之内：

func makechan64(t *chantype, size int64) *hchan {
	if int64(int(size)) != size {
		panic(plainError("makechan: size out of range"))
	}

	return makechan(t, int(size))
}

创建 channel 的主要实现在 makechan() 函数中：

func makechan(t *chantype, size int) *hchan {
	elem := t.elem

	// 编译器检查数据项大小不能超过 64KB
	if elem.size >= 1<<16 {
		throw("makechan: invalid channel element type")
	}
	// 检查对齐是否正确
	if hchanSize%maxAlign != 0 || elem.align > maxAlign {
		throw("makechan: bad alignment")
	}
    // 缓冲区大小检查，判断是否溢出
	mem, overflow := math.MulUintptr(elem.size, uintptr(size))
	if overflow || mem > maxAlloc-hchanSize || size < 0 {
		panic(plainError("makechan: size out of range"))
	}

	var c *hchan
	switch {
	case mem == 0:
		// 队列或者元素大小为 zero 时
		c = (*hchan)(mallocgc(hchanSize, nil, true))
		// Race 竞争检查利用这个地址来进行同步操作
		c.buf = c.raceaddr()
	case elem.ptrdata == 0:
		// 元素不包含指针时。一次分配 hchan 和 buf 的内存。
		c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
		c.buf = add(unsafe.Pointer(c), hchanSize)
	default:
		// 元素包含指针时
		c = new(hchan)
		c.buf = mallocgc(mem, elem, true)
	}

    // 设置属性
	c.elemsize = uint16(elem.size)
	c.elemtype = elem
	c.dataqsiz = uint(size)
	lockInit(&c.lock, lockRankHchan)

	if debugChan {
		print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
	}
	return c
}

上面这段 makechan() 代码主要目的是生成 *hchan 对象。重点关注 switch-case 中的 3 种情况：

• 当队列或者元素大小为 0 时，调用 mallocgc() 在堆上为 channel 开辟一段大小为 hchanSize 的内存空间。
• 当元素类型不是指针类型时，调用 mallocgc() 在堆上开辟为 channel 和底层 buf 缓冲区数组开辟一段大小为 hchanSize + mem 连续的内存空间。
• 默认情况元素类型中有指针类型，调用 mallocgc() 在堆上分别为 channel 和 buf 缓冲区分配内存。

完成第一步的内存分配之后，再就是 hchan 数据结构其他字段的初始化和 lock 的初始化。值得说明的一点是，当存储在 buf 中的元素不包含指针时，Hchan 中也不包含 GC 关心的指针。buf 指向一段相同元素类型的内存，elemtype 固定不变。SudoG 是从它们自己的线程中引用的，因此垃圾回收的时候无法回收它们。受到垃圾回收器的限制，指针类型的缓冲 buf 需要单独分配内存。官方在这里加了一个 TODO，垃圾回收的时候这段代码逻辑需要重新考虑。

就是因为 channel 的创建全部调用的 mallocgc()，在堆上开辟的内存空间，channel 本身会被 GC 自动回收。有了这一性质，所以才有了下文关闭 channel 中优雅关闭的方法。

四. 发送数据

向 channel 中发送数据常见代码：

ch <- 1

编译器编译上述代码，在检查 ir 节点时，根据节点 op 不同类型，进行不同的检查，如下源码：

func walkExpr1(n ir.Node, init *ir.Nodes) ir.Node {
	switch n.Op() {
	default:
		ir.Dump("walk", n)
		base.Fatalf("walkExpr: switch 1 unknown op %+v", n.Op())
		panic("unreachable")

	case ir.OSEND:
		n := n.(*ir.SendStmt)
		return walkSend(n, init)

	......
}

walkExpr1() 函数在创建 channel 提到了，这里不再赘述。操作类型是 OSEND，对应调用 walkSend() 函数：

func walkSend(n *ir.SendStmt, init *ir.Nodes) ir.Node {
	n1 := n.Value
	n1 = typecheck.AssignConv(n1, n.Chan.Type().Elem(), "chan send")
	n1 = walkExpr(n1, init)
	n1 = typecheck.NodAddr(n1)
	return mkcall1(chanfn("chansend1", 2, n.Chan.Type()), nil, init, n.Chan, n1)
}

// entry point for c <- x from compiled code
//go:nosplit
func chansend1(c *hchan, elem unsafe.Pointer) {
	chansend(c, elem, true, getcallerpc())
}

walkSend() 函数中主要逻辑调用了 chansend1()，而 chansend1() 只是 chansend() 的“外壳”。所以 channel 发送数据的核心实现在 chansend() 中。根据 channel 的阻塞和唤醒，又可以分为 2 部分逻辑代码。接下来笔者讲 chansend() 代码拆成 4 部分详细分析。

1. 异常检查

chansend() 函数一开始先进行异常检查：

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
    // 判断 channel 是否为 nil
	if c == nil {
		if !block {
			return false
		}
		gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
		throw("unreachable")
	}

	if debugChan {
		print("chansend: chan=", c, "\n")
	}

	if raceenabled {
		racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
	}

	// 简易快速的检查
	if !block && c.closed == 0 && full(c) {
		return false
	}
......
}

chansend() 一上来对 channel 进行检查，如果被 GC 回收了会变为 nil。朝一个为 nil 的 channel 发送数据会发生阻塞。gopark 会引发以 waitReasonChanSendNilChan 为原因的休眠，并抛出 unreachable 的 fatal error。当 channel 不为 nil，再开始检查在没有获取锁的情况下会导致发送失败的非阻塞操作。

当 channel 不为 nil，并且 channel 没有 close 时，还需要检查此时 channel 是否做好发送的准备，即判断 full(c)

func full(c *hchan) bool {
	if c.dataqsiz == 0 {
		// 假设指针读取是近似原子性的
		return c.recvq.first == nil
	}
	// 假设读取 uint 是近似原子性的
	return c.qcount == c.dataqsiz
}

full() 方法作用是判断在 channel 上发送是否会阻塞（即通道已满）。它读取单个字节大小的可变状态(recvq.first 和 qcount)，尽管答案可能在一瞬间是 true，但在调用函数收到返回值时，正确的结果可能发生了更改。值得注意的是 dataqsiz 字段，它在创建完 channel 以后是不可变的，因此它可以安全的在任意时刻读取。

回到 chansend() 异常检查中。一个已经 close 的 channel 是不可能从“准备发送”的状态变为“未准备好发送”的状态。所以在检查完 channel 是否 close 以后，就算 channel close 了，也不影响此处检查的结果。可能有读者疑惑，“能不能把检查顺序倒一倒？先检查是否 full()，再检查是否 close？”。这样倒过来确实能保证检查 full() 的时候，channel 没有 close。但是这种做法也没有实质性的改变。channel 依旧可以在检查完 close 以后再关闭。其实我们依赖的是 chanrecv() 和 closechan() 这两个方法在锁释放后，它们更新这个线程 c.close 和 full() 的结果视图。

2. 同步发送

channel 异常状态检查以后，接下来的代码是发送的逻辑。

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
......

	lock(&c.lock)

	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("send on closed channel"))
	}

	if sg := c.recvq.dequeue(); sg != nil {
		send(c, sg, ep, func() { unlock(&c.lock) }, 3)
		return true
	}

......

}

在发送之前，先上锁，保证线程安全。并再一次检查 channel 是否关闭。如果关闭则抛出 panic。加锁成功并且 channel 未关闭，开始发送。如果有正在阻塞等待的接收方，通过 dequeue() 取出头部第一个非空的 sudog，调用 send() 函数：

func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
	if sg.elem != nil {
		sendDirect(c.elemtype, sg, ep)
		sg.elem = nil
	}
	gp := sg.g
	unlockf()
	gp.param = unsafe.Pointer(sg)
	sg.success = true
	if sg.releasetime != 0 {
		sg.releasetime = cputicks()
	}
	goready(gp, skip+1)
}

send() 函数主要完成了 2 件事：

• 1. 调用 sendDirect() 函数将数据拷贝到了接收变量的内存地址上
• 2. 调用 goready() 将等待接收的阻塞 goroutine 的状态从 Gwaiting 或者 Gscanwaiting 改变成 Grunnable。下一轮调度时会唤醒这个接收的 goroutine。

这里重点说说 goready() 的实现。理解了它的源码，就能明白为什么往 channel 中发送数据并非立即可以从接收方获取到。

func goready(gp *g, traceskip int) {
	systemstack(func() {
		ready(gp, traceskip, true)
	})
}

func ready(gp *g, traceskip int, next bool) {
......

	casgstatus(gp, _Gwaiting, _Grunnable)
	runqput(_g_.m.p.ptr(), gp, next)
	wakep()
	releasem(mp)
}

在 runqput() 函数的作用是把 g 绑定到本地可运行的队列中。此处 next 传入的是 true，将 g 插入到 runnext 插槽中，等待下次调度便立即运行。因为这一点导致了虽然 goroutine 保证了线程安全，但是在读取数据方面比数组慢了几百纳秒。

Read	Channel	Slice
Time	x * 100 * nanosecond	0
Thread safe	Yes	No

所以在写测试用例的某些时候，需要考虑到这个微弱的延迟，可以适当加 sleep()。再比如刷 LeetCode 题目的时候，并非无脑使用 goroutine 就能带来 runtime 的提升，例如 509. Fibonacci Number，感兴趣的同学可以用 goroutine 来写一写这道题，笔者这里实现了goroutine 解法，性能方面完全不如数组的解法。

3. 异步发送

如果初始化 channel 时创建的带缓冲区的异步 Channel，当接收者队列为空时，这是会进入到异步发送逻辑：

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
......

	if c.qcount < c.dataqsiz {
		qp := chanbuf(c, c.sendx)
		if raceenabled {
			racenotify(c, c.sendx, nil)
		}
		typedmemmove(c.elemtype, qp, ep)
		c.sendx++
		if c.sendx == c.dataqsiz {
			c.sendx = 0
		}
		c.qcount++
		unlock(&c.lock)
		return true
	}
	
......
}

如果 qcount 还没有满，则调用 chanbuf() 获取 sendx 索引的元素指针值。调用 typedmemmove() 方法将发送的值拷贝到缓冲区 buf 中。拷贝完成，需要维护 sendx 索引下标值和 qcount 个数。这里将 buf 缓冲区设计成环形的，索引值如果到了队尾，下一个位置重新回到队头。

至此，两种直接发送的逻辑分析完了，接下来是发送时 channel 阻塞的情况。

4. 阻塞发送

当 channel 处于打开状态，但是没有接收者，并且没有 buf 缓冲队列或者 buf 队列已满，这时 channel 会进入阻塞发送。

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
......

	if !block {
		unlock(&c.lock)
		return false
	}
	
	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	mysg.elem = ep
	mysg.waitlink = nil
	mysg.g = gp
	mysg.isSelect = false
	mysg.c = c
	gp.waiting = mysg
	gp.param = nil
	c.sendq.enqueue(mysg)
	atomic.Store8(&gp.parkingOnChan, 1)
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
	KeepAlive(ep)
......
}

• 调用 getg() 方法获取当前 goroutine 的指针，用于绑定给一个 sudog。
• 调用 acquireSudog() 方法获取一个 sudog，可能是新建的 sudog，也有可能是从缓存中获取的。设置好 sudog 要发送的数据和状态。比如发送的 Channel、是否在 select 中和待发送数据的内存地址等等。
• 调用 c.sendq.enqueue 方法将配置好的 sudog 加入待发送的等待队列。
• 设置原子信号。当栈要 shrink 收缩时，这个标记代表当前 goroutine 还 parking 停在某个 channel 中。在 g 状态变更与设置 activeStackChans 状态这两个时间点之间的时间窗口进行栈 shrink 收缩是不安全的，所以需要设置这个原子信号。
• 调用 gopark 方法挂起当前 goroutine，状态为 waitReasonChanSend，阻塞等待 channel。
• 最后，KeepAlive() 确保发送的值保持活动状态，直到接收者将其复制出来。 sudog 具有指向堆栈对象的指针，但 sudog 不能被当做堆栈跟踪器的 root。发送的数值是分配在堆上，这样可以避免被 GC 回收。

这里提一下 sudog 的二级缓存复用体系。在 acquireSudog() 方法中：

func acquireSudog() *sudog {
	mp := acquirem()
	pp := mp.p.ptr()
	// 如果本地缓存为空
	if len(pp.sudogcache) == 0 {
		lock(&sched.sudoglock)
		// 首先尝试将全局中央缓存存一部分到本地
		for len(pp.sudogcache) < cap(pp.sudogcache)/2 && sched.sudogcache != nil {
			s := sched.sudogcache
			sched.sudogcache = s.next
			s.next = nil
			pp.sudogcache = append(pp.sudogcache, s)
		}
		unlock(&sched.sudoglock)
		// 如果全局中央缓存是空的，则 allocate 一个新的
		if len(pp.sudogcache) == 0 {
			pp.sudogcache = append(pp.sudogcache, new(sudog))
		}
	}
	// 从尾部提取，并调整本地缓存
	n := len(pp.sudogcache)
	s := pp.sudogcache[n-1]
	pp.sudogcache[n-1] = nil
	pp.sudogcache = pp.sudogcache[:n-1]
	if s.elem != nil {
		throw("acquireSudog: found s.elem != nil in cache")
	}
	releasem(mp)
	return s
}

上述代码涉及到 2 个新的重要的结构体，由于这 2 个结构体特别复杂，暂时此处只展示和 acquireSudog() 有关的部分：

type p struct {
......
	sudogcache []*sudog
	sudogbuf   [128]*sudog
......
}

type schedt struct {
......
	sudoglock  mutex
	sudogcache *sudog
......
}

sched.sudogcache 是全局中央缓存，可以认为它是“一级缓存”，它会在 GC 垃圾回收执行 clearpools 被清理。p.sudogcache 可以认为它是“二级缓存”，是本地缓存不会被 GC 清理掉。

chansend 最后的代码逻辑是当 goroutine 唤醒以后，解除阻塞的状态：

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
......

	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	gp.activeStackChans = false
	closed := !mysg.success
	gp.param = nil
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	mysg.c = nil
	releaseSudog(mysg)
	if closed {
		if c.closed == 0 {
			throw("chansend: spurious wakeup")
		}
		panic(plainError("send on closed channel"))
	}
	return true
}

sudog 算是对 g 的一种封装，里面包含了 g，要发送的数据以及相关的状态。goroutine 被唤醒后会完成 channel 的阻塞数据发送。发送完最后进行基本的参数检查，解除 channel 的绑定并释放 sudog。

func releaseSudog(s *sudog) {
	if s.elem != nil {
		throw("runtime: sudog with non-nil elem")
	}
	if s.isSelect {
		throw("runtime: sudog with non-false isSelect")
	}
	if s.next != nil {
		throw("runtime: sudog with non-nil next")
	}
	if s.prev != nil {
		throw("runtime: sudog with non-nil prev")
	}
	if s.waitlink != nil {
		throw("runtime: sudog with non-nil waitlink")
	}
	if s.c != nil {
		throw("runtime: sudog with non-nil c")
	}
	gp := getg()
	if gp.param != nil {
		throw("runtime: releaseSudog with non-nil gp.param")
	}
	// 防止 rescheduling 到了其他的 P
	mp := acquirem() 
	pp := mp.p.ptr()
	// 如果本地缓存已满
	if len(pp.sudogcache) == cap(pp.sudogcache) {
		// 转移一半本地缓存到全局中央缓存中
		var first, last *sudog
		for len(pp.sudogcache) > cap(pp.sudogcache)/2 {
			n := len(pp.sudogcache)
			p := pp.sudogcache[n-1]
			pp.sudogcache[n-1] = nil
			pp.sudogcache = pp.sudogcache[:n-1]
			if first == nil {
				first = p
			} else {
				last.next = p
			}
			last = p
		}
		lock(&sched.sudoglock)
		// 将提取的链表挂载到全局中央缓存中
		last.next = sched.sudogcache
		sched.sudogcache = first
		unlock(&sched.sudoglock)
	}
	pp.sudogcache = append(pp.sudogcache, s)
	releasem(mp)
}

releaseSudog() 虽然释放了 sudog 的内存，但是它会被 p.sudogcache 这个“二级缓存”缓存起来。

chansend() 函数最后返回 true 表示成功向 Channel 发送了数据。

5. 小结

关于 channel 发送的源码实现已经分析完了，针对 channel 各个状态做一个小结。

	Channel Status	Result
Write	nil	阻塞
Write	打开但填满	阻塞
Write	打开但未满	成功写入值
Write	关闭	panic
Write	只读	Compile Error

channel 发送过程中包含 2 次有关 goroutine 调度过程：

• 1. 当接收队列中存在 sudog 可以直接发送数据时，执行 goready()将 g 插入 runnext 插槽中，状态从 Gwaiting 或者 Gscanwaiting 改变成 Grunnable，等待下次调度便立即运行。
• 2. 当 channel 阻塞时，执行 gopark() 将 g 阻塞，让出 cpu 的使用权。

需要强调的是，通道并不提供跨 goroutine 的数据访问保护机制。如果通过通道传输数据的一份副本，那么每个 goroutine 都持有一份副本，各自对自己的副本做修改是安全的。当传输的是指向数据的指针时，如果读和写是由不同的 goroutine 完成的，那么每个 goroutine 依旧需要额外的同步操作。

五. 接收数据

从 channel 中接收数据常见代码：

tmp := <-ch
tmp, ok := <-ch

先看等号左边赋值一个值的情况，编译器编译上述代码，在检查 ir 节点时，根据节点 op 不同类型，进行不同的检查，如下源码：

// walkAssign walks an OAS (AssignExpr) or OASOP (AssignOpExpr) node.
func walkAssign(init *ir.Nodes, n ir.Node) ir.Node {
......

	switch as.Y.Op() {
	default:
		as.Y = walkExpr(as.Y, init)

	case ir.ORECV:
		// x = <-c; as.Left is x, as.Right.Left is c.
		// order.stmt made sure x is addressable.
		recv := as.Y.(*ir.UnaryExpr)
		recv.X = walkExpr(recv.X, init)

		n1 := typecheck.NodAddr(as.X)
		r := recv.X // the channel
		return mkcall1(chanfn("chanrecv1", 2, r.Type()), nil, init, r, n1)
		
......
}

as 是入参 ir 节点强制转化成 AssignStmt 类型。AssignStmt 这个类型是赋值的一个说明：

type AssignStmt struct {
	miniStmt
	X   Node
	Def bool
	Y   Node
}

Y 是等号右边的值，它是 Node 类型，里面包含 op 类型。walkAssign 是检查赋值语句，如果 Y.Op() 是 ir.ORECV 类型，说明是 channel 接收的过程。调用 chanrecv1() 函数。as.X 是赋值语句左边的元素，它是接收 channel 中的值，所以它必须是可寻址的。

当从 channel 中读取数据等号左边是 2 个值的时候，编译器在 walkExpr1 中检查这个赋值语句：

func walkExpr1(n ir.Node, init *ir.Nodes) ir.Node {
	switch n.Op() {
	default:
		ir.Dump("walk", n)
		base.Fatalf("walkExpr: switch 1 unknown op %+v", n.Op())
		panic("unreachable")
......

	case ir.OAS2RECV:
		n := n.(*ir.AssignListStmt)
		return walkAssignRecv(init, n)
		
......
}

n.Op() 是 ir.OAS2RECV 类型，将 n 强转成 AssignListStmt 类型：

type AssignListStmt struct {
	miniStmt
	Lhs Nodes
	Def bool
	Rhs Nodes
}

AssignListStmt 和 AssignStmt 作用一样，只是 AssignListStmt 表示等号两边赋值语句不再是一个对象，而是多个。回到 walkExpr1() 中，如果是 ir.OAS2RECV 类型，调用 walkAssignRecv() 继续检查。

func walkAssignRecv(init *ir.Nodes, n *ir.AssignListStmt) ir.Node {
	init.Append(ir.TakeInit(n)...)
	r := n.Rhs[0].(*ir.UnaryExpr) // recv
	walkExprListSafe(n.Lhs, init)
	r.X = walkExpr(r.X, init)
	var n1 ir.Node
	if ir.IsBlank(n.Lhs[0]) {
		n1 = typecheck.NodNil()
	} else {
		n1 = typecheck.NodAddr(n.Lhs[0])
	}
	fn := chanfn("chanrecv2", 2, r.X.Type())
	ok := n.Lhs[1]
	call := mkcall1(fn, types.Types[types.TBOOL], init, r.X, n1)
	return typecheck.Stmt(ir.NewAssignStmt(base.Pos, ok, call))
}

Lhs[0] 是实际接收 channel 值的对象，Lhs[1] 是赋值语句左边第二个 bool 值。赋值语句右边由于只有一个 channel，所以这里 Rhs 也只用到了 Rhs[0]。

//go:nosplit
func chanrecv1(c *hchan, elem unsafe.Pointer) {
	chanrecv(c, elem, true)
}

//go:nosplit
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
	_, received = chanrecv(c, elem, true)
	return
}

综合上述的分析，2 种不同的 channel 接收方式会转换成 runtime.chanrecv1 和 runtime.chanrecv2 两种不同函数的调用，但是最终核心逻辑还是在 runtime.chanrecv 中。

1. 异常检查

chanrecv() 函数一开始先进行异常检查：

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	if debugChan {
		print("chanrecv: chan=", c, "\n")
	}

	if c == nil {
		if !block {
			return
		}
		gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
		throw("unreachable")
	}

	// 简易快速的检查
	if !block && empty(c) {
		if atomic.Load(&c.closed) == 0 {
			return
		}
		if empty(c) {
			// channel 不可逆的关闭了且为空
			if raceenabled {
				raceacquire(c.raceaddr())
			}
			if ep != nil {
				typedmemclr(c.elemtype, ep)
			}
			return true, false
		}
	}

chanrecv() 一上来对 channel 进行检查，如果被 GC 回收了会变为 nil。从一个为 nil 的 channel 中接收数据会发生阻塞。gopark 会引发以 waitReasonChanReceiveNilChan 为原因的休眠，并抛出 unreachable 的 fatal error。当 channel 不为 nil，再开始检查在没有获取锁的情况下会导致接收失败的非阻塞操作。

这里进行的简易快速的检查，检查中状态不能发生变化。这一点和 chansend() 函数有区别。在 chansend() 简易快速的检查中，改变顺序对检查结果无太大影响，但是此处如果检查过程中状态发生变化，如果发生了 racing，检查结果会出现完全相反的错误的结果。例如以下这种情况：channel 在第一个和第二个 if 检查时是打开的且非空，于是在第二个 if 里面 return。但是 return 的瞬间， channel 关闭且空。这样判断出来认为 channel 是打开的且非空。明显是错误的结果，实际上 channel 是关闭且空的。同理检查是否为空的时候也会发生状态反转。为了防止错误的检查结果，c.closed 和 empty() 都必须使用原子检查。

func empty(c *hchan) bool {
	// c.dataqsiz 是不可变的
	if c.dataqsiz == 0 {
		return atomic.Loadp(unsafe.Pointer(&c.sendq.first)) == nil
	}
	return atomic.Loaduint(&c.qcount) == 0
}

这里总共检查了 2 次 empty()，因为第一次检查时， channel 可能还没有关闭，但是第二次检查的时候关闭了，在 2 次检查之间可能有待接收的数据到达了。所以需要 2 次 empty() 检查。

不过就算按照上述源码检查，细心的读者可能还会举出一个反例，例如，关闭一个已经阻塞的同步的 channel，最开始的 !block && empty(c) 为 false，会跳过这个检查。这种情况不能算在正常 chanrecv() 里面。上述是不获取锁的情况检查会接收失败的情况。接下来在获取锁的情况下再次检查一遍异常情况。

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
......
	lock(&c.lock)

	if c.closed != 0 && c.qcount == 0 {
		if raceenabled {
			raceacquire(c.raceaddr())
		}
		unlock(&c.lock)
		if ep != nil {
			typedmemclr(c.elemtype, ep)
		}
		return true, false
	}
......

如果 channel 已经关闭且不存在缓存数据了，则清理 ep 指针中的数据并返回。这里也是从已经关闭的 channel 中读数据，读出来的是该类型零值的原因。

2. 同步接收

同 chansend 逻辑类似，检查完异常情况，紧接着是同步接收。

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
......

	if sg := c.sendq.dequeue(); sg != nil {
		recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
		return true, true
	}
......

在 channel 的发送队列中找到了等待发送的 goroutine。取出队头等待的 goroutine。如果缓冲区的大小为 0，则直接从发送方接收值。否则，对应缓冲区满的情况，从队列的头部接收数据，发送者的值添加到队列的末尾（此时队列已满，因此两者都映射到缓冲区中的同一个下标）。同步接收的核心逻辑见下面 recv() 函数：

func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
	if c.dataqsiz == 0 {
		if raceenabled {
			racesync(c, sg)
		}
		if ep != nil {
			// 从 sender 里面拷贝数据
			recvDirect(c.elemtype, sg, ep)
		}
	} else {
	    // 这里对应 buf 满的情况
		qp := chanbuf(c, c.recvx)
		if raceenabled {
			racenotify(c, c.recvx, nil)
			racenotify(c, c.recvx, sg)
		}
		// 将数据从 buf 中拷贝到接收者内存地址中
		if ep != nil {
			typedmemmove(c.elemtype, ep, qp)
		}
		// 将数据从 sender 中拷贝到 buf 中
		typedmemmove(c.elemtype, qp, sg.elem)
		c.recvx++
		if c.recvx == c.dataqsiz {
			c.recvx = 0
		}
		c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
	}
	sg.elem = nil
	gp := sg.g
	unlockf()
	gp.param = unsafe.Pointer(sg)
	sg.success = true
	if sg.releasetime != 0 {
		sg.releasetime = cputicks()
	}
	goready(gp, skip+1)
}

需要注意的是由于有发送者在等待，所以如果存在缓冲区，那么缓冲区一定是满的。这个情况对应发送阶段阻塞发送的情况，如果缓冲区还有空位，发送的数据直接放入缓冲区，只有当缓冲区满了，才会打包成 sudog，插入到 sendq 队列中等待调度。注意理解这一情况。

接收时主要分为 2 种情况，有缓冲且 buf 满和无缓冲的情况：

• 无缓冲。ep 发送数据不为 nil，调用 recvDirect() 将发送队列中 sudog 存储的 ep 数据直接拷贝到接收者的内存地址中。

• 有缓冲并且 buf 满。有 2 次 copy 操作，先将队列中 recvx 索引下标的数据拷贝到接收方的内存地址，再将发送队列头的数据拷贝到缓冲区中，释放一个 sudog 阻塞的 goroutine。

  有缓冲且 buf 满的情况需要注意，取数据从缓冲队列头取出，发送的数据放在队列尾部，由于 buf 装满，取出的 recvx 指针和发送的 sendx 指针指向相同的下标。

最后调用 goready() 将等待接收的阻塞 goroutine 的状态从 Gwaiting 或者 Gscanwaiting 改变成 Grunnable。下一轮调度时会唤醒这个发送的 goroutine。这部分逻辑和同步发送中一致，关于 goready() 底层实现的代码不在赘述。

3. 异步接收

如果 Channel 的缓冲区中包含一些数据时，从 Channel 中接收数据会直接从缓冲区中 recvx 的索引位置中取出数据进行处理：

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
......

	if c.qcount > 0 {
		// 直接从队列中接收
		qp := chanbuf(c, c.recvx)
		if raceenabled {
			racenotify(c, c.recvx, nil)
		}
		if ep != nil {
			typedmemmove(c.elemtype, ep, qp)
		}
		typedmemclr(c.elemtype, qp)
		c.recvx++
		if c.recvx == c.dataqsiz {
			c.recvx = 0
		}
		c.qcount--
		unlock(&c.lock)
		return true, true
	}

	if !block {
		unlock(&c.lock)
		return false, false
	}
......

上述代码比较简单，如果接收数据的内存地址 ep 不为空，则调用 runtime.typedmemmove() 将缓冲区内的数据拷贝到内存中，并通过 typedmemclr() 清除队列中的数据。

维护 recvx 下标，如果移动到了环形队列的队尾，下标需要回到队头。最后减少 qcount 计数器并释放持有 Channel 的锁。

4. 阻塞接收

如果 channel 发送队列上没有待发送的 goroutine，并且缓冲区也没有数据时，将会进入到最后一个阶段阻塞接收：

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
......

	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	mysg.elem = ep
	mysg.waitlink = nil
	gp.waiting = mysg
	mysg.g = gp
	mysg.isSelect = false
	mysg.c = c
	gp.param = nil
	c.recvq.enqueue(mysg)
	atomic.Store8(&gp.parkingOnChan, 1)
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
......

• 调用 getg() 方法获取当前 goroutine 的指针，用于绑定给一个 sudog。
• 调用 acquireSudog() 方法获取一个 sudog，可能是新建的 sudog，也有可能是从缓存中获取的。设置好 sudog 要发送的数据和状态。比如发送的 Channel、是否在 select 中和待发送数据的内存地址等等。
• 调用 c.recvq.enqueue 方法将配置好的 sudog 加入待发送的等待队列。
• 设置原子信号。当栈要 shrink 收缩时，这个标记代表当前 goroutine 还 parking 停在某个 channel 中。在 g 状态变更与设置 activeStackChans 状态这两个时间点之间的时间窗口进行栈 shrink 收缩是不安全的，所以需要设置这个原子信号。
• 调用 gopark 方法挂起当前 goroutine，状态为 waitReasonChanReceive，阻塞等待 channel。

上面这段代码与 chansend() 中阻塞发送几乎完全一致，区别在于最后一步没有 KeepAlive(ep)。

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
......

	// 被唤醒
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	gp.activeStackChans = false
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	success := mysg.success
	gp.param = nil
	mysg.c = nil
	releaseSudog(mysg)
	return true, success
}

goroutine 被唤醒后会完成 channel 的阻塞数据接收。接收完最后进行基本的参数检查，解除 channel 的绑定并释放 sudog。

5. 小结

关于 channel 接收的源码实现已经分析完了，针对 channel 各个状态做一个小结。

	Channel status	Result
Read	nil	阻塞
Read	打开且非空	读取到值
Read	打开但为空	阻塞
Read	关闭	<默认值>, false
Read	只读	Compile Error

chanrecv 的返回值有几种情况：

tmp, ok := <-ch

Channel status	Selected	Received
nil	false	false
打开且非空	true	true
打开但为空	false	false
关闭且返回值是零值	true	false

received 值会传递给读取 channel 外部的 bool 值 ok，selected 值不会被外部使用。

channel 接收过程中包含 2 次有关 goroutine 调度过程：

1. 当 channel 为 nil 时，执行 gopark() 挂起当前的 goroutine。
2. 当发送队列中存在 sudog 可以直接接收数据时，执行 goready()将 g 插入 runnext 插槽中，状态从 Gwaiting 或者 Gscanwaiting 改变成 Grunnable，等待下次调度便立即运行。
3. 当 channel 缓冲区为空，且没有发送者时，这时 channel 阻塞，执行 gopark() 将 g 阻塞，让出 cpu 的使用权并等待调度器的调度。

六. 关闭 Channel

关于 channel 常见代码：

close(ch)

编译器会将其转换为 runtime.closechan() 方法。

1. 异常检查

func closechan(c *hchan) {
	if c == nil {
		panic(plainError("close of nil channel"))
	}

	lock(&c.lock)
	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("close of closed channel"))
	}

	if raceenabled {
		callerpc := getcallerpc()
		racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
		racerelease(c.raceaddr())
	}
	
	c.closed = 1
......
}

关闭一个 channel 有 2 点需要注意，当 Channel 是一个 nil 空指针或者关闭一个已经关闭的 channel 时，Go 语言运行时都会直接 panic。上述 2 种情况都不存在时，标记 channel 状态为 close。

2. 释放所有 readers 和 writers

关闭 channel 的主要工作是释放所有的 readers 和 writers。

func closechan(c *hchan) {
......
	var glist gList

	for {
		sg := c.recvq.dequeue()
		if sg == nil {
			break
		}
		if sg.elem != nil {
			typedmemclr(c.elemtype, sg.elem)
			sg.elem = nil
		}
		if sg.releasetime != 0 {
			sg.releasetime = cputicks()
		}
		gp := sg.g
		gp.param = unsafe.Pointer(sg)
		sg.success = false
		if raceenabled {
			raceacquireg(gp, c.raceaddr())
		}
		glist.push(gp)
	}
......
}

上述代码是回收接收者的 sudog。将所有的接收者 readers 的 sudog 等待队列（recvq）加入到待清除队列 glist 中。注意这里是先回收接收者。就算从一个 close 的 channel 中读取值，不会发生 panic，顶多读到一个默认零值。

func closechan(c *hchan) {
......

	for {
		sg := c.sendq.dequeue()
		if sg == nil {
			break
		}
		sg.elem = nil
		if sg.releasetime != 0 {
			sg.releasetime = cputicks()
		}
		gp := sg.g
		gp.param = unsafe.Pointer(sg)
		sg.success = false
		if raceenabled {
			raceacquireg(gp, c.raceaddr())
		}
		glist.push(gp)
	}
	unlock(&c.lock)
......
}

再回收发送者 writers。回收步骤和回收接收者是完全一致的，将发送者的等待队列 sendq 中的 sudog 放入待清除队列 glist 中。注意这里可能会产生 panic。在第四章发送数据中分析过，往一个 close 的 channel 中发送数据，会产生 panic，这里不再赘述。

3. 协程调度

最后一步更改 goroutine 的状态。

func closechan(c *hchan) {
......
	for !glist.empty() {
		gp := glist.pop()
		gp.schedlink = 0
		goready(gp, 3)
	}
......
}

最后会为所有被阻塞的 goroutine 调用 goready 触发调度。将所有 glist 中的 goroutine 状态从 _Gwaiting 设置为 _Grunnable 状态，等待调度器的调度。

4. 优雅关闭

“Channel 有几种优雅的关闭方法？” 这种问题常常出现在面试题中，究其原因是因为 Channel 创建容易，但是关闭“不易”：

• 在不改变 Channel 自身状态的条件下，无法知道它是否已经关闭。“不易”之一，关闭时机未知。
• 如果一个 Channel 已经关闭，重复关闭 Channel 会导致 panic。“不易”之二，不能无脑关闭。
• 往一个 close 的 Channel 内写数据，也会导致 panic。“不易”之三，写数据之前也需要关注是否 close 的状态。

	Channel Status	Result
close	nil	panic
close	打开且非空	关闭 Channel；读取成功，直到 Channel 耗尽数据，然后读取产生值的默认值
close	打开但为空	关闭 Channel；读到生产者的默认值
close	关闭	panic
close	只读	Compile Error

那究竟什么时候关闭 Channel 呢？由上面三个“不易”，可以浓缩为 2 点：

• 不能简单的从消费者侧关闭 Channel。
• 如果有多个生产者，它们不能关闭 Channel。

解释一下这 2 个问题。第一个问题，消费者不知道 Channel 何时该关闭。如果关闭了已经关闭的 Channel 会导致 panic。而且分布式应用通常有多个消费者，每个消费者的行为一致，这么多消费者都尝试关闭 Channel 必然会导致 panic。第二个问题，如果有多个生产者往 Channel 内写入数据，这些生产者的行为逻辑也都一致，如果其中一个生产者关闭了 Channel，其他的生产者还在往里写，这个时候会 panic。所以为了防止 panic，必须解决上面这 2 个问题。

关闭 Channel 的方式就 2 种：

• Context
• done channel

Context 的方式在本篇文章不详细展开，详细的可以查看笔者 Context 的那篇文章。本节聊聊 done channel 的做法。假设有多个生产者，有多个消费者。在生产者和消费者之间增加一个额外的辅助控制 channel，用来传递关闭信号。

type session struct {
	done     chan struct{}
	doneOnce sync.Once
	data     chan int
}

func (sess *session) Serve() {
	go sess.loopRead()
	sess.loopWrite()
}

func (sess *session) loopRead() {
	defer func() {
		if err := recover(); err != nil {
			sess.doneOnce.Do(func() { close(sess.done) })
		}
	}()

	var err error
	for {
		select {
		case <-sess.done:
			return
		default:
		}

		if err == io.ErrUnexpectedEOF || err == io.EOF {
			goto failed
		}
	}
failed:
	sess.doneOnce.Do(func() { close(sess.done) })
}

func (sess *session) loopWrite() {
	defer func() {
		if err := recover(); err != nil {
			sess.doneOnce.Do(func() { close(sess.done) })
		}
	}()

	var err error
	for {
		select {
		case <-sess.done:
			return
		case sess.data <- rand.Intn(100):
		}
		
		if err != nil {
			goto done
		}
	}
done:
	if err != nil {
		log("sess: loop write failed: %v, %s", err, sess)
	}
}

func (sess *session) ForceClose() {
	sess.doneOnce.Do(func() { close(sess.done) })
}

消费者侧发送关闭 done channel，由于消费者有多个，如果每一个都关闭 done channel，会导致 panic。所以这里用 doneOnce.Do() 保证只会关闭 done channel 一次。这解决了第一个问题。生产者收到 done channel 的信号以后自动退出。多个生产者退出时间不同，但是最终肯定都会退出。当生产者全部退出以后，data channel 最终没有引用，会被 gc 回收。这也解决了第二个问题，生产者不会去关闭 data channel，防止出现 panic。

总结一下 done channel 的做法：消费者利用辅助的 done channel 发送信号，并先开始退出协程。生产者接收到 done channel 的信号，也开始退出协程。最终 data channel 无人持有，被 gc 回收关闭。

---

Reference：

ACM Communicating sequential processes
Stanford project about csp

线段树 Segment tree 是一种二叉树形数据结构，1977年由 Jon Louis Bentley 发明，用以存储区间或线段，并且允许快速查询结构内包含某一点的所有区间。

一个包含 $ n $ 个区间的线段树，空间复杂度为 $ O(n) $ ，查询的时间复杂度则为$ O(log n+k) $ ，其中 $ k $ 是符合条件的区间数量。线段树的数据结构也可推广到高维度。

一. 什么是线段树

以一维的线段树为例。

令 S 是一维线段的集合。将这些线段的端点坐标由小到大排序，令其为$ x_{1},x_{2},\cdots ,x_{m} $ 。我们将被这些端点切分的每一个区间称为“单位区间”（每个端点所在的位置会单独成为一个单位区间），从左到右包含：

$$
(-\infty ,x_{1}),[x_{1},x_{1}],(x_{1},x_{2}),[x_{2},x_{2}],...,(x_{m-1},x_{m}),[x_{m},x_{m}],(x_{m},+\infty )
$$

线段树的结构为一个二叉树，每个节点都代表一个坐标区间，节点 N 所代表的区间记为 Int(N)，则其需符合以下条件：

• 其每一个叶节点，从左到右代表每个单位区间。
• 其内部节点代表的区间是其两个儿子代表的区间之并集。
• 每个节点（包含叶子）中有一个存储线段的数据结构。若一个线段 S 的坐标区间包含 Int(N) 但不包含 Int(parent(N))，则节点 N 中会存储线段 S。

线段树是二叉树，其中每个节点代表一个区间。通常，一个节点将存储一个或多个合并的区间的数据，以便可以执行查询操作。

二. 为什么需要这种数据结构

许多问题要求我们基于对可用数据范围或区间的查询来给出结果。这可能是一个繁琐而缓慢的过程，尤其是在查询数量众多且重复的情况下。线段树让我们以对数时间复杂度有效地处理此类查询。

线段树可用于计算几何和地理信息系统领域。例如，距中心参考点/原点一定距离的空间中可能会有大量点。假设我们要查找距原点一定距离范围内的点。一个普通的查找表将需要对所有可能的点或所有可能的距离进行线性扫描（假设是散列图）。线段树使我们能够以对数时间实现这一需求，而所需空间却少得多。这样的问题称为平面范围搜索。有效地解决此类问题至关重要，尤其是在处理动态数据且瞬息万变的情况下（例如，用于空中交通的雷达系统）。下文会以线段树解决 Range Sum Query problem 为例。

上图即作为范围查询的线段树。

三. 构造线段树

假设数据存在 size 为 n 的 arr[] 中。

1. 线段树的根通常代表整个数据区间。这里是 arr[0：n-1]。
2. 树的每个叶子代表一个范围，其中仅包含一个元素。 因此，叶子代表 arr[0]，arr[1] 等等，直到 arr[n-1]。
3. 树的内部节点将代表其子节点的合并或并集结果。
4. 每个子节点可代表其父节点所代表范围的大约一半。(二分的思想)

使用大小为 $ \approx 4 \ast n $ 的数组可以轻松表示 n 个元素范围的线段树。（Stack Overflow 对原因进行了很好的讨论。如果你还不确定，请不要担心。本文将在稍后进行讨论。）

下标为 i 的节点有两个节点，下标分别为 $ (2 \ast i + 1) $ 和 $ (2 \ast i + 2)$ 。

线段树看上去很直观并且非常适合递归构造。

我们将使用数组 tree[] 来存储线段树的节点（初始化为全零）。 下标从 0 开始。

• 树的节点在下标 0 处。因此 tree[0] 是树的根。
• tree[i] 的孩子存在 tree[2 * i + 1] 和 tree[2 * i + 2] 中。
• 用额外的 0 或 null 值填充 arr[]，使得 $ n = 2^{k} $ （其中 n 是 arr[] 的总长度，而 k 是非负整数。）
• 叶子节点的下标取值范围在 $ \in [2^{k}-1, 2^{k+1}-2]$

构造线段树的代码如下：

// SegmentTree define
type SegmentTree struct {
	data, tree, lazy []int
	left, right      int
	merge            func(i, j int) int
}

// Init define
func (st *SegmentTree) Init(nums []int, oper func(i, j int) int) {
	st.merge = oper
	data, tree, lazy := make([]int, len(nums)), make([]int, 4*len(nums)), make([]int, 4*len(nums))
	for i := 0; i < len(nums); i++ {
		data[i] = nums[i]
	}
	st.data, st.tree, st.lazy = data, tree, lazy
	if len(nums) > 0 {
		st.buildSegmentTree(0, 0, len(nums)-1)
	}
}

// 在 treeIndex 的位置创建 [left....right] 区间的线段树
func (st *SegmentTree) buildSegmentTree(treeIndex, left, right int) {
	if left == right {
		st.tree[treeIndex] = st.data[left]
		return
	}
	midTreeIndex, leftTreeIndex, rightTreeIndex := left+(right-left)>>1, st.leftChild(treeIndex), st.rightChild(treeIndex)
	st.buildSegmentTree(leftTreeIndex, left, midTreeIndex)
	st.buildSegmentTree(rightTreeIndex, midTreeIndex+1, right)
	st.tree[treeIndex] = st.merge(st.tree[leftTreeIndex], st.tree[rightTreeIndex])
}

func (st *SegmentTree) leftChild(index int) int {
	return 2*index + 1
}

func (st *SegmentTree) rightChild(index int) int {
	return 2*index + 2
}

笔者将线段树合并的操作变成了一个函数。合并操作根据题意变化，常见的有加法，取 max，min 等等。

我们以 arr[] = [18, 17, 13, 19, 15, 11, 20, 12, 33, 25 ] 为例构造线段树：

线段树构造好以后，数组里面的数据是：

tree[] = [ 183, 82, 101, 48, 34, 43, 58, 35, 13, 19, 15, 31, 12, 33, 25, 18, 17, 0, 0, 0, 0, 0, 0, 11, 20, 0, 0, 0, 0, 0, 0 ]

线段树用 0 填充到 4*n 个元素。

LeetCode 对应题目是 218. The Skyline Problem、303. Range Sum Query - Immutable、307. Range Sum Query - Mutable、699. Falling Squares

四. 线段树的查询

线段树的查询方法有两种，一种是直接查询，另外一种是懒查询。

1. 直接查询

当查询范围与当前节点表示的范围完全匹配时，该方法返回结果。否则，它会更深入地遍历线段树树，以找到与节点的一部分完全匹配的节点。

// 查询 [left....right] 区间内的值

// Query define
func (st *SegmentTree) Query(left, right int) int {
	if len(st.data) > 0 {
		return st.queryInTree(0, 0, len(st.data)-1, left, right)
	}
	return 0
}

// 在以 treeIndex 为根的线段树中 [left...right] 的范围里，搜索区间 [queryLeft...queryRight] 的值
func (st *SegmentTree) queryInTree(treeIndex, left, right, queryLeft, queryRight int) int {
	if left == queryLeft && right == queryRight {
		return st.tree[treeIndex]
	}
	midTreeIndex, leftTreeIndex, rightTreeIndex := left+(right-left)>>1, st.leftChild(treeIndex), st.rightChild(treeIndex)
	if queryLeft > midTreeIndex {
		return st.queryInTree(rightTreeIndex, midTreeIndex+1, right, queryLeft, queryRight)
	} else if queryRight <= midTreeIndex {
		return st.queryInTree(leftTreeIndex, left, midTreeIndex, queryLeft, queryRight)
	}
	return st.merge(st.queryInTree(leftTreeIndex, left, midTreeIndex, queryLeft, midTreeIndex),
		st.queryInTree(rightTreeIndex, midTreeIndex+1, right, midTreeIndex+1, queryRight))
}

在上面的示例中，查询的区间范围为[2，8] 的元素之和。没有任何线段可以完全代表[2，8] 范围。但是可以观察到，可以使用范围 [2，2]，[3，4]，[5，7]，[8，8] 这 4 个区间构成 [8，8]。快速验证 [2,8] 处的输入元素之和为 13 + 19 + 15 + 11 + 20 + 12 + 33 = 123。[2，2]，[3，4]，[5，7] 和 [8，8] 的节点总和是 13 + 34 + 43 + 33 = 123。答案正确。

2. 懒查询

懒查询对应懒更新，两者是配套操作。在区间更新时，并不直接更新区间内所有节点，而是把区间内节点增减变化的值存在 lazy 数组中。等到下次查询的时候再把增减应用到具体的节点上。这样做也是为了分摊时间复杂度，保证查询和更新的时间复杂度在 O(log n) 级别，不会退化成 O(n) 级别。

懒查询节点的步骤：

1. 先判断当前节点是否是懒节点。通过查询 lazy[i] 是否为 0 判断。如果是懒节点，将它的增减变化应用到该节点上。并且更新它的孩子节点。这一步和更新操作的第一步完全一样。
2. 递归查询子节点，以找到适合的查询节点。

具体代码如下：

// 查询 [left....right] 区间内的值

// QueryLazy define
func (st *SegmentTree) QueryLazy(left, right int) int {
	if len(st.data) > 0 {
		return st.queryLazyInTree(0, 0, len(st.data)-1, left, right)
	}
	return 0
}

func (st *SegmentTree) queryLazyInTree(treeIndex, left, right, queryLeft, queryRight int) int {
	midTreeIndex, leftTreeIndex, rightTreeIndex := left+(right-left)>>1, st.leftChild(treeIndex), st.rightChild(treeIndex)
	if left > queryRight || right < queryLeft { // segment completely outside range
		return 0 // represents a null node
	}
	if st.lazy[treeIndex] != 0 { // this node is lazy
		for i := 0; i < right-left+1; i++ {
			st.tree[treeIndex] = st.merge(st.tree[treeIndex], st.lazy[treeIndex])
			// st.tree[treeIndex] += (right - left + 1) * st.lazy[treeIndex] // normalize current node by removing lazinesss
		}
		if left != right { // update lazy[] for children nodes
			st.lazy[leftTreeIndex] = st.merge(st.lazy[leftTreeIndex], st.lazy[treeIndex])
			st.lazy[rightTreeIndex] = st.merge(st.lazy[rightTreeIndex], st.lazy[treeIndex])
			// st.lazy[leftTreeIndex] += st.lazy[treeIndex]
			// st.lazy[rightTreeIndex] += st.lazy[treeIndex]
		}
		st.lazy[treeIndex] = 0 // current node processed. No longer lazy
	}
	if queryLeft <= left && queryRight >= right { // segment completely inside range
		return st.tree[treeIndex]
	}
	if queryLeft > midTreeIndex {
		return st.queryLazyInTree(rightTreeIndex, midTreeIndex+1, right, queryLeft, queryRight)
	} else if queryRight <= midTreeIndex {
		return st.queryLazyInTree(leftTreeIndex, left, midTreeIndex, queryLeft, queryRight)
	}
	// merge query results
	return st.merge(st.queryLazyInTree(leftTreeIndex, left, midTreeIndex, queryLeft, midTreeIndex),
		st.queryLazyInTree(rightTreeIndex, midTreeIndex+1, right, midTreeIndex+1, queryRight))
}

五. 线段树的更新

1. 单点更新

单点更新类似于 buildSegTree。更新树的叶子节点的值，该值与更新后的元素相对应。这些更新的值会通过树的上层节点把影响传播到根。

// 更新 index 位置的值

// Update define
func (st *SegmentTree) Update(index, val int) {
	if len(st.data) > 0 {
		st.updateInTree(0, 0, len(st.data)-1, index, val)
	}
}

// 以 treeIndex 为根，更新 index 位置上的值为 val
func (st *SegmentTree) updateInTree(treeIndex, left, right, index, val int) {
	if left == right {
		st.tree[treeIndex] = val
		return
	}
	midTreeIndex, leftTreeIndex, rightTreeIndex := left+(right-left)>>1, st.leftChild(treeIndex), st.rightChild(treeIndex)
	if index > midTreeIndex {
		st.updateInTree(rightTreeIndex, midTreeIndex+1, right, index, val)
	} else {
		st.updateInTree(leftTreeIndex, left, midTreeIndex, index, val)
	}
	st.tree[treeIndex] = st.merge(st.tree[leftTreeIndex], st.tree[rightTreeIndex])
}

在此示例中，下标为（在原始输入数据中）1、3 和 6 处的元素分别增加了 +3，-1 和 +2。可以看到更改如何沿树传播，一直到根。

2. 区间更新

线段树仅更新单个元素，非常有效，时间复杂度 O(log n)。 但是，如果我们要更新一系列元素怎么办？按照当前的方法，每个元素都必须独立更新，每个元素都会花费一些时间。分别更新每一个叶子节点意味着要多次处理它们的共同祖先。祖先节点可能被更新多次。如果想要减少这种重复计算，该怎么办？

在上面的示例中，根节点被更新了三次，而编号为 82 的节点被更新了两次。这是因为更新叶子节点对上层父亲节点有影响。最差的情况，查询的区间内不包含频繁更新的元素，于是需要花费很多时间更新不怎么访问的节点。增加额外的 lazy 数组，可以减少不必要的计算，并且能按需处理节点。

使用另一个数组 lazy[]，它的大小与我们的线段树 array tree[] 完全相同，代表一个惰性节点。当访问或查询该节点时，lazy[i] 中保留需要增加或者减少该节点 tree[i] 的数量。 当 lazy[i] 为 0 时，表示 tree[i] 该节点不是惰性的，并且没有缓存的更新。

更新区间内节点的步骤：

1. 先判断当前节点是否是懒节点。通过查询 lazy[i] 是否为 0 判断。如果是懒节点，将它的增减变化应用到该节点上。并且更新它的孩子节点。
2. 如果当前节点代表的区间位于更新范围内，则将当前更新操作应用于当前节点。
3. 递归更新子节点。

具体代码如下：

// 更新 [updateLeft....updateRight] 位置的值
// 注意这里的更新值是在原来值的基础上增加或者减少，而不是把这个区间内的值都赋值为 x，区间更新和单点更新不同
// 这里的区间更新关注的是变化，单点更新关注的是定值
// 当然区间更新也可以都更新成定值，如果只区间更新成定值，那么 lazy 更新策略需要变化，merge 策略也需要变化，这里暂不详细讨论

// UpdateLazy define
func (st *SegmentTree) UpdateLazy(updateLeft, updateRight, val int) {
	if len(st.data) > 0 {
		st.updateLazyInTree(0, 0, len(st.data)-1, updateLeft, updateRight, val)
	}
}

func (st *SegmentTree) updateLazyInTree(treeIndex, left, right, updateLeft, updateRight, val int) {
	midTreeIndex, leftTreeIndex, rightTreeIndex := left+(right-left)>>1, st.leftChild(treeIndex), st.rightChild(treeIndex)
	if st.lazy[treeIndex] != 0 { // this node is lazy
		for i := 0; i < right-left+1; i++ {
			st.tree[treeIndex] = st.merge(st.tree[treeIndex], st.lazy[treeIndex])
			//st.tree[treeIndex] += (right - left + 1) * st.lazy[treeIndex] // normalize current node by removing laziness
		}
		if left != right { // update lazy[] for children nodes
			st.lazy[leftTreeIndex] = st.merge(st.lazy[leftTreeIndex], st.lazy[treeIndex])
			st.lazy[rightTreeIndex] = st.merge(st.lazy[rightTreeIndex], st.lazy[treeIndex])
			// st.lazy[leftTreeIndex] += st.lazy[treeIndex]
			// st.lazy[rightTreeIndex] += st.lazy[treeIndex]
		}
		st.lazy[treeIndex] = 0 // current node processed. No longer lazy
	}

	if left > right || left > updateRight || right < updateLeft {
		return // out of range. escape.
	}

	if updateLeft <= left && right <= updateRight { // segment is fully within update range
		for i := 0; i < right-left+1; i++ {
			st.tree[treeIndex] = st.merge(st.tree[treeIndex], val)
			//st.tree[treeIndex] += (right - left + 1) * val // update segment
		}
		if left != right { // update lazy[] for children
			st.lazy[leftTreeIndex] = st.merge(st.lazy[leftTreeIndex], val)
			st.lazy[rightTreeIndex] = st.merge(st.lazy[rightTreeIndex], val)
			// st.lazy[leftTreeIndex] += val
			// st.lazy[rightTreeIndex] += val
		}
		return
	}
	st.updateLazyInTree(leftTreeIndex, left, midTreeIndex, updateLeft, updateRight, val)
	st.updateLazyInTree(rightTreeIndex, midTreeIndex+1, right, updateLeft, updateRight, val)
	// merge updates
	st.tree[treeIndex] = st.merge(st.tree[leftTreeIndex], st.tree[rightTreeIndex])
}

LeetCode 对应题目是 218. The Skyline Problem、699. Falling Squares

六. 时间复杂度分析

让我们看一下构建过程。我们访问了线段树的每个叶子（对应于数组 arr[] 中的每个元素）。因此，我们处理大约 2 * n 个节点。这使构建过程时间复杂度为 O(n)。对于每个递归更新的过程都将丢弃区间范围的一半，以到达树中的叶子节点。这类似于二分搜索，只需要对数时间。更新叶子后，将更新树的每个级别上的直接祖先。这花费时间与树的高度成线性关系。

4*n 个节点可以确保将线段树构建为完整的二叉树，从而树的高度为 log(4*n + 1) 向上取整。线段树读取和更新的时间复杂度都为 O(log n)。

七. 常见题型

1. Range Sum Queries

Range Sum Queries 是 Range Queries 问题的子集。给定一个数据元素数组或序列，需要处理由元素范围组成的读取和更新查询。线段树 Segment Tree 和树状数组 Binary Indexed Tree (a.k.a. Fenwick Tree)) 都能很快的解决这类问题。

Range Sum Query 问题专门处理查询范围内的元素总和。这个问题存在许多变体，包括不可变数据，可变数据，多次更新，单次查询 和 多次更新，多次查询。

2. 单点更新

• HDU 1166 敌兵布阵 update:单点增减 query:区间求和
• HDU 1754 I Hate It update:单点替换 query:区间最值
• HDU 1394 Minimum Inversion Number update:单点增减 query:区间求和
• HDU 2795 Billboard query:区间求最大值的位子(直接把update的操作在query里做了)

3. 区间更新

• HDU 1698 Just a Hook update:成段替换 (由于只query一次总区间,所以可以直接输出 1 结点的信息)
• POJ 3468 A Simple Problem with Integers update:成段增减 query:区间求和
• POJ 2528 Mayor’s posters 离散化 + update:成段替换 query:简单hash
• POJ 3225 Help with Intervals update:成段替换,区间异或 query:简单hash

4. 区间合并

这类题目会询问区间中满足条件的连续最长区间,所以PushUp的时候需要对左右儿子的区间进行合并

• POJ 3667 Hotel update:区间替换 query:询问满足条件的最左端点

5. 扫描线

这类题目需要将一些操作排序,然后从左到右用一根扫描线扫过去最典型的就是矩形面积并,周长并等题

• HDU 1542 Atlantis update:区间增减 query:直接取根节点的值
• HDU 1828 Picture update:区间增减 query:直接取根节点的值

6. 计数问题

在 LeetCode 中还有一类问题涉及到计数的。315. Count of Smaller Numbers After Self，327. Count of Range Sum，493. Reverse Pairs 这类问题可以用下面的套路解决。线段树的每个节点存的是区间计数。

// SegmentCountTree define
type SegmentCountTree struct {
	data, tree  []int
	left, right int
	merge       func(i, j int) int
}

// Init define
func (st *SegmentCountTree) Init(nums []int, oper func(i, j int) int) {
	st.merge = oper

	data, tree := make([]int, len(nums)), make([]int, 4*len(nums))
	for i := 0; i < len(nums); i++ {
		data[i] = nums[i]
	}
	st.data, st.tree = data, tree
}

// 在 treeIndex 的位置创建 [left....right] 区间的线段树
func (st *SegmentCountTree) buildSegmentTree(treeIndex, left, right int) {
	if left == right {
		st.tree[treeIndex] = st.data[left]
		return
	}
	midTreeIndex, leftTreeIndex, rightTreeIndex := left+(right-left)>>1, st.leftChild(treeIndex), st.rightChild(treeIndex)
	st.buildSegmentTree(leftTreeIndex, left, midTreeIndex)
	st.buildSegmentTree(rightTreeIndex, midTreeIndex+1, right)
	st.tree[treeIndex] = st.merge(st.tree[leftTreeIndex], st.tree[rightTreeIndex])
}

func (st *SegmentCountTree) leftChild(index int) int {
	return 2*index + 1
}

func (st *SegmentCountTree) rightChild(index int) int {
	return 2*index + 2
}

// 查询 [left....right] 区间内的值

// Query define
func (st *SegmentCountTree) Query(left, right int) int {
	if len(st.data) > 0 {
		return st.queryInTree(0, 0, len(st.data)-1, left, right)
	}
	return 0
}

// 在以 treeIndex 为根的线段树中 [left...right] 的范围里，搜索区间 [queryLeft...queryRight] 的值，值是计数值
func (st *SegmentCountTree) queryInTree(treeIndex, left, right, queryLeft, queryRight int) int {
	if queryRight < st.data[left] || queryLeft > st.data[right] {
		return 0
	}
	if queryLeft <= st.data[left] && queryRight >= st.data[right] || left == right {
		return st.tree[treeIndex]
	}
	midTreeIndex, leftTreeIndex, rightTreeIndex := left+(right-left)>>1, st.leftChild(treeIndex), st.rightChild(treeIndex)
	return st.queryInTree(rightTreeIndex, midTreeIndex+1, right, queryLeft, queryRight) +
		st.queryInTree(leftTreeIndex, left, midTreeIndex, queryLeft, queryRight)
}

// 更新计数

// UpdateCount define
func (st *SegmentCountTree) UpdateCount(val int) {
	if len(st.data) > 0 {
		st.updateCountInTree(0, 0, len(st.data)-1, val)
	}
}

// 以 treeIndex 为根，更新 [left...right] 区间内的计数
func (st *SegmentCountTree) updateCountInTree(treeIndex, left, right, val int) {
	if val >= st.data[left] && val <= st.data[right] {
		st.tree[treeIndex]++
		if left == right {
			return
		}
		midTreeIndex, leftTreeIndex, rightTreeIndex := left+(right-left)>>1, st.leftChild(treeIndex), st.rightChild(treeIndex)
		st.updateCountInTree(rightTreeIndex, midTreeIndex+1, right, val)
		st.updateCountInTree(leftTreeIndex, left, midTreeIndex, val)
	}
}