作者：周爽
出版社：机械工业出版社
格式: AZW3, DOCX, EPUB, MOBI, PDF, TXT
实时流计算系统设计与实现试读：
前言
为什么要写这本书大概但凡写文章，都应该是在夜深人静的时候吧。
回顾来沪四年，思绪一下子回到了到沪的第一天。那是我第一次来上海。虽然没有小马哥勇闯上海滩的豪情万丈，但似乎也并未十分惆怅。拖着行李箱去新公司报到，当时住的地方尚未找好，新公司的名字竟也还未确定。报到的新公司，真的是一家“新”公司，刚刚成立不久，办公室在一个略显陈旧的大楼内，后来才知道那个办公室内还有另外一个公司的人员在那儿办公。当时辉哥接待了我，交给我一个Mac电脑，给我讲了一下公司的业务和系统现状。那是我第一次使用Mac电脑，操作还十分生疏。而现在，我正用着这台电脑，写着这篇自序（前言）。后来海阳到了，他给我展示和讲解了公司当时的产品。再后来，他还教了我怎样用Mac电脑。下午的时候我提前下班，拖着行李箱找住的地方去了。那天，当忙完一切躺到床上时，我意识到一切真的都是新的开始了。新的城市、新的公司、新的同事、新的业务、新的领域和新的工作内容。
之后就是正式的工作了。
因为是做Java开发的，所以花一天半时间学了Django，之后接手了Python后台开发。
因为是做Java开发的，所以花一天时间装了Android Studio，开始了SDK与后台的适配。
因为是做Java开发的，所以Java后台服务更是当仁不让了。
因为是做Java开发的，所以也要负责业务核心算法的开发。
因为是做Java开发的，所以DevOps也得负责推进吧。
因为是做Java开发的，所以……
那应该是我最专心地做开发的一段时间，甚至在一年之后的公司年会上，我还因此获得了一个“最佳沉默奖”。其实并非我不说话，而是每次市场部的同事看到我时，我都是在座位上写程序，好像我从来没离开过座位一样。是的，我一直在开发着，但是并不感到忙碌和紧迫，因为有充足的时间去思考问题、验证猜想并最终解决问题。
从2015年开始，公司因为业务需要，开始涉及实时流计算领域。当时流计算技术远非像现在这样普及，Flink还没有在国内流行起来，Spark Streaming处于“半吊子”状态，Storm则还是简单的TopologyBuilder。不过最主要的问题还是，虽然当时公司意识到要使用实时流计算技术，但没有人真正理解实时流计算系统到底该怎么用，应该怎样将流计算真正贴切地运用到我们的业务需求中。
在这种情况下，作为后台开发的我开始了自己的思考和探索。其实解决问题的思路非常简单，用最自然、最贴切、最实际、最节省资源的方式去解决真实的业务问题。最开始，我们选择Akka来作为流计算框架，但是因为对Akka的特性理解不到位，后来开发出的程序出现各种问题。例如，在Actor中光明正大地休眠（sleep）、对反向压力不管不顾，结果程序总是时不时宕掉、错误使用Akka Cluster导致集群脑裂等。
在程序开发过程中，我是一个谨小慎微的人，对于任何不确定性的因素，只要想到了就一定会尽力去避免，即使当时其他人不甚理解。所以从一开始，我就在自己负责的模块中，对Akka添加了反向压力的支持，就是为了避免执行步调不一致时导致的OOM。从后来的结果看，当时的做法是非常正确的。虽然在Akka中添加了反向压力的支持，但回看起来，实现得过于复杂。虽然保证了反向压力带来的程序稳定性，可是在50行的代码中，只有1行代码是涉及业务处理的。这种解决方案显然非常不明智。
当时，我还在极力尝试尽可能提高程序的性能，希望充分“榨干”机器的CPU和I/O资源，以尽可能降低硬件成本。经过一段时间的调研和思考，我逐渐发现，NIO和异步才是彻底“榨干”CPU与I/O资源的关键所在。虽然纤程（或协程）也是一种充分利用资源的完美手段，但可惜当时JVM领域尚无一个公开好用的完美纤程实现方案。
那时候，我开始隐隐约约地意识到，似乎“流”是一种非常好的编程模式。
首先，“流”与“异步”不谋而合。“流”的各个节点通过队列传递消息，不同节点的执行正好就是完全异步的。另外，由于队列隔离，不同节点的执行完全不用考虑并发安全的问题。
其次，如果“流”的执行节点间使用的是阻塞队列，那么整个流的各个执行环节就天然地带有了反向压力能力，不需要像之前在Akka中那样非常复杂的实现。
再次，“流”能够非常自然地描述业务执行的流程。不管是大到整个产品线的各个服务模块，还是小到每个服务模块中的具体实现步骤，就像“分形”一样，“流”能够做任意细粒度的划分。这是一种非常普遍的描述事情发生过程的模式。
最后，通过Kafka这种消息中间件的隔离，我可以非常清晰地定义好自己负责开发模块的责任边界，与其他同事的程序隔离开来，避免纠缠不清。当然，这是一种自私的想法，但是从设计模式高内聚、低耦合的角度来看，这又何尝不是一种非常不错的实践呢？更何况Kafka这种好用到“爆”的消息队列，真的是让人爱不释手！
于是，说干就干，我花了一个周末的时间，编写了第一个版本的流计算框架。之后又经过几次大大小小的调整和改进，最终，这个流计算框架进入了公司所有产品的主要业务模块中。再后来，我又在这个流计算框架上开发了一个特征引擎，支持DSL和脚本，可以非常灵活、方便、快速地在流数据上即写即算地实现各种特征计算。所有特征计算都是并发处理，并且自动解析特征依赖与优化执行过程，可以说还是有一丁点儿小小的惊艳了。
4年的时光转瞬即逝。其实这4年还有太多太多的事情，从后台到前端、从开发到运维、从数据到系统、从服务器到嵌入式，编写程序、负责项目、担任架构师，一路走来，我有太多的收获，也有太多的感触。所以，我希望赶在而立之年前，对自己的这些收获和感触做一个总结，一方面是给自己人生阶段的交代，另一方面希望这些经验能够给后来的开发者带来些许帮助。读者对象
本书主要适合于以下读者：
·Java软件开发人员；
·实时计算工程师和架构师；
·分布式系统工程师和架构师。本书特色
本书总结了实时流计算系统的通用架构模式。通过从无到有构建一个流计算编程框架，让读者了解流计算应用计算的任务类型，学会解决计算过程遇到的各种问题和难点。本书希望让读者领会Java程序开发中“流”这种编程方式的优势和乐趣所在。另外，通过将单节点流计算应用扩展为分布式集群，让读者理解分布式系统的架构模式，并能准确看待开源社区中各种眼花缭乱的流计算框架，看透这些流计算框架的本质，避免选择恐惧症。本书还探讨了实时流计算能够与不能够解决的问题，让读者对流计算系统的能力了然于胸，不至于钻牛角尖。总而言之，读者在阅读本书后，能够对实时流计算系统有清晰的认识和理解，在架构设计、系统实现和具体应用方面都能做到心有丘壑，最终做出优秀的实时流计算应用产品。如何阅读本书
首先需要澄清的是，本书的“非目标”是什么：
·各种流计算框架实战，诸如教读者如何使用Storm、Spark、Flink等流计算框架。笔者相信，针对每一种具体的流计算框架已经有许多优秀的书籍了。如果笔者再讲，就是不自量力、班门弄斧、狗尾续貂了。
澄清了本书的“非目标”，就可以定义本书的“目标”了：
·总结实时流计算系统的通用架构模式。所谓架构模式，是一种“形而上”的东西，也就是所谓的“道”。实时流计算系统体现出的软件设计之“道”，是笔者试图阐述的东西。
·从无到有构建一个“麻雀虽小，五脏俱全”的单节点实时流计算框架。通过这个造轮子的过程，我们会深入理解流计算系统中最本质、最困难、最容易混淆的概念。之后通过在多种开源流计算框架中多次验证这些概念，实现“道”向“形而
·下”的具象，让我们以后面对各种流计算框架时，都能够做到胸有成竹。
·通过将单节点的实时流计算框架扩展为分布式实时流计算框架，让读者理解多种不同的分布式系统构建模式。
·通过“流”这种异步编程模式，让读者理解并掌握编写高性能程序的编程之道，领略Java高并发编程的乐趣。
·不仅探讨实时流计算能够解决的问题，而且要明白当实在做不到“实时”时该如何进行架构设计。
·尽可能全面覆盖一个完整的实时流计算系统，包括许多周边系统，如存储系统、服务治理和配置管理等。如果这些“绿叶”点缀得不好，有时也会给实时流计算系统带来不利影响。
整体而言，本书的内容按照“总分”的结构组织。全书分为11章。
第1章介绍实时流计算技术的产生背景、使用场景和通用架构。
第2章通过实时流计算数据的采集，详细分析Java平台NIO和异步编程的基础，并初步讨论了“异步”和“流”这两种编程模式之间的关系。
第3～5章通过从零开始构造分布式实时流计算应用，详细剖析了实时流计算系统的计算任务类型、核心概念和技术关键点。
第6章通过多种开源流计算框架，验证第3～5章所讨论的实时流计算系统核心概念和技术关键点。
第7章讨论当实在做不到“实时”时，我们应该做出的备选方案。
第8～10章讨论构建完整实时流计算系统时必要的周边辅助系统。
第11章详细讨论两个实时流计算应用的具体案例。
另外，本书包含许多示例代码，但由于篇幅有限，这些代码不能完全展现在书中。读者可从GitHub仓库（https://github.com/alain898/real_time_stream_computing_book_source_code）获取本书完整的配套源代码。勘误和支持
由于笔者水平有限，编写时间仓促，书中难免有一些错误或者不准确的地方，恳请读者批评指正。大家可以通过电子邮箱347041583@qq.com联系笔者。期待得到大家的真挚反馈，在技术之路上互勉共进。致谢
首先要感谢来沪四年遇到的许多同事，辉哥、海阳、克克、波叔、Lex、沛沛、亘哥、佳哥、牙膏、Kevin、军军、建军、叶晗、Mike、国聪、俊华、培良、凯哥、国圣、志源、顾大神、渊总、荣哥、君姐、Vivi姐及其他更多同事，我从你们身上学到了太多太多，这些知识会让我受用终身！
感谢挚友飞哥、乔姐和Lisa，是你们让我在沪四年不再有漂泊感。
感谢机械工业出版社华章公司的高婧雅编辑，给予这本书出版的机会，并在写作期间一直给予我支持和鼓励，并引导我顺利完成全部书稿。
最后感谢我的爸爸、妈妈、弟弟、三叔、四叔、小姨、爷爷和奶奶，感谢你们伴随我成长。虽然很多话我说不出口，但我是在乎你们的！
谨以此书献给我最亲爱的家人，以及众多热爱Java的朋友们！周爽2019年10月第1章 实时流计算
两千多年以前，孔老夫子站在大河边，望着奔流而去的河水，不禁感叹：“逝者如斯夫，不舍昼夜。”老夫子是在叹惜着韶华白首，时光易逝！
两千多年以后的今天，当你我抱着手机读书、追剧、抢票、“剁手”、刷小视频、发红包的时候，一道道信息流正在以光速在世界范围内传递和传播。自从互联网和物联网诞生以来，人与人、人与物、物与物之间的互联和互动愈加紧密与频繁，大量各式各样的数据在互联和互动的过程中产生。海量的数据洪流将我们的时间和空间越占越满，以至于让我们开始疲于奔命，鲜有时间和能力再去感受与思考那些一瞬间的百万种可能。1.1 大数据时代的新挑战：实时流计算
社会需求和科技进步是螺旋式相互促进和提升的。“大数据”一词最早由Roger Mougalas在2005年提出，所以我们姑且认为2005年是大数据时代的元年吧。大数据技术之所以出现，是因为社会发展的程度已经开始要求我们具备处理海量数据的能力。之后，大数据技术逐渐发展和日趋完善的过程又反过来进一步促进社会产生更多、更丰富的数据。随着大数据技术的普及，IBM公司为我们总结了大数据的五大特点（也称为5V特点），即Volume（大量）、Velocity（快速）、Variety（多样）、Veracity（真实）和Value（价值），如图1-1所示。图1-1 大数据的五大特点
大数据时代为人们带来了丰富多彩的生活方式，让人们充分享受着从大数据中挖掘而来的价值。但也正因为大数据产生得太多太快，让我们开始疲于对正在发生的事情做出及时反应。就像火灾已经爆发后才知道救火，交通已经阻塞后才知道疏通，羊毛已经被“羊毛党”薅光后才知道堵上漏洞，股价已经拉升后才知道后悔……为什么我们不能在这些事情发生之前，或者至少是刚刚发生的时候就提前收到预警和通知，并且及时采取应对措施呢？
是的，面对无穷无尽的数据洪流，我们急需一种手段来帮助我们抓住并思考那些一闪而逝的瞬间。在这样的背景下，实时流计算技术应运而生。虽然不能像电影《超体》中女主角直接用手抓住并分析电磁波信息那样，但至少实时流计算技术能够帮助我们抓住数据流的瞬间，分析并挖掘出数据的实时价值。千万不要小瞧了数据的实时价值。据说在很久以前的欧洲战场上，每次最先知道战争结果的不是后方的政府机构，而是股票交易所里的那些股票投资者。俗话说，时间就是金钱，效率就是生命。所有实时流计算的目的都是为了获得数据的实时价值。如果数据没有实时价值，那么实时流计算也就失去了它存在的意义。1.2 实时流计算使用场景
话说有一句至理名言：“天下武功，无坚不摧，唯快不破！”由此足可见“快”的重要性。更快、更完整地获取数据，更快、更充分地挖掘出数据价值，已成为大数据时代各行各业的共识。在线系统监控、移动数据和物联网、金融风控、推荐系统等，虽然行业各不相同，但是它们有个共同点——实时流计算技术在这些领域发挥着越来越重要的作用。1.在线系统监控
互联网行业蓬勃发展的背后，是各家企业机房里成千上万的服务器。服务器在7×24小时（传说中的007工作制）的作业过程中产生大量监控数据。这些数据包含着服务器本身的健康状况，如硬件状态、资源使用情况和负载压力等。第一时间知道服务器的健康状况是非常重要的，可以避免因为一台服务器宕机而后续造成的各种雪崩效应。除了服务器本身以外，复杂的线上业务系统产生着更多的数据。如今一个每天亿万级别访问量的系统已经司空见惯，产品花样更是层出不穷。业务系统产生日志的数量级由GB变TB，再由TB变PB。将线上日志导入实时流计算系统，我们可以实现一系列有实时价值的功能。最基本的功能是监控业务是否运营正常，如监控业务关键指标、发现故障模式等。高级些的功能是最大程度优化业务使用服务器的成本，如根据CPU、内存和I/O等资源的使用率动态扩展或缩减业务使用的服务器数量。更高级的功能是挖掘和探索新的业务模式，如CEP（Complex Event Processing，复杂事件处理）和在线统计学习或机器学习模型的各种运用等。通过实时流计算技术，实时展现业务系统的健康状况，提前避免可能的业务故障，最大程度优化业务使用服务器的成本，抢先发现新的业务模式和商机……这些都是实时流计算技术在在线系统监控领域价值的体现。2.移动数据和物联网
移动终端、智能交通、共享单车、5G、工业4.0……如今在我们生活的时代，一波又一波的新名词层出不穷。“移动”和“物联”让数据变得随时随地可得。数据越来越多，单位数据自身的价值却越来越小。实时处理海量数据洪流，已成为移动和物联网领域的当务之急。例如，对于智能交通系统，传统智能交通系统采用离线方式对交通数据做分析，交通决策不能及时做出；而通过对交通数据流进行实时分析，实时展现交通热点路段、优化信号灯配时、指导行车线路，可实实在在减轻当前热点路段压力、缩减平均行车时间，如图1-2所示。像智能交通这样，优化生活环境，正是实时流计算技术在移动数据和物联网领域体现的价值之一。[1]图1-2 基于Spar.Streaming的Uber交通热点路段分析及可视化系统3.金融风控
金融风控是实时流计算技术又一常用领域，如图1-3所示。通常针对贷款的风控，可以分为贷前、贷中和贷后。在贷中和贷后，大多采用离线数据分析和数据可视化技术来实现风险控制。但是在贷前，特别是在许多现金贷产品中，为了给用户带来更好的产品体验，必须在很短的时间内对用户的信用、还款能力和还款意愿等做出评估。除了针对用户本身的信用风险作分析外，还需要防止金融欺诈问题，如“薅羊毛”和多头借贷。通过实时流计算技术，在秒级甚至亚秒级，对用户信用和欺诈风险做出判定，在保证可控风险的同时，提供良好的用户体验，进一步提高现金贷产品整体的竞争力。[2]图1-3 基于Flink的实时欺诈检测平台4.实时推荐
实时推荐是实时流计算技术的另一个常见应用场景。如今手机几乎成为每一个年轻人的必备品。打开手机，听音乐、浏览新闻、阅读小说、看到心仪的东西买买买……有一天你突然发现，手机应用越来越了解自己。它们知道推荐什么样的音乐、新闻、小说和商品，并且推荐的东西大抵还是你所喜欢的。现代推荐系统（见图1-4）背后越来越多地出现了实时流计算技术的影子，通过实时分析从用户手机上收集而来的行为数据，发掘用户的兴趣、偏好，给用户推荐可能感兴趣的内容或商品。或许很多人并不喜欢这种被机器引导的感觉，但是我们还是不可避免地越来越多地被它们所影响。[3]图1-4 基于Spar.Streaming的实时零售推荐系统
这里只是简单地列举了几个流计算技术使用的场景。其实在越来越多的行业，很多传统上用离线批处理技术完成的事情也逐渐转变为采用实时流计算技术完成。所以，读者不妨大胆发挥想象力，试着将实时流计算技术运用到生活的各个方面去，挖掘实时信息的潜在价值，说不定就会获得一份惊喜。[1] 图1-2源自https://mapr.com/blog/real-time-analysis-popular-uber-locations-spark-structured-streaming-machine-learningkafka-and-mapr-db，这篇博客详细讲解了使用Kafka、Spark Streaming 和机器学习等技术实现Uber 交通数据实时分析和展现的方法，感兴趣的读者可自行查阅。[2] 图1-3引用自https://www.linkedin.com/pulse/fraud-detection-fi ntech-ecosystem-oluwaseyi-otun，这篇博客描述了使用Flink 流计算平台实现实时欺诈检测的方法，对实时风控系统感兴趣的读者可自行查阅。[3] 图1-4引用自https://www.talend.com/blog/2016/11/01/setting-up-an-apache-spark-powered-recommendation-engine，这篇博客介绍了一种基于Spark Streaming 流计算平台的实时零售推荐系统，感兴趣的读者请自行查阅。1.3 实时流数据的特点
1.3节介绍了实时流计算技术的使用场景。实时流计算技术的处理对象是实时流数据。尽管实时流数据的来源千变万化、丰富多彩，但归纳起来，实时流数据通常具有实时性、随机性、无序性和无限性。1.实时性
之所以要采集实时流数据，并对其进行实时处理，是因为这些数据具有实时价值。例如，提前预警避免火灾，贷前反欺诈避免骗贷，量化交易抢得市场先机等。如果事后再分析这些数据，这个时候火灾已经发生，骗子已经卷款而逃，市场机会已经错过，分析数据带来的价值也只限于“前事不忘，后事之师”了。因此，对实时流数据的计算和分析一定要在其实时价值消退之前完成，这就要求计算的时延必须小。有时候数据量大、计算复杂的原因会导致实时计算无法完成，这时甚至会牺牲结果的准确性，在保证误差在可接受范围的前提下，优先满足计算的实时性。2.随机性
流数据是真实世界发生各种事件的体现。真实世界事件的随机发生，使得流数据的产生在时间和数量上具有随机性。有时候在很长一段时间内只产生少量数据，有时候又会在很短时间内产生大量数据。实时流数据的随机性对实时流计算系统在各种流量和突发情况下的处理能力与服务稳定性提出要求。我们可以从数据采样、数据缓冲、计算资源动态调整3个角度来解决实时流数据随机性的问题。有些情况下，流数据量很大，暂时超过了系统的处理能力，如果业务需求允许，则可以考虑丢弃部分数据，或者使用带采样性质的算法，减少计算压力。如果数据不允许丢失，则可以采用带缓冲和持久化能力的消息中间件来暂时缓冲数据，让系统平稳处理数据流，削平流量高峰。另外，在一些资源敏感的情况下，可能还需要实时流计算系统能够根据流量压力情况，动态增加或减少计算资源，使得在满足实时流计算的同时，最大化计算资源的使用效率。3.无序性
流数据是一个关于时间的事件序列。我们通常希望事件会按照它们发生的时刻依次到达系统，但由于异步、并发、网络延时、时间不同步和系统故障等诸多原因，严格意义上的全局有序是很难保证的，甚至几乎不可能。于是退而求其次，我们可以让数据在局部时间窗口内有序。在目前主流的实时流计算框架中，常见的做法是将接收到的事件，按时间戳分发到一个个的时间窗口分片中，在等待一段时候后，再触发时间窗口分片内数据的统一处理操作。流数据中的时间有两类：事件发生时间和事件处理时间。事件发生时间是指事件发生的时刻，而事件处理时间则是系统处理事件的时刻。这两种时间会导致流计算的过程和结果都有所不同，具体使用哪种时间因场景而异。4.无限性
流数据是一种随时间无限增长的数据序列。这是流数据和批数据最本质的区别。批数据在每次处理时数据量是有限的，而流数据没有“每次”的概念，它总在不断产生，无穷无尽。流数据和批数据的区别，导致它们在系统架构和算法实现上都有所不同。
在系统架构上，实时流数据的无限性要求系统必须具备高可用性和实时处理能力。一方面，当系统发生故障时，如果系统没有高可用性，则流数据会丢失，并会暂停流计算。这与实时流计算的目标（即在实时流数据上获取实时价值）是相违背的，因此不可容忍。另一方面，当系统处理能力不能跟上数据流产生的速度时，待处理的消息会越积越多。当积压数量超过阈值后，具有有限存储空间的系统必然会崩溃。为了消除已经存在的积压消息，系统处理能力必须超过数据流产生的速度，否则积压情况会一直存在。
在算法实现上，实时流数据的无限性对原本针对批数据设计的算法提出挑战。一方面，实时流计算过程中的可用空间和可用时间都有更严苛的限制；另一方面，流计算的输入数据随时间无限增加，这和批处理算法的输入是有限数据集有本质区别。因此，实时流计算使用的算法相比批处理算法，在算法实现和算法复杂度方面会有明显不同。
在实时流数据的四大特点中，无限性是流数据相比批数据最大的区别，这直接导致了流处理和批处理的查询模式有所不同。批处理是在固定数据集上进行不同的查询，而流处理是在无限数据集上进行固定的查询。实时性、随机性和无序性既是实时流计算系统的特点，也是我们要解决的问题。在1.4节中，我们将针对这些问题来分析实时流计算系统的架构特点。1.4 实时流计算系统架构
1.2节介绍了实时流计算系统的多种使用场景。仔细分析这些系统的组成，我们不难发现，虽然使用场景多种多样、不尽相同，但这些系统都包含了5个部分：数据采集、数据传输、数据处理、数据存储和数据展示。事实上，也正是这5个部分构成了一般通用的实时流计算系统，如图1-5所示。图1-5 实时流计算系统的组成部分1.4.1 数据采集
数据采集是接收来自于各种数据源的数据，并将这些数据经过初步的提取和转换后，发送到数据传输系统的过程。为了使数据接收的性能最优，在设计数据采集方案时，必须充分考虑所接收数据的特点。例如：
·数据接收的性能要求如何？
·数据是逐条发送还是批次发送？
·客户端到服务器的连接是长连接还是短连接？
·最大并发连接数是多少？
表1-1列举了部分数据特点及相应处理方案。表1-1 部分数据特点及相应处理方案
大多数情况下，数据采集服务器选择诸如Netty的非阻塞I/O方案会更加合适。数据被接收后，一般还需要对其做简单的处理，主要是一些字段提取和转化操作，最终将数据表示为统一的数据格式，如JSON、AVRO、Protobuf等。通常而言，数据组织的结构越简单越好，平坦的数据结构比嵌套式数据结构更好，嵌套浅的数据结构比嵌套深数据结构更好。最后，将整理好的数据序列化发往数据传输系统。1.4.2 数据传输
数据传输是流数据在各个模块间流转的过程。数据传输系统的核心是消息中间件，常用的消息队列中间件有Apache Kafka、RabbitMQ等。数据传输系统就像人体的血管系统，承载着整个实时流计算系统的数传输工作。选择消息中间件时，需要考虑以下因素。
1）吞吐量：消息中间件每秒能够处理的消息数。消息中间件自身的吞吐量决定了实时流计算系统吞吐量的上限，所以选择消息中间件时，首先要确定消息中间件本身的吞吐量对业务没有明显的限制。
2）延迟：消息从发送端到消费端所消耗的时间。如同吞吐量一样，消息中间件自身的延迟决定了实时流计算系统延迟的下限。选择消息中间件时，需确定消息中间件本身延迟对业务没有明显限制。
3）高可用：消息中间件的一个或多个节点发生故障时，仍然能够持续提供正常服务。高可用消息中间件必须支持在转移故障并恢复服务后，客户端能自动重新连接并使用服务。千万不能让客户端进入僵死状态，否则即便消息中间件依然在提供服务，而上层的业务服务已然停止。
4）持久化：消息中间件中的消息写入日志或文件，在重启后消息不丢失。大部分业务场景下，支持持久化是一个可靠线上系统的必要条件。数据持久化从高可用角度看，还需要提供支持数据多副本存储功能。当一部分副本数据所在节点出现故障，或这部分副本数据本身被破坏时，可以通过剩余部分的副本数据恢复出来。
5）水平扩展：消息中间件的处理能力能够通过增加节点来提升。当业务量逐渐增加时，原先的消息中间件处理能力逐渐跟不上，这时需要增加新节点以提升消息中间件的处理能力。例如，Kafka可以通过增加Kafka节点和topic分区数的方式水平扩展处理能力。1.4.3 数据处理
数据处理是实时流计算系统的核心。从数据传输系统读取数据流后，需要对数据流做处理。数据处理的目标可以分为4类：数据转化、指标统计、模式匹配以及模型学习和预测。
·数据转化包括数据抽取、清洗、转换和加载，如常见的流式函数filter和map，分别用于完成数据抽取和转化的操作。
·指标统计是在流数据上统计各种指标，如计数、求和、均值、标准差、极值、聚合、关联、直方图等。
·模式匹配是在流数据上寻找预先设定的事件序列模式，我们常说的CEP（复杂事件处理）就属于模式匹配。
·模型学习和预测是数据挖掘和机器学习在流数据上的扩展应用，基于流的模型学习算法可以实时动态地训练或更新模型参数，进而根据模型做出预测，更加准确地描述数据背后当时正在发生的事情。
我们通常使用DAG（Directed Acyclic Graph，有向无环图）来描述流计算过程。常见的开源流处理框架有Apache Storm、Apache Spark、Apache Flink、Apache Samza 和Akka Streams等。在这些流处理框架中，都会使用DAG或类似的概念来表示流计算应用。1.4.4 数据存储
数据存储方案的选型要充分考虑计算类型和查询目标。由于实时流数据的无限性和实时性特点，针对流处理的算法经常需要专门设计。
例如，针对“过去一天同一设备上登录的不同用户数”这种查询目标，在数据量较小时，传统关系型数据库（RBDM）和结构化查询语言（SQL）是不错选择。但当数据量变得很大后，基于关系型数据库的方案会变得越来越吃力，直到最后根本不可能在实时级别的时延内完成查询。
相同的查询目标，采用NoSQL数据库不仅能够做到实时查询，而且能获得更高的吞吐能力。相比传统SQL数据库，实时流计算中会更多地使用NoSQL数据库。越来越多的NoSQL数据库开始提供类似于SQL的查询语言，但查询语言不是数据库的本质所在，数据库的本质是底层的查询执行和数据存储。选择数据存储方案时，上层查询语言的通用性和易用性是重要的考虑因素。但更重要的是，所选数据库的查询和存储本身能够贴合所要进行查询的计算复杂度。
除了在实时流计算过程中需要使用数据库外，数据本身和计算结果通常需要保存起来，以做数据备份、离线报表或离线分析等。离线数据存储一般选择诸如HDFS或S3这样的分布式文件系统。特别是如今Hadoop已经非常成熟，构建在其上的查询和分析工具多种多样，如MapReduce、Hive和Spark等都是非常好的分析工具。这些工具统一在Hadoop生态体系内，为以后的工具选择留下很大的余地。
如果需要针对实时流计算结果构建实时点查询服务，即根据一个或多个键来查询一条特定的实时流计算结果记录，则可以选择NoSQL数据库并配置缓存的方案。
有时候实时流计算的结果使用UI呈现。很多UI会提供交互式查询体验，这就涉及Ad-Hoc查询。设计用于Ad-Hoc查询的存储方案时，一定要考虑UI可能的需求变化，而不能选择一个“僵硬”的数据存储方案，否则当未来UI需求变化、各种查询条件调整时，后端数据库的变更将是一个巨大而且痛苦的挑战。这种情况下，使用搜索引擎一类的存储方案（如ElasticSearch）会是一个明智的选择。
综合而言，在相对复杂的业务场景下，实时流计算只是系统中的一个环节。针对不同计算类型和查询目的，要合理选择相应的数据存储方案。更有甚者，很多时候必须将相同内容的数据根据不同的需求，同时存入多种不同功能的存储方案中。至少目前为止，还没有一种称之为“银弹”的数据库。在本书第9章中，我们将详细讨论各种数据存储方案。1.4.5 数据展示
数据展示是将数据呈现给最终用户的过程。数据呈现的形式可以是API，也可以是UI。API的方式通常以REST服务形式提供。大多数UI是以We.UI的方式实现的，在移动终端大行其道的今天，诸如手机的客户端应用程序也是常用的数据呈现方式。对于We.UI而言，基于Web的数据展示方式有很多优点。一方面，Web服务实现和部署都非常简单，只需提供Web服务器就可以在浏览器中进行访问了。另一方面，各种丰富的前端框架和数据可视化框架为开发提供了更多的便利和选择，如前端常用的框架有React、Vue、Angular等，常用的数据可视化框架有ECharts、D3.js等。
数据可视化是数据展示的核心所在，数据可视化的内容也很丰富、精彩。本书会讨论如何为数据展示选择最合适的数据存储方案。但因为数据可视化部分更加偏向于前端（包括JS、CSS、HTML和UI设计等），这与实时流计算的主体并无太强关联，所以除了部分涉及针对数据展示该如何设计数据存储方案的内容以外，本书不会再用专门的章节讨论数据展示的有关内容。感兴趣的读者可以自行参考前端和数据可视化的有关资料和书籍。1.5 本章小结
整体而言，本书内容按照“总分”的结构组织。阅读本章，我们对实时流计算系统的使用场景和通用架构组成有了一个整体的了解。在后续的章节中，我们将对实时流计算系统的各个部分进行具体的分析和讨论。第2章 数据采集
从本章开始，我们将逐一讨论实时流计算系统各方面的内容。为了更加方便和清楚地阐述问题，本书将以互联网金融风控为场景，构建一个实时流计算风控系统。虽然是以互联网金融风控为场景，但大多数情形下实时流计算系统在架构上是大同小异的，或者具有异曲同工之妙。所以，本书在互联网金融风控场景下讨论的有关实时流计算系统的各种概念、问题和解决方法也能推广应用到其他使用场景。
常言道“巧妇难为无米之炊”，没有数据，我们就没有了讨论的基础。大多数情况下，数据采集是我们构建实时流计算系统的起点，所以本书将首先从数据采集讲起。事实上，我们不能小瞧数据采集的过程。数据采集通常涉及对外提供服务，涵盖许多I/O、网络、异步和并发的技术，在性能、可靠和安全等方面都不容大意。
本章将讨论实时流计算系统的数据采集部分，不过我们会将重心放在讲解有关BIO和NIO、同步和异步、异步和流之间的关联关系等内容。这些内容不仅有助于我们在实际生产中构建高性能数据采集服务器，而且有助于我们加深对异步和高并发编程的理解，并为后续章节对“流”的讨论和理解打下坚实基础。2.1 设计数据采集的接口
在本书中，我们以互联网金融风控场景来展开对实时流计算系统的讨论。在金融风控场景下，分析的风险总体上可以分成两类：一类是贷款对象信用风险，另一类是贷款对象欺诈风险。两类风险的风控因素和模型不同。
贷款对象信用风险关注的是贷款对象自身的信用状况、还款意愿和还款能力。信用风险评估常用的分析因素有四要素认证和征信报告，使用的风控模型主要是可解释性强的逻辑回归评分卡。
贷款对象欺诈风险关注的则是贷款对象是不是在骗贷。在欺诈情形下，贷款对象提供的所有征信信息可能都是正常的，但是这些信息是通过伪造或“黑产”渠道得来的，贷款对象以大量具有良好信用的不同身份获得贷款后，欺诈成功，卷款而逃。欺诈风险评估使用的分析因素多种多样。例如，网络因素（如IP是否集中），用户属性因素（如年龄和职业），用户行为因素（如是否在某个时间段集中贷款），社会信用因素（如社保缴纳情况），第三方征信（如芝麻信用得分），还有各种渠道而来的黑名单等。总体而言，欺诈风险评估使用的因素来源更多，使用的模型也更加多样，如决策树、聚类分析等。
互联网金融风控的一般流程如下所述。用户在手机或网页等客户端发出注册、贷款申请等事件时，客户端将用户属性、行为、生物识别、终端设备信息、网络状况、TCP/IP协议栈等信息发送到数据采集服务器；数据采集服务器收到数据后，进行字段提取和转化，发送给特征提取模块；特征提取模块按照预先设定的特征清单进行特征提取，然后以提取出来的特征清单作为模型或规则系统的输入；最终依据模型或规则系统的评估结果做出决策。
根据上面描述的业务流程，完整的互联网金融风控系统架构设计如图2-1所示。图2-1 完整的互联网金融风控系统架构设计
从手机或网页等客户端，通过互联网发送事件到采集服务器，是金融风控场景下常用的数据采集方式之一。客户端发送的事件包含用户属性、行为、生物识别、终端设备信息、网络状况、TCP/IP协议栈等信息。HTTP/HTTPS协议筑造了整个互联网的基石，也是当前最主要的应用层通信协议。没有特别必要，我们采用HTTP/HTTPS协议来进行客户端和数据采集服务器之间的数据通信。
确定数据通信协议后，还需要制定事件上报API。以REST风格为代表的API设计方式提供了相对标准的API设计准则。依照REST风格，设计事件上报API如下。POST event/{ "user_id": "u200710918", "client_timestamp": "1524646221000", "event_type": "loan", "amount": 1000, "……": "……"}
上面的RES.API表示向服务器上报一个事件，其中：用户账号“user_id”是“u200710918”，发送时间戳“client_timestamp”是“1524646221000”，事件类型“event_type”是“loan”，金额“amount”是“1000”，其他信息用“……”表示。
至此通信协议和API都确定了，接下来实现采集服务器。2.2 使用Sprin Boot实现数据采集服务器
说到REST风格Web服务器开发，大部分Java编程开发者首先想到的是Spring系列中的Spring Boot。毫无疑问，Spring Boot使得用Java做Web服务开发的体验相比过去有了极大的提升。几乎在数分钟之内，一个可用的Web服务就可以开发完毕。所以，我们也用Spring Boot来实现数据采集服务器，具体实现如下。@Controller@EnableAutoConfigurationpublic class SpringDataCollector { private static final Logger logger = LoggerFactory.getLogger(SpringDataCollector.class); private JSONObject doExtractCleanTransform(JSONObject event) { // TODO: 实现抽取、清洗、转化的具体逻辑 return event; } private final String kafkaBroker = "127.0.0.1:9092"; private final String topic = "collector_event"; private final KafkaSender kafkaSender = new KafkaSender(kafkaBroker); @PostMapping(path = "/event", produces = MediaType.APPLICATION_JSON_UTF8_VALUE) @ResponseBody() public String uploadEvent(@RequestBody byte[] body) { // step1: 对消息进行解码 JSONObject bodyJson = JSONObject.parseObject(new String(body, Charsets.UTF_8)); // step2: 对消息进行抽取、清洗、转化 JSONObject normEvent = doExtractCleanTransform(bodyJson); // step3: 将格式规整化的消息发到消息中间件Kafka kafkaSender.send(topic, normEvent.toJSONString().getBytes(Charsets.UTF_8)); // 通知客户端数据采集成功 return RestHelper.genResponse(200, "ok").toJSONString(); } public static void main(String[] args) throws Exception { SpringApplication.run(SpringDataCollector.class, args); }}注意：为了节省篇幅，本书中的样例代码均只保留了主要逻辑以阐述问题，大部分略去了异常处理和日志打印。如需将这些代码用于真实产品环境，则需要读者自行添加异常处理和日志打印相关内容。异常处理和日志打印是可靠软件的重要因素，在编程开发时务必重视这两点。
在上面的示例代码中，uploadEvent实现了事件上报接口。收到上报事件后，首先对数据进行解码，解码结果用FastJson中的通用JSON类JSONObject表示；然后在JSONObject对象基础上进行抽取、清洗和转化，规整为统一格式数据；最后将规整好的数据发往数据传输系统Kafka。这个程序在实现功能上并没有特别的地方，我们只是感觉到基于Spring Boot的服务开发体验是如此轻松、愉快。2.3 BIO与NIO
我们使用Spring Boot非常迅速地开发好了数据采集服务器，之后的测试和上线工作也一帆风顺。客户开始接入流量，服务运转良好，似乎一切都预示着程序员的工作就是这样轻松、美好。但好景不长，随着业务流量的增加，晴朗天空不知不觉飘来两朵“乌云”。
·随着用户越来越多，采集服务器连接数逐渐增加，甚至在高峰时出现成千上万并发连接的情况。每个连接的服务质量急剧下降，不时返回408或503错误。
·监控显示，客户请求响应的时延非常大，进一步分析发现是doExtractCleanTransform函数比较耗时。这个函数耗时的原因可能是计算比较复杂，也可能是有较多的I/O操作，还可能是有较多的外部请求调用。数据采集服务器的性能表现很差，但是看系统监控又发现CPU和I/O的使用效率并不高，似乎它们都在“偷懒”不干活。
基本上，当我们初次开始认真关注程序的性能问题时，都会碰到上面的问题。根据笔者经验，如果此时能够深入地钻研下去，我们将从此掌握编写高性能程序的高级技能点，将在以后的程序开发过程中受益良多。
我们先看采集服务器的连接问题。当使用Spring Boot做Web服务开发时，默认情况下Spring Boot使用Tomcat容器。早期版本的Tomcat默认使用BIO连接器。虽然现在的版本已经去掉BIO连接器，并默认采用NIO连接器，但是我们还是来比较下BIO和NIO连接器的区别，这样对理解BIO和NIO、同步和异步的原理，以及编写高性能程序都有很大的帮助。2.3.1 BIO连接器
在Java中，最基础的I/O方式是BIO（Blocking I/O，阻塞式I/O）。BIO是一种同步并且阻塞式的I/O方式。图2-2描述了BIO连接器的工作原理，当接收器（acceptor）线程接收到新的请求连接套接字（socket）时，从工作线程栈（worker stack）中取出一个空闲的工作线程（worker），用于处理新接收的连接套接字。如果工作线程栈没有空闲工作线程，且创建的工作线程数量没有达到设置的上限值，则新建一个工作线程用于处理连接套接字。而如果工作线程栈没有空闲工作线程，且创建的工作线程数量已达到设置上限值，则接收器被阻塞，它将暂停接收新的连接请求。只有当某个工作线程处理完其对应的请求后，它会被重新放入工作线程栈，成为空闲线程之后，接收器才能继续接收新的请求，并交由工作线程栈中的空闲工作线程处理。工作线程从连接套接字中读取请求数据并进行处理，处理完成后再将结果通过连接套接字发送回客户端。图2-2 BIO连接器的工作原理
在请求连接数比较小、请求处理逻辑比较简单、工作线程请求处理时延很短的场景下，使用BIO连接器是很合适的。但很显然，在实际工作中的大多数场景下，这些前提条件都是可遇而不可求的。就如在互联网金融风控系统中，上报数据的客户端是分布在全世界各地的成千上万，甚至数十万、数百万的手机、平板和个人电脑，这些终端平均下来每秒发送到数据采集服务器的请求少则数千，多则上万。
再考虑工作线程处理较慢的情况，如计算逻辑较复杂或外部I/O较多。当所有工作线程都在工作时，可用工作线程耗尽，这时请求接收器将阻塞，等待工作线程可用，而不能接收新的请求套接字。当工作线程处理完请求后，由于没有立即可用的新请求需要处理，它必须等到请求接收器接收新的请求之后，才能继续工作。经过以上分析就会发现，这种处理方案的性能比较低下。一方面请求接收线程和工作线程都很忙碌，另一方面请求接收线程和工作线程却要时不时地相互等待，这就导致请求接收器和工作线程时不时处于空闲状态。进一步深入到操作系统层面，表现在CPU和网络I/O很多时候处于空闲状态。操作系统资源大量空闲，造成资源浪费，性能却还十分低下。很显然，这是我们不能接受的情况，必须对其做出改进和提升。
为了在使用BIO连接器时提高资源的使用效率，一种行之有效的方法是增加工作线程数量。理想情况下，如果有成千上万甚至上百万个工作线程来处理连接套接字，那么请求接收器不用担心工作线程不够用，因为任何时候总会有工作线程可用。这样，数据采集服务器的并发连接数也能够达到成千上万。当然，如果要支持百万并发连接，还需要专门配置一些操作系统参数，这里不做详细讨论，感兴趣的读者可以自行搜索相关资料。
当前大多数操作系统在处理上万个甚至只需几千个线程时，性能就会明显下降。这是因为，当需要调度的线程变得非常多后，操作系统在进行线程调度和上下文切换时，需要占用大量CPU时间，使得真正用于有效任务计算的CPU时间变少。以Linux操作系统为例，在现代处理器上一次线程上下文切换的典型时延为数微秒（microsecond）。如果以5微秒来计算，则全部1万个线程各做一次上下文切换就要占用50毫秒，这个时延已相当明显。除了线程切换的时间显著增加外，由于每个线程拥有自己独立的线程栈，过多的线程还会占用大量内存，这也是一个主要的资源消耗和性能损耗因素。虽然启用过多线程会对CPU资源和内存资源造成浪费，但是充足的线程还是有一定好处的，毕竟足够多的线程能够同时触发足够多的I/O任务，从而使I/O资源使用得更加充分。
Linux操作系统线程调度原理如图2-3所示。我们在开发多线程应用时常说的线程，在Linux操作系统中实际上被实现为轻量级进程。而每个轻量级进程以1︰1的关系对应一个内核线程。所有内核线程会根据其运行已消耗CPU时间、线程所处状态及线程优先级等多种因素被调度器不停轮流调度执行。通常而言，当有数千个线程时，调度器尚可以高效处理；但当有数十万、数百万线程时，调度器就会“累趴下”了。图2-3 Linux操作系统线程调度原理
既然不能在一台机器上运行太多线程，我们很自然地想到可以用多台机器来分担计算任务。不错，这是一个很好的办法。在多个对等的服务节点之前，架设一个负载均衡器（如Nginx），可以有效地将请求分发到多台服务器，这既可以提高服务整体的吞吐能力，也能在一定程度上降低因为请求积压造成的服务响应时延。但除非是线上情况紧急，需要立刻提升服务处理能力以应对突发的流量高峰冲击，否则我们不应该立刻这样做！作为有极客精神的程序员，同时为了降低成本着想，在将一台机器的资源充分利用前，我们不能简单地寄希望于通过横向增加机器数量来提高服务的性能。
既不能运行太多线程，也不愿意水平扩展机器数量，那怎样才能提升程序的性能呢？我们不妨这样思考，接收器无阻塞地接收连接套接字，并将新接收的连接套接字暂存到一个缓冲区。当工作线程在处理完一个连接套接字后，从缓冲区取出暂存的连接套接字进行处理。如此一来，接收器可以不停地接收新的连接套接字，而工作线程的任务也被安排得满满当当。
因此，BIO连接器的本质缺陷是接收器和工作线程执行步调耦合太紧。如果将接收器和工作线程通过缓冲区隔离开来，让它们互不干扰地独立运行，那么接收器和工作线程的工作效率都会得到提高，进而提升程序性能。图2-4展示了改进BIO的方法，在接收器接收到新的连接套接字时，不再需要获取一个处于空闲状态的工作线程，而是只需将其放入连接套接字队列即可。而工作线程则完全不需要理会接收器在做什么，它只需要看队列有没有待处理的连接套接字即可：如果有，就将连接套接字取出来处理；如果没有，说明暂时没有请求，它可以休息一会儿了。接下来我们将看到，Tomcat的NIO连接器正是按照类似的思路做的。图2-4 改进BIO的方法2.3.2 NIO连接器
在编写本书时，最新版本的Tomcat已经将NIO作为默认连接器。图2-5描述了NIO连接器的工作原理，当接收器接收新的连接套接字时，先将其依次封装成NioChannel对象和PollerEvent对象，再将PollerEvent对象放入PollerEvent队列。与此同时，轮询器不断从其PollerEvent队列中取出新的PollerEvent对象，获得代表连接套接字的NioChannel，再将其SocketChannel注册到选择器。选择器从注册在其上的SocketChannel中挑选出处于Rea.Ready状态的SocketChannel，再将其交到工作线程池的队列。工作线程池中的各个工作线程从队列中取出连接套接字，并读取请求数据进行处理，在处理完成时再将结果通过连接套接字发送回客户端。图2-5 NIO连接器的工作原理
从NIO连接器的工作过程可以看出，Tomcat的NIO连接器相比BIO连接器，主要做出了两大改进。除了类似于我们在图2-4中提到的使用“队列”将接收器和工作线程隔离开的改进方法之外，Tomcat的NIO连接器还引入选择器（包含在轮询器中）来更加精细地管理连接套接字，也就是说，选择器只有在连接套接字中的数据处于可读（Read Ready）状态时，才将其交由工作线程来处理，避免了工作线程等待连接套接字准备数据的过程。NIO连接器的这两点改进带来了两种好处。
1）接收器和工作线程隔离开，让它们彼此之间不会因为对方阻塞而影响自己的连续运行。这样接收器和工作线程都能尽其所能地工作，从而更加充分地使用I/O和CPU资源。
2）因为有了队列缓存待处理的连接套接字，NIO连接器能够保持的并发连接数也就不再受限于工作线程数量，而只受限于系统设置的上限值（由LimitLatch指定）。这样，无须分配大量线程，数据采集服务器就能支持大量并发连接了。2.4 NIO和异步
晴朗天空上的第一朵“乌云”终于被我们驱散了，但还有另外一朵“乌云”在悠悠然地飘着，它仿佛正眯着眼俯视着我们，幸灾乐祸地等待着发生什么事情。于是我们小心翼翼地查看了在线监控系统。不看不知道，一看吓一跳。我们注意到，虽然并发处理的连接数增加了，但是请求的平均响应时间依然很高，数据采集服务器的吞吐能力还是很低，这可与我们的预想相差甚远！于是，我们进一步使用JVisualVM（参见3.4.2节）这个“神器”连接到运行着的数据采集服务器，希望能够找到造成程序性能依旧低下的“元凶”。在对JVM的运行时状态进行采样后，我们立刻发现原来是doExtractCleanTransform（）函数的执行耗时占用了整个请求处理用时的90%以上，处理时延明显过高！现在我们就来仔细分析下doExtractCleanTransform（）可能耗时的原因。
试读结束[说明：试读内容隐藏了图片]