我们很少看到有人公开谈论自己的错误。

人非圣贤，孰能无过？虽然难言出口，但反思过去所犯的错误可以让人不会在未来——至少是短期的未来，犯同样的错误。

Mohamed Barouma是位工作5年的专业程序员，但和任何人一样，他也犯过错误。

他这么说：“通常情况下，我不会马上意识到自己做错了什么；只有在接触到正确的做事方法后，我才知道犯了哪些错。”

结合他的原文，本文总结了他犯下的七大误区。

没有使用合适的ORM

数据访问层代码总是混乱、乏味和无聊的。不幸的是，往往这点到*后才能发现。

Mohamed和ORM有着一段孽缘。ORM，即Object Relational Mapping，它是对象关系模型的简称。它的作用是在关系型数据库和对象之间作一个映射，使程序能够通过操纵描述对象方式来操纵数据库。

当Mohamed*次做一个简单的内部会计应用程序时，他发现只是为了完成基本的程序，就不得不编写大量的代码。于是他开始埋头于ADO.NET，并手动编写了一个自制的、具有非常特殊的、自定义的表模式的ORM，来满足目的。

在一段时间里，这个ORM工作的相当不错，直到几个月后，公司的业务需求发生了一些变化，这导致了整个表格模式的变化，然后，就是对ORM的反复修改。这个流程的痛苦之*，让Mohamed*终选择了强类型数据集适配器。

虽然这件事因此解决，但如果能找到一个更合适的ORM(比如NHibernate)来完成工作，Mohamed仍会义不容辞，至少当他的用户数量增加时，不必担心更改数据库的供应商。

没有学会使用泛型

Mohamed Barouma的职业程序员生涯始于Net 1.1。而在当时，Net 1.1的主要问题在于它没有泛型支持，这代表它不能有一个强类型列表，只能满足于乏味的ArrayList。但是使用Arraylist在Java代码中进行类型转换和装箱，会导致读写起来十分痛苦。

因此，Net 1.1的程序员们使用CodeSmith生成一个强类型集合列表。

但随着代码库的增长，那些定制生成的列表本身就变成了一个无法收拾的怪物。只要经常为了创建对象或调用方法去达到目的，随后就会因为修改代码导致混乱和错误。

如果切换到Net 2.0，并在它可用时立即开始使用泛型，而不是创建越来越多的难以维护的自定义集合列表，那么一切问题都将迎刃而解。

没有放弃“造轮子”

这是个老生常谈的话题，“反复造轮子”（Reinvent The Wheel）。新程序员总是喜欢反复“造轮子”，自认为当前的实现不够好，所以不得不从头重写整个东西。

为什么叫“造轮子”？就像真正的的轮子早在几千年前就被确定是圆形的一样，很多数据库也早就已经成熟易用了，但还是有数不胜数的程序员们锲而不舍的去“造轮子”，有的人飞蛾扑火、蚍蜉撼树，有的人别具一格、推陈出新，这就是“造轮子”魔法一般的吸引力。

这其中也不乏Mohamed，他想重新编写自己的UI控件，因为Windows Forms UI控件实在是太简单了。*后，他造的GUI工具被商业化成体系的.Net UI控件轻松打败，又一辆新生程序员造的“轮子”被击沉到了代码海洋里。

没有精简过多的文档

很多刚入行的程序员，会在一开始觉得代码文档很好，因为它用简单的英语注释了代码在做什么。但事实上这些文档通常在修改了几次代码之后变成了一摊废纸，变得陈旧、过时亦或是完全错误。

常常有人花了很多时间编写代码文档——比如XML文档，结果发现在更改代码时需要更新文档。因为它的功能可能都已经改变了。更新代码是必须的，但更新XML文档不是必须的：这是一种负担，它消耗时间，而且毫无用处。

*终，反复地更改XML文档使人逐渐失去耐心，转而做其他事情。

没有使用自动化构建

应用程序的部署和打包比编程相对容易，所以它往往被放在了非常低的优先级上。但很快，粗制滥造的构建就会因为无法工作，受到各式各样的投诉：

“先决条件缺失，该如何修复？”

“dll没有更新，你能给我一个补丁吗？”

“我的图标怎么不见了？”

紧接着，电话像雪崩一样源源不断地打到桌旁。这是Mohamed的真实经历，并让他那天精疲力尽——不是因为编程，而是因为令人麻木的重新部署和重新包装过程。

而这一切，本可以通过编写自动化脚本节省一些时间，否则在事后debug浪费的时间*对比可以节省的时间多上数倍。应该让软件可以一键构建，否则再多都是一种浪费。

没有停止对视觉检测和debug的依赖

Visual Studio让人们可以很容易地调试代码并进行动态检查，这也使得创建一个表单并显示输出非常简单。但如果太沉迷于调试器，这项好处就要变成坏处了。

为什么呢？想象一下，如果一个方法只在应用程序启动并运行45分钟后才被调用，那难道要打算等45分钟再开始调试吗?

所以，动动手将应用程序分解为可以独立调用的子模块，这样就可以准备产生错误输出的输入值，并从那里开始测试它。

没有做单元测试

不少程序员可能这么想过：“我的这个应用程序微不足道，它可以很容易地被手工测试覆盖；单元测试是针对大型和复杂的东西，而不是针对我的程序。”

可想而知，这会直接亲手创造一个没有关注点分离，难以重构，完全不可维护的代码库。

“蹑手蹑脚”几乎是许多小白程序员的通病，害怕对代码进行哪怕是*轻微的修改，因为任何更改都可能导致或不会导致破坏性的更改。结果到*后一发不可收拾，出现的问题无法解决。使用这种遗留代码不仅仅是无聊和紧张，而且精神上也有压力。

但是使用单元测试，能让代码的寿命大大提高。Mohamed希望自己能学会单元测试的“艺术”，从入学*天开始练习单元测试，可惜学校并不教这个。

世界上，无数令人为之一振的创新发明都源自无数次的试错，但即便如此，避免基础性的错误依旧是很有必要的。在你的程序人生中，还遇到过哪些令人啼笑皆非的“常见误区”？亦或者是创造了一些百思不得其解的“致命误区”？欢迎下方留言，分享你学习编程时的心得体验。

参考文献：

https://betterprogramming.pub/7-big-mistakes-i-have-made-in-my-career-as-a-software-engineer-f14ef540be10

哪个国家的程序员水平*高？

软件工程师水平测定平台HackRank发起了一系列编程挑战竞赛，给出了一份出乎意料的数据。据悉，这份数据研究分析了HackerRank社区里30万名开发者完成的140万项编码挑战。*终的结果显示中国程序员排在了*位，紧随其后的是俄罗斯、波兰、瑞士和匈牙利，而美国只排在第28名。

由此可见，随着计算机和互联网的发展和普及，中国对程序员的需求和培养力度都在逐渐加大。在技术迅猛发展的今天，物联网、数据分析、以及人工智能这三大技术之间的合作将会在世界上创造出一个巨大的智能机器网络，市场对程序员的需求将会越来越大，中国对编程人员的重视程度也将越来越高。

那么一个合格的程序员应该具备哪些能力呢？很多在校学生以及初入行业的新手，在刚开始的时候都会陷入程序员应该多刷题、多写代码的误区，然而在人才供需市场上，合格的工程师更需要具备的能力是工程能力、交付能力，而不是只会写代码的能力。

工程能力就是在一个团队中将项目做好的能力。以敏捷开发的思维，按照一定的流程、规范和方法论，在单元测试和自动化测试的基础上，完成一个项目的初版，并在初版的基础上做到快速响应需求和迭代开发。

在日常的工作中，我们需要将项目做好做完善，线上运行稳定、没有bug，能快速响应产品的需求并迭代开发，这就十分需要我们在具备良好的技术能力的同时，也具备*佳的工程能力。

交付能力是指你将自己大脑里的想法完整传递给另一人的能力。很多人会说：我说的话，你怎么听不懂？尤其是在工作中跟同事对接时，你说了半天之后，对方说自己理解了，但是做起来却发现对方完全理解错了方向。这其中一个很重要的原因可能就是自己的交付能力不过关。你说的东西，对方并没有get到重点。想要锻炼好自己的交付能力就需要站在对方的角度考虑问题。比如，对方是否是外行？对方究竟想了解什么？对方目前知道多少？你们需要解决什么问题？

工程能力和交付能力在日常工作中的重要性不言而喻，但是我们在日常的学习生活中很难有机会能够锻炼这两项能力。为此，2021年3月1日，CSDN正式推出“软件工程师能力认证标准”，帮助开发者高效学习与成长。C系列认证清晰定义软件工程师能力模型，面向开发者、技术爱好者、在校大学生等群体，通过机试（真人露脸、全程录屏、限时提交）测出应试者的真能力，筛选合格软件人才，建立应聘者与企业之间的信任关系。

与社会上其他认证考试不同，C认证的核心是检验考生的工程化/交付能力。

通过C认证的Java/Python/前端/全栈工程师等方向的相关考试，意味着考生具备了商业化项目的实施能力，企业可以放心选择，并节约大量的培养成本。

在C认证的考试过程，考生将全程真人露脸、全程录屏、限时提交，考试结果是考生能力的真实体现，同时考试过程视频与代码会保存在C认证考试中心的服务器端，形成“立体能力简历”，有效提升考生的应聘成功率。

除此之外，在C认证中的项目试题都是由优秀导师精心挑选的名企热门核心项目。整个项目由大厂CTO级导师设计，完全贴合大厂的项目标准，让你在备考时就能深入接触大企业核心项目。

雾计算是一种分布式计算结构。将数据和应用程序等资源放置在数据源和云之间的逻辑位置。雾计算的优点之一是让许多用户同时连接到互联网上。从本质上说，它提供的网络和服务与基于云的解决方案提供的相同，但增加了分散式网络的安全性。

云计算和雾计算的区别

云计算

云计算是提供地理分布的计算和存储资源。计算可以在各种平台上进行，包括公共云和私有云。

云计算平台通过可伸缩系统在用户之间共享和均衡工作负载。从本质上讲，云计算具有从非站点位置存储和恢复数据的能力。

云计算是传统手机变得“智能”的主要原因之一。手机没有足够的内置空间来存储和访问应用程序和服务所需的数据。所有的数据都传输到云上，以提供我们需要的服务。尽管如此，云计算技术仍然面临着一个挑战——带宽限制。

雾计算

在不久的将来，雾计算将主宰整个行业。雾的主导将受更接近用户设备的数据源所驱动，设备将在本地而不在云中执行处理。不过，设备也面临物理上的限制——低功耗和小尺寸。

本地处理数据的能力显得更为重要，雾计算增加了数据的安全性。随着物联网的发展，越来越多的设备被添加到网络中。每个设备都通过无线连接进行数据传输和接收。

雾计算能够有效地存储和访问数据。雾计算将分散在网络中的边缘计算节点组网，使其虽然在地理上分布，但节点之间仍能有组织的通信。

使用雾计算涉及到互连边缘设备的复杂过程。边缘设备包括传感器、存储系统和网络基础设施，它们协同工作以捕获和分发数据。

然而，雾计算的灵活性，以及它从集中的云和网络的边缘设备收集和处理数据的能力，使其成为处理我们今天所面临的信息过载的*有用的方法之一。

图片来源：nikhomk panumas;pexels

雾计算和边缘计算是一回事吗?

雾计算有时也被称为“边缘计算”。边缘计算的目的是通过将数据存储在离“地面”更近的地方来解决问题。换句话说，边缘将数据存储在存储设备和本地计算机中，而不是通过云端运行所有数据。

从本质上说，雾计算支持快速响应时间，减少网络延迟和流量，并节约骨干带宽，以实现更好的服务质量(QoS)。它还支持将相关数据传输到云端。

IDC估计，到2025年底，全球大约有45%的数据将被移向网络边缘。雾计算被认为是未来几年唯一能够抵御人工智能、5G和物联网的技术。

IDC的另一项研究预测，2020年边缘设备将产生全球10%的数据。边缘设备将满足雾计算更有效解决方案的需求，从而减少延迟。

边缘计算

边缘计算基本上是雾计算的一个子集。它是在数据出现的地方附近进行处理。雾计算支持更有效的数据处理，从而减少数据延迟的可能性。

可以将雾计算看作是一种处理数据的广义方法，从数据生成的地方到数据存储的地方。边缘计算指的是在数据产生的地方附近对数据进行处理；而雾计算封装了从边缘到末端传输数据所需的边缘处理和网络连接。

随着物联网设备的能力不断增强，如可编程自动化控制器。自动化控制器执行数据处理、通信和其他任务。利用雾计算，数据从端点传输到网关；然后将数据传输到云中进行处理和回传。地理分布的基础设施与云服务保持一致，以便能够以*小的延迟进行数据分析。

雾和边缘计算都有助于更快地将数据转化为可操作的见解，从而使用户能够更快、更明智地做出决策。雾和边缘允许公司更有效地使用带宽，同时增强安全和解决隐私问题。因为雾节点可以安装在任何有网络连接的地方，因此雾计算在工业物联网应用中越来越受欢迎。

雾计算在物联网中的作用

当一个设备或应用程序产生或收集大量信息时，数据存储就会变得越来越复杂和昂贵。在处理这些数据时，网络带宽也变得昂贵，需要大型数据中心来存储和共享这些信息。

雾计算已经替代了传统处理数据的方法。雾计算收集数据和分配计算、存储网络连接的资源和服务。它显著降低了能源消耗，*小化了空间和时间复杂度，并*大化了数据的效用和性能。

智慧城市

让我们以智慧城市为例。城市中所有物联网设备传输、存储和访问的数据量不断增加，这将需要一种新的基础设施来处理这些数据量。正是这些应用需要雾计算来体现物联网带来的全部价值。

公共事业

智慧城市中的供水、医院、执法、运输和应急管理应用需要*新的数据和技术来提供信息和服务，以支持其高效运作。

关于水泄漏、碳排放、公共设施损坏等信息可以用于政府监管、改进操作、挽救生命和提高效率。

雾计算不会从一个地方移动到另一个地方。相反，雾是一种部署物联网网络的方法，可以提供*好的投资回报。

雾计算的优点

雾计算可用于处理大量数据、网络事务和快速响应的应用程序。雾计算包括实时、混合和自主数据处理中心，这些数据处理中心可以提高运行效率和安全性。此外，雾计算可以确保系统的可用性和*优性，而无需在电力、数据中心安全性和可靠性方面进行投资。

雾计算通过集中多个节点的计算资源来降低开销成本。雾节点的位置是根据它们的可用性、效率和使用情况来选择的。它还减少了企业数据中心的负载。减少数据流量是雾计算的另一个主要优势。

许多公司正在使用雾计算来部署分布在许多地方的软件应用程序。公司在一个网络上部署许多系统，以实现更高的效率和可达性。

从根本上说，雾计算为企业处理数据提供了更大的灵活性。对于某些应用，数据处理必须尽可能快，例如在制造业，连网设备应该尽快对事故做出反应。

雾计算还能使公司实时了解他们的客户或员工的情况。企业有望利用物联网技术抓住新的机遇，增加利润。随着雾计算的实施，这项技术有可能为政府、公司甚至个人用户节省一大笔钱。

结论

随着云技术不断渗透到企业环境中，雾计算的使用也将继续增加。云计算通过弹性计算来分配工作负载，从而实现对云中的数据的实时处理。

边缘计算是物联网雾计算领域的一个重要领域。边缘计算是将计算资源部署在云之外的网络边缘的技术。它允许网络边缘的计算资源被访问、分析。

雾计算解决方案将使公司能够在物联网中实现实时计算。因此，物联网雾计算市场将成为云计算市场的主要贡献者。

从2005年到2019年间，中国数字经济总体规模由2.6万亿元增加至35.8万亿元，数字经济在GDP的占比也由14.2%提升至36.2%。随着数字经济蓬勃发展，数据也成为当之无愧的关键生产要素，是基础性资源和战略性资源。数据洪流的到来进一步驱动了数据价值链的成熟，进而驱动企业的决策和创新，获取数字红利。数据中心作为数据的载体，承担着数据资源集中存储、计算、交换的功能，转型升级迫在眉睫。

4月13日，在2021华为全球分析师大会上，华为数据存储与机器视觉营销工程部总裁庞鑫向全行业解读了华为对存储产业以及全闪存数据中心的理解与洞察。

全闪存市场是刚需，但当前全球全闪存市场占比仍然很低

庞鑫表示，全闪存凭借出色的性能优势和绿色节能已经成为市场刚需。在性能方面，从IOPS、低延时、吞吐量三个常用的存储性能指标来看，闪存SSD在性能上远超机械硬盘HDD。闪存具有高的IOPS，一个SSD的IOPS可达数万次之多，而一个HDD的IOPS只能达到几百次，这表明需要许多HDD才能达到一个SSD的性能。同时闪存还具有低延迟（常以毫秒或微秒为度量单位）性能和大吞吐量。

在耗能方面，数据中心已经成为事实上的耗电大户。目前全球数据中心的耗电大约占据全球每年总用电量的2%-3%，电费在数据中心运营成本中占比超过50%。而在国内，全国数据中心的耗电量已连续八年以超过12%的速度增长，未来占社会总用电量的比例将持续增长。相对于HDD，在相同的容量下，SSD的电力能耗降低70%，空间占用节省50%，不仅能大幅降低数据中心的TCO总成本，并且能有效推进数据中心低碳运转。

虽然闪存在性能及绿色节能上优势明显，但是当前闪存在存储市场的占比仍然不高。数据显示，全球存储市场闪存占比43.7%，在中国市场这一数字仅为24.7%。加速闪存替代机械硬盘已经势在必行。

四大变化，将全面激发全闪存数据中心潜能释放

对于过去十年闪存存储产业的发展，庞鑫将其总结为2010~2015年闪存1.0时代，2015~2020年闪存2.0时代，未来5年，将迈入闪存3.0时代。在闪存1.0时代，仅关注性能，主要通过高性能硬件实现，全闪存作为存储系统的加速层，软件特性非常缺乏，价格昂贵。在闪存2.0时代，闪存开始进入主存储系统，通过系统化设计，提供丰富企业特性，同时通过数据缩减技术不断降低闪存存储成本，全闪存进入快速发展期。进入闪存3.0时代，四大变化将全面激发全闪存数据中心潜能释放，进一步提升端到端性能，并实现绿色节能、安全可靠。

*个变化是全场景闪存普惠。全闪存将从之前只用于核心数据存储，比如金融核心、运营商计费系统，越来越多地走向全行业、全场景，比如超算、大数据场景也逐渐开始使用性能好、耗能低的全闪存存储。

第二个变化是全场景数据保护。随着全闪存应用场景不断演进，数据全生命周期保护出现新的要求，包括保护等级的变化、保护对象的变化、保护性能的变化以及保护规模的变化。保护等级上，以银行为例，资产规模千亿以下时要求建立同城灾备，当总资产规模超过千亿，就需要建立两地三中心灾备。保护对象上，以前是结构化数据保护，现在需要同时保护结构化和非结构化数据。比如医院HIS系统是结构化数据，而PACS影像是非结构化数据，要求保存10年以上，并且至少3年在线，可供快速调阅和诊断使用。保护性能上，灾备的冷数据需要能变成温热数据，发挥数据的价值。保护规模上，数据从PB级增长到EB级。比如L4级自动驾驶数据，每车每天路测数据达60TB，是L2级的50倍。

第三个变化是多元融合管理。从传统的烟囱式建设到全闪存整合，形成PB级资源池，这就需要存储系统具备多元融合管理的能力。

第四个变化是高速网络联接。未来的存储网络需要用NVMe协议来替代SCSI协议，让全闪存*大程度释放性能潜力。

全闪存数据中心四大能力引领未来建设

在庞鑫看来，建设全闪存数据中心，不仅仅是对介质进行升级，同时也是将数据中心资源进行整合，架构进行重构，以应对未来业务多样性的需求。华为通过提供全场景闪存化、全场景数据保护、全生命周期自动化运维管理以及全IP数据中心网络四大能力实现数据中心效率更高、安全等级更高、运维更智能、网络无瓶颈、绿色低成本的目标。

1. 全场景闪存化，绿色节能，迈向碳中和

全场景闪存化是指多类型的数据负载均应通过闪存存储进行业务加速，除了主存的全闪化替代之外，华为建议存储灾备系统也应加大闪存配比，让灾备数据快取快用，实现存储容灾的常态化和备份存储的生产化能力。

在主存上，借助*新的SCM介质，可以对热数据进一步加速，时延低至亚毫秒级；在备份上，通过闪存加速，有效提升备份系统性能，将原来的备份时间由小时级缩短为分钟级，确保备份数据即时可用；在归档上，SSD作为归档缓存提供高访问性能，归档数据访问时间从小时级降低到分钟级。

华为还致力于*数据缩减算法研究，实现有限的空间能存放更多的数据，加速闪存普惠。华为在全球数学圣地法国巴黎和俄罗斯莫斯科，分别建立了算法研究中心，将数据的缩减做到*致。去年华为举办的全球数据压缩大赛，里面有不少算法就开始应用到华为全闪存存储产品上。

2. 全场景数据保护，保护对象更广、业务响应更快、留存时间更久

随着数据量越来越多，数据越来越重要，更大规模和更多类型的数据需要保护，数据灾备速度、恢复速度、副本使用频度、乃至留存时间，都需要大幅的提升。

从保护对象的角度来看，灾备数据的范围变大，从核心到通用到一般，从生产核心扩大到开发测试、分析训练、视频监控，从结构化数据扩大到非结构化数据。这一趋势导致灾备数据量迅速变大，但备份窗口不变或变小。数据灾备的等级需要提升，从本地的双活/主备容灾，到异地3DC，再到灾备上云，进行全方位的数据保护。同时备份数据需要立即可用，不需要数据恢复到生产设备后再使用。在数据留存方面，数据合规通常要求重要数据留存30年以上，同时具备防删改、防病毒，适应较为宽泛的机房环境。

3. 全生命周期自动化运维管理，提升运维效率

全闪存的普及将加速数据资源池化，而数据池化加剧了数据中心运维的难度，华为全闪存数据中心具备全生命周期自动化运维的能力，即在规划、建设、运维、优化的各个阶段，均具备自动化智能化的能力。

在规划阶段从关注设备到关注数据的全生命周期，实现精准规划；在建设阶段实现从人工到自动化，核心是全局资源发放；在运维阶段变被动巡检为主动发现，实现全栈智能运维；在优化阶段变依赖专家经验为智能算法自治，达到资源配置敏捷优化。

4. 全IP数据中心网络，*大化释放闪存性能潜力

全场景闪存化也推动了数据中心的网络改革，NVMe*大化释放了SSD介质的能力，更快的介质和协议呼唤更快的网络。NoF（NVMe over Fabric）存储网络应运而生，通过使用IP网络进行对专用网络的创新性革新，实现了更快吞吐带宽和更低时延，同时也兼具IP易管理优势，是更好地实现端到端NVMe存储网络的*佳方案，NoF方案也是目前业界*主流的趋势。

基于当下业界主流标准NoF方案，华为又依托在网络和存储领域的深厚积累，从可靠性、性能、易用性上对其进行提升，基于智能无损网络面向华为OceanStor存储提出了NoF+方案，将下一代数据中心存储网络推向更广阔的发展空间。

在演讲*后，庞鑫表示，包含了OceanStor全闪存、OceanProtect数据保护、DME全生命周期智能运维、NoF+高速存储网络在内的华为全闪存数据中心解决方案，是打造未来安全高效、绿色节能数据中心行之有效的方法。全闪存数据中心必将在未来大展身手，推动社会经济生产迈向新征程。

2021年初因为Elon Musk“带货”而走红的音频社交App Clubhouse，又以肉眼可见的速度跌落神坛，下载量从2月的960 万/月跌至4月的92万/月。不过在5月，Clubhouse终于推出了安卓版，并表示接下来也会对所有用户开放。

另一边，已经很久没有新噱头的社交巨头 Twitter 和 Facebook因为Clubhouse的爆红，也在跟进音频社交。Clubhouse能否借安卓版翻身，音频社交会不会成为下一个全民级应用，都仍是悬念。

音频社交背后的RTC技术，早在2020年，就因为新冠疫情，被各类场景广泛使用，几乎无处不在。例如疫情导致无法逛街，电商视频直播导购，网红明星直播带货已经成为常规操作；疫情导致学校线上教学，疫情导致了居家办公与远程视频会议的接受度大大提高等等。

这些应用快速普及，都需要产品与技术的稳定性，毕竟画面和声音卡顿，体验太糟糕的话，用户不会想再尝试，可能就直接放弃了（当然不能买买买可以省钱/不用在线被老师催回答问题/不用穿正装和老板视频会议了），而这些背后都是RTC实时音视频技术在支撑。

也因为疫情，实时音视频背后的公司与技术也在2020年受到了资本的追捧。2020年6月声网在纳斯达克上市，年底即构获腾讯5000万美元C轮投资，资本在持续加码RTC技术的赛道。

随着5G落地，IoT、VR、AR等领域有望涌现出大量的新兴应用场景。社交泛娱乐、在线教育、互动游戏、金融服务、互联网医疗，RTC技术所能施展的空间想象力巨大。

 

三位顶级技术大咖，畅聊 RTC 新机会

 

音视频开发的大趋势下，作为开发者，你有没有关注过RTC技术关键点、创新的使用场景、RTC将引爆什么行业？通过RTC技术，创业/跳槽/加薪的机会又在哪里？

5月27日晚20：00-21：00，CSDN重磅打造的高端对话栏目《大咖来了》第十三期将聚焦2021年后疫情风暴中的 RTC技术发展趋势，邀请国内顶尖的音视频技术公司技术大牛，探讨音视频技术与发展趋势，分享真知灼见。

在直播中，将讨论如下话题：

• RTC技术的趋势与价值，成熟度如何？
• RTC在通信领域的前世今生，发展历程是什么？
• RTC和低代码、AI、5G等新技术的融合趋势
• AI、机器学习在RTC领域的应用
• 下一个将被实时音视频引爆或改变的行业会是什么？
• RTC对于开发者的机会在哪里？

我们将邀请：

蒋涛，CSDN 创始人&董事长、*客帮创投创始合伙人 

25 年软件开发经验，曾领导开发了巨人手写电脑、金山词霸和超级解霸。

1999 年创办 CSDN（China Software Developer Network）。CSDN 是专业的中文 IT 技术社区，Alexa 全球排名 26，目前注册会员超过 3100 万，涵盖 90% 国内 IT 开发从业人员。CSDN 成立二十年，为中国 IT 从业人员提供知识传播、在线学习、职业发展等全职业生命周期服务，致力于成为技术人学习和成长的家园。

2011 年创办*客帮创投，作为懂技术的投资人，先后投资了聚合数据、巨杉数据库、传智播客、乐动卓越（我叫 MT）、IT 桔子等 100 余家高科技创业公司。被投项目中 SequoiaDB 巨杉数据库作为业界*的金融级分布式关系型数据库产品，连续两年入选 Gartner 数据库报告，传智播客、柠檬微趣已于 2019 年、2017 年分别申报中小板和创业板上市，淘手游也被贵州省列为*批科创板备案名单。

任杰，融云 CTO

负责融云音视频整体技术规划和演进； 拥有 20 年以上互联网通信领域产品设计、架构设计、技术研发经验，是移动通信领域顶级技术专家，致力于为全球开发者提供专业的互联网通信云服务。

加入融云前，曾担任神州泰岳集团副总裁；原飞信CTO，带领飞信平台从零开始到千万在线；曾担任微软中国顾问咨询部电信行业负责人，带领团队完成多个千万级项目的实施。

刘莉 即构科技副总裁

刘莉博士，教授，拥有17年海外/国内运营商全球组网、数据安全、视频云平台开发和运营经验，曾任亿联网络副总经理，华为通讯云服务CTO，中国电信系统集成（广东） 事业部总经理。

刘莉博士在云计算架构和音视频领域发表中英文期刊论文17篇，国家发明专利20项。获得福建省百人计划、福建省高层次人才等多项荣誉称号。

1. kafka概述

##1.1 kafka简介

Apache Kafka 是一个快速、可扩展的、高吞吐的、可容错的分布式“发布-订阅”消息系统， 使用 Scala 与 Java 语言编写，能够将消息从一个端点传递到另一个端点，较之传统的消息中 间件(例如 ActiveMQ、RabbitMQ)，Kafka 具有高吞吐量、内置分区、支持消息副本和高容 错的特性，非常适合大规模消息处理应用程序。

Kafka 官网: http://kafka.apache.org/

Kafka主要设计目标如下：

• 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
• 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。
• 支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
• 同时支持离线数据处理和实时数据处理。
• 支持在线水平扩展。

Kafka通常用于两大类应用程序：

• 建立实时流数据管道，以可靠地在系统或应用程序之间获取数据
• 构建实时流应用程序，以转换或响应数据流

要了解Kafka如何执行这些操作，让我们从头开始深入研究Kafka的功能。

首先几个概念：

• Kafka在一个或多个可以跨越多个数据中心的服务器上作为集群运行。
• Kafka集群将记录流存储在称为主题的类别中。
• 每个记录由一个键，一个值和一个时间戳组成

1.2 kafka架构体系

1.3 kafka的应用场景

kafka的应用场景非常多, 下面我们就来举几个我们*常见的场景

1.3.1 用户的活动跟踪

用户在网站的不同活动消息发布到不同的主题中心，然后可以对这些消息进行实时监测、实时处理。当然，也可以加载到Hadoop或离线处理数据仓库，对用户进行画像。像淘宝、天猫、京东这些大型电商平台，用户的所有活动都要进行追踪的。

1.3.2 日志收集

1.3.3 限流削峰

1.3.4 高吞吐率实现

Kafka与其他MQ相比，*大的特点就是高吞吐率。为了增加存储能力，Kafka将所有的消息都写入到了低速大容量的硬盘。按理说，这将导致性能损失，但实际上，Kafka仍然可以保持超高的吞吐率，并且其性能并未受到影响。其主要采用如下方式实现了高吞吐率。

1. 顺序读写：Kafka将消息写入到了分区partition中，而分区中的消息又是顺序读写的。顺序读写要快于随机读写。
2. 零拷贝：生产者、消费者对于Kafka中的消息是采用零拷贝实现的。
3. 批量发送：Kafka允许批量发送模式。
4. 消息压缩：Kafka允许对消息集合进行压缩。

1.4 kafka的优点

1. 解耦：

在项目启动之初来预测将来项目会碰到什么需求，是*其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

2. 冗余:(副本)

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

3. 扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。

4. 灵活性&峰值处理能力

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

5. 可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

6. 顺序保证

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。

7. 缓冲

在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务*高效率的执行———写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。

8. 异步通信

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

1.5 kafka于其他MQ对比

1. RabbitMQ

RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量级，更适合于企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时先在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。

2. Redis

Redis是一个基于Key-Value对的NoSQL数据库，开发维护很活跃。虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。

3. ZeroMQ

ZeroMQ号称*快的消息队列系统，尤其针对大吞吐量的需求场景。ZeroMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序将扮演这个服务器角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果宕机，数据将会丢失。其中，Twitter的Storm 0.9.0以前的版本中默认使用ZeroMQ作为数据流的传输（Storm从0.9版本开始同时支持ZeroMQ和Netty作为传输模块）。

4. ActiveMQ

ActiveMQ是Apache下的一个子项目。 类似于ZeroMQ，它能够以代理人和点对点的技术实现队列。同时类似于RabbitMQ，它少量代码就可以高效地实现高级应用场景。

5. Kafka/Jafka

Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。

1.6 kafka的几种重要角色

1.6.1 kafka作为存储系统

任何允许发布与使用无关的消息发布的消息队列都有效地充当了运行中消息的存储系统。Kafka的不同之处在于它是一个非常好的存储系统。

写入Kafka的数据将写入磁盘并进行复制以实现容错功能。Kafka允许生产者等待确认，以便直到完全复制并确保即使写入服务器失败的情况下写入也不会完成。

Kafka的磁盘结构可以很好地扩展使用-无论服务器上有50 KB还是50 TB的持久数据，Kafka都将执行相同的操作。

由于认真对待存储并允许客户端控制其读取位置，因此您可以将Kafka视为一种专用于高性能，低延迟提交日志存储，复制和传播的专用分布式文件系统。

1.6.2 kafka作为消息传递系统

Kafka的流概念与传统的企业消息传递系统相比如何？

传统上，消息传递具有两种模型：排队和发布-订阅。在队列中，一组使用者可以从服务器中读取内容，并且每条记录都将转到其中一个。在发布-订阅记录中广播给所有消费者。这两个模型中的每一个都有优点和缺点。排队的优势在于，它允许您将数据处理划分到多个使用者实例上，从而扩展处理量。不幸的是，队列不是多用户的—一次进程读取了丢失的数据。发布-订阅允许您将数据广播到多个进程，但是由于每条消息都传递给每个订阅者，因此无法扩展处理。

Kfka的消费者群体概念概括了这两个概念。与队列一样，使用者组允许您将处理划分为一组进程（使用者组的成员）。与发布订阅一样，Kafka允许您将消息广播到多个消费者组。

Kafka模型的优点在于，每个主题都具有这些属性-可以扩展处理范围，并且是多订阅者-无需选择其中一个。

与传统的消息传递系统相比，Kafka还具有更强的订购保证。

传统队列将记录按顺序保留在服务器上，如果多个使用者从队列中消费，则服务器将按记录的存储顺序分发记录。但是，尽管服务器按顺序分发记录，但是这些记录是异步传递给使用者的，因此它们可能在不同的使用者上乱序到达。这实际上意味着在并行使用的情况下会丢失记录的顺序。消息传递系统通常通过“专有使用者”的概念来解决此问题，该概念仅允许一个进程从队列中使用，但是，这当然意味着在处理中没有并行性。

Kafka做得更好。通过在主题内具有并行性（即分区）的概念，Kafka能够在用户进程池中提供排序保证和负载均衡。这是通过将主题中的分区分配给消费者组中的消费者来实现的，以便每个分区都由组中的一个消费者完全消费。通过这样做，我们确保使用者是该分区的唯一读取器，并按顺序使用数据。由于存在许多分区，因此仍然可以平衡许多使用者实例上的负载。但是请注意，使用者组中的使用者实例不能超过分区。

1.6.3 kafka用作流处理

仅读取，写入和存储数据流是不够的，目的是实现对流的实时处理。

在Kafka中，流处理器是指从输入主题中获取连续数据流，对该输入进行一些处理并生成连续数据流以输出主题的任何东西。

例如，零售应用程序可以接受销售和装运的输入流，并输出根据此数据计算出的重新订购和价格调整流。

可以直接使用生产者和消费者API进行简单处理。但是，对于更复杂的转换，Kafka提供了完全集成的Streams API。这允许构建执行非重要处理的应用程序，这些应用程序计算流的聚合或将流连接在一起。

该功能有助于解决此类应用程序所面临的难题：处理无序数据，在代码更改时重新处理输入，执行状态计算等。

流API建立在Kafka提供的核心原语之上：它使用生产者和使用者API进行输入，使用Kafka进行状态存储，并使用相同的组机制来实现流处理器实例之间的容错。

2. kafka中的关键术语解释

2.1 Topic

主题。在 Kafka 中，使用一个类别属性来划分消息的所属类，划分消息的这个类称为 topic。 topic 相当于消息的分类标签，是一个逻辑概念

物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处

2.2 Partition

分区。topic 中的消息被分割为一个或多个 partition，其是一个物理概念，对应到系统上 就是一个或若干个目录。partition 内部的消息是有序的，但 partition 间的消息是无序的。

2.3 Segment

段。将 partition 进一步细分为了若干的 segment，每个 segment 文件的大小相等。

2.4 Broker

Kafka 集群包含一个或多个服务器，每个服务器节点称为一个 broker。

broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。

如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。

如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。

2.5 Producer

生产者, 即消息的发布者. 生产者将数据发布到他们选择的主题。生产者负责选择将哪个记录分配给主题中的哪个分区。即: 生产者生产的一条消息，会被写入到某一个 partition。

2.6 Consumer

消费者。可以从 broker 中读取消息。

一个消费者可以消费多个 topic 的消息

一个消费者可以消费同一个 topic 中的多个 partition 中的消息

一个 partiton 允许多个 consumer 同时消费

2.7 Consumer Group

consumer group 是 kafka 提供的可扩展且具有容错性的消费者机制。组内可以有多个消 费者，它们共享一个公共的 ID，即 group ID。组内的所有消费者协调在一起来消费订阅主题 的所有分区。

Kafka 保证同一个 consumer group 中只有一个 consumer 会消费某条消息，实际上，Kafka 保证的是稳定状态下每一个 consumer 实例只会消费某一个或多个特定的 partition，而某个 partition 的数据只会被某一个特定的 consumer 实例所消费。

下面我们用官网的一张图, 来标识consumer数量和partition数量的对应关系

由两台服务器组成的Kafka群集，其中包含四个带有两个使用者组的分区（P0-P3）。消费者组A有两个消费者实例，组B有四个。

其实对于这个消费组, 以前一直搞不明白, 我自己的总结是:

topic中的partitoin到group是发布订阅的通信方式，即一条topic的partition的消息会被所有的group消费，属于一对多模式；group到consumer是点对点通信方式，属于一对一模式。

举个例子: 不使用group的话，启动10个consumer消费一个topic，这10个consumer都能得到topic的所有数据，相当于这个topic中的任一条消息被消费10次。

使用group的话，连接时带上groupid，topic的消息会分发到10个consumer上，每条消息只被消费1次

2.8 Replizcas of partition

分区副本。副本是一个分区的备份，是为了防止消息丢失而创建的分区的备份。

2.9 Partition Leader

每个 partition 有多个副本，其中有且仅有一个作为 Leader，Leader 是当前负责消息读写 的 partition。即所有读写操作只能发生于 Leader 分区上。

2.10 Partition Follower

所有Follower都需要从Leader同步消息，Follower与Leader始终保持消息同步。Leader 与 Follower 的关系是主备关系，而非主从关系。

2.11 ISR

• ISR，In-Sync Replicas，是指副本同步列表。 ISR列表是由Leader负责维护。
• AR，Assigned Replicas，指某个 partition 的所有副本, 即已分配的副本列表。
• OSR，Outof-Sync Replicas, 即非同步的副本列表。
• AR = ISR + OSR

2. 12 offset

偏移量。每条消息都有一个当前Partition下唯一的64字节的offset，它是相当于当前分区*条消息的偏移量。

2.13 Broker Controller

Kafka集群的多个broker中，有一个会被选举controller，负责管理整个集群中partition和replicas的状态。

只有 Broker Controller 会向 zookeeper 中注册 Watcher，其他 broker 及分区无需注册。即 zookeeper 仅需监听 Broker Controller 的状态变化即可。

2.14 HW与LEO

• HW，HighWatermark，高水位，表示 Consumer 可以消费到的*高 partition 偏移量。HW 保证了 Kafka 集群中消息的一致性。确切地说，是保证了 partition 的 Follower 与 Leader 间数 据的一致性。
• LEO，Log End Offset，日志*后消息的偏移量。消息是被写入到 Kafka 的日志文件中的， 这是当前*后一个写入的消息在 Partition 中的偏移量。
• 对于 leader 新写入的消息，consumer 是不能立刻消费的。leader 会等待该消息被所有 ISR 中的 partition follower 同步后才会更新 HW，此时消息才能被 consumer 消费。

我相信你看完上面的概念还是懵逼的, 好吧, 下面我们就用图来形象话的表示两者的关系吧。

2.15 zookeeper

Zookeeper 负责维护和协调 broker，负责 Broker Controller 的选举。

在 kafka0.9 之前版本，offset 是由 zk 负责管理的。

总结:zk 负责 Controller 的选举，Controller 负责 leader 的选举。

2.16 Coordinator

Coordinator一般指的是运行在每个broker上的group Coordinator进程，用于管理Consumer Group中的各个成员，主要用于offset位移管理和Rebalance。一个Coordinator可以同时管理多个消费者组。

2. 17 Rebalance

当消费者组中的数量发生变化，或者topic中的partition数量发生了变化时，partition的所有权会在消费者间转移，即partition会重新分配，这个过程称为再均衡Rebalance。

再均衡能够给消费者组及broker带来高性能、高可用性和伸缩，但在再均衡期间消费者是无法读取消息的，即整个broker集群有小一段时间是不可用的。因此要避免不必要的再均衡。

2.18 offset commit

Consumer从broker中取一批消息写入buffer进行消费，在规定的时间内消费完消息后，会自动将其消费消息的offset提交给broker，以记录下哪些消息是消费过的。当然，若在时限内没有消费完毕，其是不会提交offset的。

3. kafka的工作原理和过程

3.1 消息写入算法

消息发送者将消息发送给broker, 并形成*终的可供消费者消费的log, 是已给比较复杂的过程：

• producer先从zookeeper中找到该partition的leader
• producer将消息发送给该leader
• leader将消息接入本地的log, 并通知ISR的followers
• ISR中的followers从leader中pull消息, 写入本地log后向leader发送ack
• leader收到所有ISR中的followers的ack后, 增加HW并向producer发送ack, 表示消息写入成功

3.2 消息路由策略

在通过 API 方式发布消息时，生产者是以 Record 为消息进行发布的。Record 中包含 key 与 value，value 才是我们真正的消息本身，而 key 用于路由消息所要存放的 Partition。消息 要写入到哪个 Partition 并不是随机的，而是有路由策略的。

• 若指定了 partition，则直接写入到指定的 partition;
• 若未指定 partition 但指定了 key，则通过对 key 的 hash 值与 partition 数量取模，该取模
• 结果就是要选出的 partition 索引;
• 若 partition 和 key 都未指定，则使用轮询算法选出一个 partition。

3.3 HW截断机制

如果 partition leader 接收到了新的消息， ISR 中其它 Follower 正在同步过程中，还未同 步完毕时 leader 宕机。此时就需要选举出新的 leader。若没有 HW 截断机制，将会导致 partition 中 leader 与 follower 数据的不一致。

当原 Leader 宕机后又恢复时，将其 LEO 回退到其宕机时的 HW，然后再与新的 Leader进行数据同步，这样就可以保证老 Leader 与新 Leader 中数据一致了，这种机制称为 HW 截断机制。

3.4 消息发送的可靠性

生产者向 kafka 发送消息时，可以选择需要的可靠性级别。通过 request.required.acks参数的值进行设置。

1、0值

异步发送。生产者向 kafka 发送消息而不需要 kafka 反馈成功 ack。该方式效率*高，但可靠性*低。其可能会存在消息丢失的情况。

• 在传输过程中会出现消息丢失。
• 在broker内部会出现消息丢失。
• 会出现写入到kafka中的消息的顺序与生产顺序不一致的情况。

2、1值

同步发送。生产者发送消息给 kafka，broker 的 partition leader 在收到消息后马上发送 成功 ack(无需等等 ISR 中的 Follower 同步)，生产者收到后知道消息发送成功，然后会再发送消息。如果一直未收到 kafka 的 ack，则生产者会认为消息发送失败，会重发消息。

该方式对于 Producer 来说，若没有收到 ACK，一定可以确认消息发送失败了，然后可以 重发;但是，即使收到了 ACK，也不能保证消息一定就发送成功了。故，这种情况，也可能 会发生消息丢失的情况。

3、-1值

同步发送。生产者发送消息给 kafka，kafka 收到消息后要等到 ISR 列表中的所有副本都 同步消息完成后，才向生产者发送成功 ack。如果一直未收到 kafka 的 ack，则认为消息发送 失败，会自动重发消息。该方式会出现消息重复接收的情况。

3.5 消费者消费过程解析

生产者将消息发送到topitc中, 消费者即可对其进行消费, 其消费过程如下:

1. consumer向broker提交连接请求，其所连接上的broker都会向其发送broker controller的通信URL，即配置文件中的listeners地址；
2. 当consumer指定了要消费的topic后，会向broker controller发送消费请求；
3. broker controller会为consumer分配一个或几个partition leader，并将该partition的当前offset发送给consumer；
4. consumer会按照broker controller分配的partition对其中的消息进行消费；
5. 当consumer消费完该条消息后，consumer会向broker发送一个消息已经被消费反馈，即该消息的offset；
6. 在broker接收到consumer的offset后，会更新相应的__consumer_offset中；

以上过程会一直重复，知道消费者停止请求消费；

Consumer可以重置offset，从而可以灵活消费存储在broker上的消息。

3.6 Partition Leader选举范围

当leader宕机后，broker controller会从ISR中挑选一个follower成为新的leader。如果ISR中没有其他副本怎么办？可以通过unclean.leader.election.enable的值来设置leader选举范围。

1、false

必须等到ISR列表中所有的副本都活过来才进行新的选举。该策略可靠性有保证，但可用性低。

2、true

在ISR列表中没有副本的情况下，可以选择任意一个没有宕机的主机作为新的leader，该策略可用性高，但可靠性没有保证。

3.7 重复消费问题的解决方案

1、同一个consumer重复消费

当Consumer由于消费能力低而引发了消费超时，则可能会形成重复消费。

在某数据刚好消费完毕，但是正准备提交offset时候，消费时间超时，则broker认为这条消息未消费成功。这时就会产生重复消费问题。

**其解决方案：**延长offset提交时间。

2、不同的consumer重复消费

当Consumer消费了消息，但还没有提交offset时宕机，则这些已经被消费过的消息会被重复消费。

**其解决方案：**将自动提交改为手动提交。

3.8 从架构设计上解决kafka重复消费的问题

其实在开发的时候, 我们在设计程序的时候, 比如考虑到网络故障等一些异常的情况, 我们都会设置消息的重试次数,

可能还有其他可能出现消息重复, 那我们应该如何解决呢?

下面提供三个方案:

3.8.1 方案一: 保存并查询

给每个消息都设置一个独一无二的uuid, 所有的消息, 我们都要存一个uuid, 我们在消费消息的时候, 首先去持久化系统中查询一下, 看这个看是否以前消费过, 如没有消费过, 在进行消费, 如果已经消费过, 丢弃就好了, 下图, 表明了这种方案：

3.8.2 方案二: 利用幂等

幂等（Idempotence）在数学上是这样定义的，如果一个函数 f(x) 满足：f(f(x)) = f(x)，则函数 f(x) 满足幂等性。

这个概念被拓展到计算机领域，被用来描述一个操作、方法或者服务。一个幂等操作的特点是，其任意多次执行所产生的影响均与一次执行的影响相同。一个幂等的方法，使用同样的参数，对它进行多次调用和一次调用，对系统产生的影响是一样的。所以，对于幂等的方法，不用担心重复执行会对系统造成任何改变。

我们举个例子来说明一下。在不考虑并发的情况下，“将 X 老师的账户余额设置为 100 万元”，执行一次后对系统的影响是，X 老师的账户余额变成了 100 万元。只要提供的参数 100万元不变，那即使再执行多少次，X 老师的账户余额始终都是 100万元，不会变化，这个操作就是一个幂等的操作。

再举一个例子，“将 X 老师的余额加 100 万元”，这个操作它就不是幂等的，每执行一次，账户余额就会增加 100 万元，执行多次和执行一次对系统的影响（也就是账户的余额）是不一样的。

所以，通过这两个例子，我们可以想到如果系统消费消息的业务逻辑具备幂等性，那就不用担心消息重复的问题了，因为同一条消息，消费一次和消费多次对系统的影响是完全一样的。也就可以认为，消费多次等于消费一次。

那么，如何实现幂等操作呢？*好的方式就是，从业务逻辑设计上入手，将消费的业务逻辑设计成具备幂等性的操作。但是，不是所有的业务都能设计成天然幂等的，这里就需要一些方法和技巧来实现幂等。

下面我们介绍一种常用的方法：利用数据库的唯一约束实现幂等。

例如，我们刚刚提到的那个不具备幂等特性的转账的例子：将 X 老师的账户余额加 100 万元。在这个例子中，我们可以通过改造业务逻辑，让它具备幂等性。

首先，我们可以限定，对于每个转账单每个账户只可以执行一次变更操作，在分布式系统中，这个限制实现的方法非常多，*简单的是我们在数据库中建一张转账流水表，这个表有三个字段：转账单 ID、账户 ID 和变更金额，然后给转账单 ID 和账户 ID 这两个字段联合起来创建一个唯一约束，这样对于相同的转账单 ID 和账户 ID，表里至多只能存在一条记录。

这样，我们消费消息的逻辑可以变为：“在转账流水表中增加一条转账记录，然后再根据转账记录，异步操作更新用户余额即可。”在转账流水表增加一条转账记录这个操作中，由于我们在这个表中预先定义了“账户 ID 转账单 ID”的唯一约束，对于同一个转账单同一个账户只能插入一条记录，后续重复的插入操作都会失败，这样就实现了一个幂等的操作。

3.8.3 方案三: 设置前提条件

为更新的数据设置前置条件另外一种实现幂等的思路是，给数据变更设置一个前置条件，如果满足条件就更新数据，否则拒*更新数据，在更新数据的时候，同时变更前置条件中需要判断的数据。

这样，重复执行这个操作时，由于*次更新数据的时候已经变更了前置条件中需要判断的数据，不满足前置条件，则不会重复执行更新数据操作。

比如，刚刚我们说过，“将 X 老师的账户的余额增加 100 万元”这个操作并不满足幂等性，我们可以把这个操作加上一个前置条件，变为：“如果X老师的账户当前的余额为 500万元，将余额加 100万元”，这个操作就具备了幂等性。

对应到消息队列中的使用时，可以在发消息时在消息体中带上当前的余额，在消费的时候进行判断数据库中，当前余额是否与消息中的余额相等，只有相等才执行变更操作。

但是，如果我们要更新的数据不是数值，或者我们要做一个比较复杂的更新操作怎么办？用什么作为前置判断条件呢？更加通用的方法是，给你的数据增加一个版本号属性，每次更数据前，比较当前数据的版本号是否和消息中的版本号一致，如果不一致就拒*更新数据，更新数据的同时将版本号 +1，一样可以实现幂等。

4 . kafka集群搭建

我们在工作中, 为了保证环境的高可用, 防止单点, kafka都是以集群的方式出现的, 下面就带领大家一起搭建一套kafka集群环境

我们在官网下载kafka, 下载地址为: http://kafka.apache.org/downloads, 下载我们需要的版本, 推荐使用稳定的版本

4.1 搭建集群

1、下载并解压

1. cd /usr/local/src
2. wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.0/kafka_2.11–2.4.0.tgz
3. mkdir /data/servers
4. tar xzvf kafka_2.11–2.4.0.tgz -C /data/servers/
5. cd /data/servers/kafka_2.11–2.4.0

2、修改配置文件

kafka的配置文件$KAFKA_HOME/config/server.properties, 主要修改一下下面几项：

1. # 确保每个机器上的id不一样
2. broker.id=0
3. # 配置服务端的监控地址
4. listeners=PLAINTEXT://192.168.51.128:9092
5. # kafka 日志目录
6. log.dirs=/data/servers/kafka_2.11–2.4.0/logs
7. # kafka设置的partitons的个数
8. num.partitions=1
9. # zookeeper的连接地址, 如果有自己的zookeeper集群, 请直接使用自己搭建的zookeeper集群
10. zookeeper.connect=192.168.51.128:2181

因为我自己是本机做实验, 所有使用的是一个主机的不同端口，在线上,，就是不同的机器，大家参考即可。

我们这里使用kafka的zookeeper，只启动一个节点， 但是正真的生产过程中，是需要zookeeper集群，自己搭建就好，后期我们也会出zookeeper的教程，大家请关注就好了。

3、拷贝3份配置文件

1. # 创建对应的日志目录
2. mkdir -p /data/servers/kafka_2.11–2.4.0/logs/9092
3. mkdir -p /data/servers/kafka_2.11–2.4.0/logs/9093
4. mkdir -p /data/servers/kafka_2.11–2.4.0/logs/9094
6. # 拷贝三份配置文件
7. cp server.properties server_9092.properties
8. cp server.properties server_9093.properties
9. cp server.properties server_9094.properties

修改不同端口对应的文件：

1. # 9092的id为0, 9093的id为1, 9094的id为2
2. broker.id=0
3. # 配置服务端的监控地址, 分别在不通的配置文件中写入不同的端口
4. listeners=PLAINTEXT://192.168.51.128:9092
5. # kafka 日志目录, 目录也是对应不同的端口
6. log.dirs=/data/servers/kafka_2.11–2.4.0/logs/9092
7. # kafka设置的partitons的个数
8. num.partitions=1
9. # zookeeper的连接地址, 如果有自己的zookeeper集群, 请直接使用自己搭建的zookeeper集群
10. zookeeper.connect=192.168.51.128:2181

4、修改zookeeper的配置文件

1. dataDir=/data/servers/zookeeper
2. server.1=192.168.51.128:2888:3888

然后创建zookeeper的myid文件

echo "1"> /data/servers/zookeeper/myid

5、启动zookeeper

使用kafka内置的zookeeper

1. cd /data/servers/kafka_2.11–2.4.0/bin
2. zookeeper-server-start.sh -daemon ../config/zookeeper.properties
3. netstat -anp |grep 2181

6、启动kafka

1. ./kafka-server-start.sh -daemon ../config/server_9092.properties
2. ./kafka-server-start.sh -daemon ../config/server_9093.properties
3. ./kafka-server-start.sh -daemon ../config/server_9094.properties

4.2 kafka的操作

1. topic

我们先来看一下创建topic常用的参数吧

–create 创建topic

–delete 删除topic

–alter 修改topic的名字或者partition个数

–list 查看topic

–describe 查看topic的详细信息

–topic <String: topic> 指定topic的名字

–zookeeper <String: hosts> 指定zookeeper的连接地址,

​ 参数提示并不赞成这样使用

​ DEPRECATED, The connection string for
​ the zookeeper connection in the form
​ host:port. Multiple hosts can be
​ given to allow fail-over.

–bootstrap-server <String: server to connect to>: 指定kafka的连接地址, 推荐使用这个,

​ 参数的提示信息显示

​ REQUIRED: The Kafka server to connect

1. to. In case of providing this, a
2. direct Zookeeper connection won‘t be
3. required.

–replication-factor <Integer: replication factor> : 对于每个partiton的备份个数

​ The replication factor for each
​ partition in the topic being
​ created. If not supplied, defaults
​ to the cluster default.

–partitions <Integer: # of partitions>: 指定该topic的分区的个数

示例:

1. cd /data/servers/kafka_2.11–2.4.0/bin
2. # 创建topic test1
3. kafka-topics.sh –create –bootstrap-server=192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094 –replication-factor 1 –partitions 1 –topic test1
4. # 创建topic test2
5. kafka-topics.sh –create –bootstrap-server=192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094 –replication-factor 1 –partitions 1 –topic test2
6. # 查看topic
7. kafka-topics.sh –list –bootstrap-server=192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094

2、自动创建topic

我们在工作中, 如果我们不想去管理topic, 可以通过kafka的配置文件来管理, 我们可以让kafka自动创建topic, 需要在我们的kafka配置文件中加入如下配置文件

auto.create.topics.enable=true

如果删除topic想达到物理删除的目的, 也是需要配置的

delete.topic.enable=true

3、发送消息

他们可以通过客户端的命令生产消息

先来看看kafka-console-producer.sh常用的几个参数吧

–topic <String: topic> 指定topic

–timeout <Integer: timeout_ms> 超时时间

–sync 异步发送消息

–broker-list <String: broker-list> 官网提示: REQUIRED: The broker list string in the form HOST1:PORT1,HOST2:PORT2. 这个参数是必须的

kafka-console-producer.sh --broker-list 192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094 --topic test1

4、消费消息

我们也还是先来看看kafka-console-consumer.sh的参数吧

–topic <String: topic> 指定topic

–group <String: consumer group id> 指定消费者组

–from-beginning : 指定从开始进行消费, 如果不指定, 就从当前进行消费

–bootstrap-server : kafka的连接地址

kafka-console-consumer.sh --bootstrap-server 192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094 --topic test1 ---beginning

4.3 kafka的日志

kafka的日志分两种:

*种日志: 是我们的kafka的启动日志, 就是我们排查问题, 查看报错信息的日志,

第二种日志:就是我们的数据日志, kafka是我们的数据是以日志的形式存在存盘中的, 我们第二种所说的日志就是我们的partiton与segment

那我们就来说说备份和分区吧

我们创建一个分区, 一个备份, 那么test就应该在三台机器上或者三个数据目录只有一个test-0, (分区的下标是从0开始的)

如果我们创建N个分区, 我们就会在三个服务器上发现, test_0-n

如果我们创建M个备份, 我们就会在发现, test_0 到test_n 每一个都是M个

5. kafaka API

5.1 使用kafaka原生的api

1.消费者自动提交:

定义自己的生产者

1. import org.apache.kafka.clients.producer.Callback;
2. import org.apache.kafka.clients.producer.KafkaProducer;
3. import org.apache.kafka.clients.producer.ProducerRecord;
4. import org.apache.kafka.clients.producer.RecordMetadata;
6. import java.util.Properties;
8. /**
9. * @ClassName MyKafkaProducer
10. * @Description TODO
11. * @Author lingxiangxiang
12. * @Date 3:37 PM
13. * @Version 1.0
14. **/
15. public class MyKafkaProducer {
16. private org.apache.kafka.clients.producer.KafkaProducer<Integer, String> producer;
18. public MyKafkaProducer() {
19. Properties properties = new Properties();
20. properties.put(“bootstrap.servers”, “192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094”);
21. properties.put(“key.serializer”, “org.apache.kafka.common.serialization.IntegerSerializer”);
22. properties.put(“value.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);
23. // 设置批量发送
24. properties.put(“batch.size”, 16384);
25. // 批量发送的等待时间50ms, 超过50ms, 不足批量大小也发送
26. properties.put(“linger.ms”, 50);
27. this.producer = new org.apache.kafka.clients.producer.KafkaProducer<Integer, String>(properties);
28. }
30. public boolean sendMsg() {
31. boolean result = true;
32. try {
33. // 正常发送, test2是topic, 0代表的是分区, 1代表的是key, hello world是发送的消息内容
34. final ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(“test2”, 0, 1, “hello world”);
35. producer.send(record);
36. // 有回调函数的调用
37. producer.send(record, new Callback() {
38. @Override
39. public void onCompletion(RecordMetadata recordMetadata, Exception e) {
40. System.out.println(recordMetadata.topic());
41. System.out.println(recordMetadata.partition());
42. System.out.println(recordMetadata.offset());
43. }
44. });
45. // 自己定义一个类
46. producer.send(record, new MyCallback(record));
47. } catch (Exception e) {
48. result = false;
49. }
50. return result;
51. }
52. }

生产者测试类:

在生产者测试类中,自己遇到一个坑, 就是*后自己没有加sleep, 就是怎么检查自己的代码都没有问题, 但是*后就是没法发送成功消息, *后加了一个sleep就可以了, 因为主函数main已经执行完退出, 但是消息并没有发送完成, 需要进行等待一下.当然, 你在生产环境中可能不会遇到这样问题, 呵呵, 代码如下：

1. import static java.lang.Thread.sleep;
3. /**
4. * @ClassName MyKafkaProducerTest
5. * @Description TODO
6. * @Author lingxiangxiang
7. * @Date 3:46 PM
8. * @Version 1.0
9. **/
10. public class MyKafkaProducerTest {
11. public static void main(String[] args) throws InterruptedException {
12. MyKafkaProducer producer = new MyKafkaProducer();
13. boolean result = producer.sendMsg();
14. System.out.println(“send msg “ + result);
15. sleep(1000);
16. }
17. }

消费者类：

1. import kafka.utils.ShutdownableThread;
2. import org.apache.kafka.clients.consumer.ConsumerRecord;
3. import org.apache.kafka.clients.consumer.ConsumerRecords;
4. import org.apache.kafka.clients.consumer.KafkaConsumer;
6. import java.util.Arrays;
7. import java.util.Collections;
8. import java.util.Properties;
10. /**
11. * @ClassName MyKafkaConsumer
12. * @Description TODO
13. * @Author lingxiangxiang
14. * @Date 4:12 PM
15. * @Version 1.0
16. **/
17. public class MyKafkaConsumer extends ShutdownableThread {
19. private KafkaConsumer<Integer, String> consumer;
21. public MyKafkaConsumer() {
22. super(“KafkaConsumerTest”, false);
23. Properties properties = new Properties();
24. properties.put(“bootstrap.servers”, “192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094”);
25. properties.put(“group.id”, “mygroup”);
26. properties.put(“enable.auto.commit”, “true”);
27. properties.put(“auto.commit.interval.ms”, “1000”);
28. properties.put(“session.timeout.ms”, “30000”);
29. properties.put(“heartbeat.interval.ms”, “10000”);
30. properties.put(“auto.offset.reset”, “earliest”);
31. properties.put(“key.deserializer”, “org.apache.kafka.common.serialization.IntegerDeserializer”);
32. properties.put(“value.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”);
33. this.consumer = new KafkaConsumer<Integer, String>(properties);
34. }
36. @Override
37. public void doWork() {
38. consumer.subscribe(Arrays.asList(“test2”));
39. ConsumerRecords<Integer, String>records = consumer.poll(1000);
40. for (ConsumerRecord record : records) {
41. System.out.println(“topic = “ + record.topic());
42. System.out.println(“partition = “ + record.partition());
43. System.out.println(“key = “ + record.key());
44. System.out.println(“value = “ + record.value());
45. }
46. }
47. }

消费者的测试类：

1. /**
2. * @ClassName MyConsumerTest
3. * @Description TODO
4. * @Author lingxiangxiang
5. * @Date 4:23 PM
6. * @Version 1.0
7. **/
8. public class MyConsumerTest {
9. public static void main(String[] args) {
10. MyKafkaConsumer consumer = new MyKafkaConsumer();
11. consumer.start();
12. System.out.println(“==================”);
13. }
14. }

2. 消费者同步手动提交

前面的消费者都是以自动提交 offset 的方式对 broker 中的消息进行消费的，但自动提交 可能会出现消息重复消费的情况。所以在生产环境下，很多时候需要对 offset 进行手动提交， 以解决重复消费的问题。

手动提交又可以划分为同步提交、异步提交，同异步联合提交。这些提交方式仅仅是 doWork()方法不相同，其构造器是相同的。所以下面首先在前面消费者类的基础上进行构造 器的修改，然后再分别实现三种不同的提交方式。

同步提交方式是，消费者向 broker 提交 offset 后等待 broker 成功响应。若没有收到响 应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量。

修改前面的MyKafkaConsumer.java, 主要修改下面的配置：

1. import kafka.utils.ShutdownableThread;
2. import org.apache.kafka.clients.consumer.ConsumerRecord;
3. import org.apache.kafka.clients.consumer.ConsumerRecords;
4. import org.apache.kafka.clients.consumer.KafkaConsumer;
6. import java.util.Arrays;
7. import java.util.Collections;
8. import java.util.Properties;
10. /**
11. * @ClassName MyKafkaConsumer
12. * @Description TODO
13. * @Author lingxiangxiang
14. * @Date 4:12 PM
15. * @Version 1.0
16. **/
17. public class MyKafkaConsumer extends ShutdownableThread {
19. private KafkaConsumer<Integer, String> consumer;
21. public MyKafkaConsumer() {
22. super(“KafkaConsumerTest”, false);
23. Properties properties = new Properties();
24. properties.put(“bootstrap.servers”, “192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094”);
25. properties.put(“group.id”, “mygroup”);
26. // 这里要修改成手动提交
27. properties.put(“enable.auto.commit”, “false”);
28. // properties.put(“auto.commit.interval.ms”, “1000”);
29. properties.put(“session.timeout.ms”, “30000”);
30. properties.put(“heartbeat.interval.ms”, “10000”);
31. properties.put(“auto.offset.reset”, “earliest”);
32. properties.put(“key.deserializer”, “org.apache.kafka.common.serialization.IntegerDeserializer”);
33. properties.put(“value.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”);
34. this.consumer = new KafkaConsumer<Integer, String>(properties);
35. }
37. @Override
38. public void doWork() {
39. consumer.subscribe(Arrays.asList(“test2”));
40. ConsumerRecords<Integer, String>records = consumer.poll(1000);
41. for (ConsumerRecord record : records) {
42. System.out.println(“topic = “ + record.topic());
43. System.out.println(“partition = “ + record.partition());
44. System.out.println(“key = “ + record.key());
45. System.out.println(“value = “ + record.value());
47. //手动同步提交
48. consumer.commitSync();
49. }
51. }
52. }

3、消费者异步手工提交

手动同步提交方式需要等待 broker 的成功响应，效率太低，影响消费者的吞吐量。异步提交方式是，消费者向 broker 提交 offset 后不用等待成功响应，所以其增加了消费者的吞吐量。

1. import kafka.utils.ShutdownableThread;
2. import org.apache.kafka.clients.consumer.ConsumerRecord;
3. import org.apache.kafka.clients.consumer.ConsumerRecords;
4. import org.apache.kafka.clients.consumer.KafkaConsumer;
6. import java.util.Arrays;
7. import java.util.Collections;
8. import java.util.Properties;
10. /**
11. * @ClassName MyKafkaConsumer
12. * @Description TODO
13. * @Author lingxiangxiang
14. * @Date 4:12 PM
15. * @Version 1.0
16. **/
17. public class MyKafkaConsumer extends ShutdownableThread {
19. private KafkaConsumer<Integer, String> consumer;
21. public MyKafkaConsumer() {
22. super(“KafkaConsumerTest”, false);
23. Properties properties = new Properties();
24. properties.put(“bootstrap.servers”, “192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094”);
25. properties.put(“group.id”, “mygroup”);
26. // 这里要修改成手动提交
27. properties.put(“enable.auto.commit”, “false”);
28. // properties.put(“auto.commit.interval.ms”, “1000”);
29. properties.put(“session.timeout.ms”, “30000”);
30. properties.put(“heartbeat.interval.ms”, “10000”);
31. properties.put(“auto.offset.reset”, “earliest”);
32. properties.put(“key.deserializer”, “org.apache.kafka.common.serialization.IntegerDeserializer”);
33. properties.put(“value.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”);
34. this.consumer = new KafkaConsumer<Integer, String>(properties);
35. }
37. @Override
38. public void doWork() {
39. consumer.subscribe(Arrays.asList(“test2”));
40. ConsumerRecords<Integer, String>records = consumer.poll(1000);
41. for (ConsumerRecord record : records) {
42. System.out.println(“topic = “ + record.topic());
43. System.out.println(“partition = “ + record.partition());
44. System.out.println(“key = “ + record.key());
45. System.out.println(“value = “ + record.value());
47. //手动同步提交
48. // consumer.commitSync();
49. //手动异步提交
50. // consumer.commitAsync();
51. // 带回调公共的手动异步提交
52. consumer.commitAsync((offsets, e) -> {
53. if(e != null) {
54. System.out.println(“提交次数, offsets = “ + offsets);
55. System.out.println(“exception = “ + e);
56. }
57. });
58. }
59. }
60. }

5.2 springboot使用kafka

现在大家的开发过程中, 很多都用的是springboot的项目, 直接启动了, 如果还是用原生的API, 就是有点low了啊, 那kafka是如何和springboot进行联合的呢?

1. maven配置

1. <!– https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients –>
2. <dependency>
3. <groupId>org.apache.kafka</groupId>
4. <artifactId>kafka-clients</artifactId>
5. <version>2.1.1</version>
6. </dependency>

2. 添加配置文件

在application.properties中加入如下配置信息:

1. # kafka 连接地址
2. spring.kafka.bootstrap-servers = 192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094
4. # 生产者
5. spring.kafka.producer.acks = 0
6. spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer
7. spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.StringSerializer
8. spring.kafka.producer.retries = 3
9. spring.kafka.producer.batch-size = 4096
10. spring.kafka.producer.buffer-memory = 33554432
11. spring.kafka.producer.compression-type = gzip
13. # 消费者
14. spring.kafka.consumer.group-id = mygroup
15. spring.kafka.consumer.auto-commit-interval = 5000
16. spring.kafka.consumer.heartbeat-interval = 3000
17. spring.kafka.consumer.key-deserializer = org.apache.kafka.common.serialization.StringDeserializer
18. spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.StringDeserializer
19. spring.kafka.consumer.auto-offset-reset = earliest
20. spring.kafka.consumer.enable-auto-commit = true
21. # listenner, 标识消费者监听的个数
22. spring.kafka.listener.concurrency = 8
23. # topic的名字
24. kafka.topic1 = topic1

3. 生产者

1. import lombok.extern.slf4j.Slf4j;
2. import org.springframework.beans.factory.annotation.Value;
3. import org.springframework.kafka.core.KafkaTemplate;
5. @Service
6. @Slf4j
7. public class MyKafkaProducerServiceImpl implements MyKafkaProducerService {
8. @Resource
9. private KafkaTemplate<String, String> kafkaTemplate;
10. // 读取配置文件
11. @Value(“${kafka.topic1}”)
12. private String topic;
14. @Override
15. public void sendKafka() {
16. kafkaTemplate.send(topic, “hell world”);
17. }
18. }

4. 消费者

1. @Component
2. @Slf4j
3. public class MyKafkaConsumer {
4. @KafkaListener(topics = “${kafka.topic1}“)
5. public void listen(ConsumerRecord<?, ?> record) {
6. Optional<?> kafkaMessage = Optional.ofNullable(record.value());
7. if (kafkaMessage.isPresent()) {
8. log.info(“—————– record =” + record);
9. log.info(“—————— message =” + kafkaMessage.get());

1.超大规模

“云”具有相当的规模，Google云计算已经拥有100多万台服务器， Amazon、IBM、微软、Yahoo等的“云”均拥有几十万台服务器。企业私有云一般拥有数百上千台服务器。“云”能赋予用户前所未有的计算能力。

2.虚拟化

云计算支持用户在任意位置、使用各种终端获取应用服务。所请求的资源来自“云”，而不是固定的有形的实体。应用在“云”中某处运行，但实际上用户无需了解、也不用担心应用运行的具体位置。只需要一台笔记本电脑或者一部手机，就可以通过网络服务来实现我们需要的一切，甚至包括超级计算这样的任务。

3.高可靠性

“云”使用了数据多副本容错、计算节点同构可互换等措施来保障服务的高可靠性，使用云计算比使用本地计算机可靠。

4.通用性

云计算不针对特定的应用，在“云”的支撑下可以构造出千变万化的应用，同一个“云”可以同时支撑不同的应用运行。

5.高可扩展性

“云”的规模可以动态伸缩，满足应用和用户规模增长的需要。

6.按需服务

“云”是一个庞大的资源池，你可按需购买；云可以像自来水、电、煤气那样计费。

7.*其廉价

由于“云”的特殊容错措施可以采用*其廉价的节点来构成云，“云”的自动化集中式管理使大量企业无需负担日益高昂的数据中心管理成本，“云”的通用性使资源的利用率较之传统系统大幅提升，因此用户可以充分享受“云”的低成本优势，经常只要花费几百美元、几天时间就能完成以前需要数万美元、数月时间才能完成的任务。

8.潜在危险性

云计算服务除了提供计算服务外，还必然提供了存储服务。但是云计算服务当前垄断在私人机构（企业）手中，而他们仅仅能够提供商业信用。对于政府机构、商业机构（特别像银行这样持有敏感数据的商业机构）对于选择云计算服务应保持足够的警惕。
一旦商业用户大规模使用私人机构提供的云计算服务，无论其技术优势有多强，都不可避免地让这些私人机构以“数据（信息）”的重要性挟制整个社会。对于信息社会而言，“信息”是至关重要的。另一方面，云计算中的数据对于数据所有者以外的其他用户云计算用户是保密的，但是对于提供云计算的商业机构而言确实毫无秘密可言。
所有这些潜在的危险，是商业机构和政府机构选择云计算服务、特别是国外机构提供的云计算服务时，不得不考虑的一个重要的前提。

输入完字符，自动提示 

1、打开 MyEclipse 8.5 -> Window -> Perferences -> Java -> Editor -> Content Assist，在右边*下面一栏找到 auto-Activation ，下面有三个选项，找到第二个“Auto activation triggers for Java：”选项

在其后的文本框中会看到一个“.”存在。这表示：只有输入“.”之后才会有代码提示和自动补全，我们要修改的地方就是这里。把该文本框中的“.”换掉，换成“abcdefghijklmnopqrstuvwxyz.”，这样，你在MyEclipse 8.5里面写Java代码就可以做到按“abcdefghijklmnopqrstuvwxyz.”中的任意一个字符都会有代码提示。当然，你还可以在里面加上更多的字符，只要是你需要的，你都可以加上去！比如说在JPA（Java persistence API）里面会用大量的“@”符号进行相应的注解，你可以把“@”一并加上去。以后在按下“@”付好后就会有相应的代码提示，此时你只需移动鼠标或是按键盘的上下键按钮进行选择，轻松解除你不熟悉Java代码的恐惧！

2、另外，MyEclipse 6.5 的版本，这里的设置稍有不同。

打开 MyEclipse 6.5 -> Window -> Perferences -> Java -> Editor -> Content（记住，不是他的下级选项，而是它本身），*下面一栏 auto-Activation ，会看到只有一个“.”存在。表示：只有输入“.”之后才会有代码提 示，我们要修改的地方就是这里，可是Eclipse默认只允许输入4个自定义字符。
不过我们可以把当前的设置导出，保 存为一个文件，然后在文件中修改，再导入设置，这样就可以突破Eclipse的限制。
先把上图中“.“的地方输入几个 随便的字符，例如“asdf“，点*下面的“OK“来保存设置。
然 后打开 Eclipse的 File -> Export,在窗口中展开 General -> Perferences–>Export all然后点击 NEXT。然后点击“Browse”选择任意的一个路径，保存配置文件，然后点击“Finish“用记事本打开刚才 保存的那个配置文件（扩展文件名：*.epf），按“ctrl + F”，输入刚才设置的“asdf”，找到刚才字符串。把“asdf”修改为“abcdefghijklmnopqrstuvwxyz.”，然后保存，退出记事本。
打 开Eclipse的 File -> Import 然后在打开的窗口里展开 General ->
Perferences，点击NEXT， 选中刚才修改过的配置文件，Finish。现在，再打开Window ->
Perferences，并依次展开 Java -> Editor -> Content
Assist，会发现已经超过了4个字符，也就是说我们输入任何字母和“.”都 会有代码提示了。

按快捷键弹出代码提示

myelipse中的内容补齐的功能是使用content assist。
myeclipse 6把alt+/分配给了单词提示,还有代码提示的快捷代码也不是alt+/,因此要恢复代码提示用alt+/.需要做两件事.
1,windows–>preference–>general–>keys 下设置Content Assist 的快捷键
2,windows–>preference–>general–>keys 下设置word completion 的快捷键 改成其它的.

如果还是不提示有可能是提示功能被关了.
如下做
1、菜单window->Preferences->Java->Editor->Content Assist->Enable auto activation 选项要打上勾
2、window->Preferences->Java->Editor->Content Assist->Advanced 上面的选项卡Select

天偶然的学习了一下几种关于消息摘要算法的知识。个人觉得很好。应着老话“好记性不如烂笔头”，我就码了几行代码咯。

---

算法嘛，没什么好说的了。毕竟是设计者智慧与汗水的结晶，也是时代进步的推动力。我们能做的就是将这种算法运用得当，造福全人类，就行了！

//格外需要注意的是采用CC（Commons codec）方式生成消息摘要时,一定要记得导入相关的jar包哦！

• 1

• 1

---

MD方式

---

1. package MD;
2. /**
3. * MD:MessageDigest—-消息摘要算法
4. */
6. import java.security.MessageDigest;
8. import org.apache.commons.codec.binary.Hex;
9. import org.apache.commons.codec.digest.DigestUtils;
10. import org.apache.commons.codec.digest.Md5Crypt;
11. import org.junit.Test;
13. /**
14. * 简单的使用MessageDigest实现消息摘要的小案例
15. * 收获：
16. * 使用摘要算法获得字节数组无法直接输出，需要转成相应的十六进制才能获得结果
17. * @author Summer
18. *
19. */
20. public class Demo {
22. private static String targetString = “I am Summer!”;
24. /**
25. * 测试MD5算法加密效果，转成了十六进制
26. * JDK实现
27. * @throws Exception
28. */
29. @Test
30. public void test1() throws Exception {
31. MessageDigest md = MessageDigest.getInstance(“MD5”);
32. byte [] bytes = md.digest(targetString.getBytes());
33. String result = Hex.encodeHexString(bytes);
34. System.out.println(“MD5加密后的串是：”+result);
35. //5b704caf20c179cdf61d7121e59dcd76
36. }
38. /**
39. * 测试MD2算法加密效果，转成了十六进制
40. * JDK实现
41. * @throws Exception
42. */
43. @Test
44. public void test2() throws Exception {
45. MessageDigest md = MessageDigest.getInstance(“MD2”);
46. byte [] bytes = md.digest(targetString.getBytes());
47. String result = Hex.encodeHexString(bytes);
48. System.out.println(“MD5加密后的串是：”+result);
49. //387992acc0a756abc7026d635383b0a7
50. }
52. /**
53. * 测试MD2算法加密效果，不转成十六进制
54. * JDK实现
55. * @throws Exception
56. */
57. @Test
58. public void test3() throws Exception {
59. MessageDigest md = MessageDigest.getInstance(“MD2”);
60. byte [] bytes = md.digest(targetString.getBytes());
61. System.out.println(bytes);
62. }
65. /**
66. * 测试MD5算法加密效果，不转成十六进制
67. * JDK实现
68. * @throws Exception
69. */
70. @Test
71. public void test4() throws Exception {
72. MessageDigest md = MessageDigest.getInstance(“MD5”);
73. byte [] bytes = md.digest(targetString.getBytes());
74. System.out.println(bytes.toString());
75. }
77. /**
78. * 使用commons codec的方式对目标字符串进行加密MD2，并使用十六进制进行输出
79. * @throws Exception
80. */
81. @Test
82. public void test5() throws Exception {
83. MessageDigest md = DigestUtils.getMd2Digest();
84. byte[] bytes = md.digest(targetString.getBytes());
85. String result = Hex.encodeHexString(bytes);
86. System.out.println(“Commons codec 方式加密MD2：”+ result);
87. //387992acc0a756abc7026d635383b0a7
88. }
91. /**
92. * 使用commons codec的方式对目标字符串进行加密MD5，并使用十六进制进行输出
93. * 对比方法发现，使用CC这个开源方式，可以大大的简化操作。但其底层仍是JDK实现的，cc知识做了一些简化，仅此而已
94. * @throws Exception
95. */
96. @Test
97. public void test6() throws Exception {
98. MessageDigest md = DigestUtils.getMd5Digest();
99. byte[] bytes = md.digest(targetString.getBytes());
100. String result = Hex.encodeHexString(bytes);
101. System.out.println(“Commons codec 方式加密MD5：”+ result);
102. //5b704caf20c179cdf61d7121e59dcd76
103. System.out.println(DigestUtils.md5Hex(targetString));
104. //5b704caf20c179cdf61d7121e59dcd76
105. }
107. }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79
• 80
• 81
• 82
• 83
• 84
• 85
• 86
• 87
• 88
• 89
• 90
• 91
• 92
• 93
• 94
• 95
• 96
• 97
• 98
• 99
• 100
• 101
• 102
• 103
• 104
• 105
• 106
• 107
• 108

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79
• 80
• 81
• 82
• 83
• 84
• 85
• 86
• 87
• 88
• 89
• 90
• 91
• 92
• 93
• 94
• 95
• 96
• 97
• 98
• 99
• 100
• 101
• 102
• 103
• 104
• 105
• 106
• 107
• 108

---

SHA方式

---

1. package SHA;
2. /**
3. * SHA: Security Hash Algorithm —–安全哈希算法
4. */
6. import java.security.MessageDigest;
8. import org.apache.commons.codec.binary.Hex;
9. import org.junit.Test;
11. public class Demo {
13. private static String targetString = “I am Summer!”;
16. /**
17. * 使用JDK方式以SHA1的方式实现消息摘要算法
18. * @throws Exception
19. */
20. @Test
21. public void testSHA1() throws Exception {
23. /*方式一：
24. * //算法名称对于SHA1方式: 可以是SHA也可以是SHA1
25. MessageDigest md = MessageDigest.getInstance(“SHA”);
26. byte[] bytes = md.digest(targetString.getBytes());
27. String result = Hex.encodeHexString(bytes);
28. System.out.println(“JDK 方式实现SHA1消息摘要的加密:\t”+ result);
29. //a621ca634410c2521ef560736a6a9da048f42961
30. */
32. /*
33. * 方式二：
34. */
35. MessageDigest md = MessageDigest.getInstance(“SHA”);
36. md.update(targetString.getBytes());
37. String result = Hex.encodeHexString(md.digest());
38. System.out.println(result);
40. }
43. /**
44. * 使用JDK方式以SHA-224的方式实现消息摘要算法
45. * @throws Exception
46. */
47. @Test
48. public void testSHA224() throws Exception {
50. MessageDigest md = MessageDigest.getInstance(“SHA-224”);
51. md.update(targetString.getBytes());
52. String result = Hex.encodeHexString(md.digest());
53. System.out.println(result);
54. //6fb07533ef05a1f1b6a0fbd25f08ed3cbcf17807a507c0224756f06d
56. }
59. /**
60. * 使用JDK方式以SHA-384的方式实现消息摘要算法
61. * @throws Exception
62. */
63. @Test
64. public void testSHA384() throws Exception {
66. MessageDigest md = MessageDigest.getInstance(“SHA-384”);
67. md.update(targetString.getBytes());
68. String result = Hex.encodeHexString(md.digest());
69. System.out.println(result);
70. //c640d1c73a8e2078b290f284fce59da103ecfc1c3c9442cc90ebd5f08900d8e6f19551da7b2f213d96dd055bd2759698
72. }
75. /**
76. * 使用JDK方式以SHA-256的方式实现消息摘要算法
77. * @throws Exception
78. */
79. @Test
80. public void testSHA256() throws Exception {
82. MessageDigest md = MessageDigest.getInstance(“SHA-256”);
83. md.update(targetString.getBytes());
84. String result = Hex.encodeHexString(md.digest());
85. System.out.println(result);
86. //8305809696717b31c6765b7ba89cfd67c17ef62c79a700ae4e305fee5ebdf457
88. }
91. /**
92. * 使用JDK方式以SHA-512的方式实现消息摘要算法
93. * @throws Exception
94. */
95. @Test
96. public void testSHA512() throws Exception {
98. MessageDigest md = MessageDigest.getInstance(“SHA-512”);
99. md.update(targetString.getBytes());
100. String result = Hex.encodeHexString(md.digest());
101. System.out.println(result);
102. //520abdfdc5e0ac43c795bee9da1cac3fa2f55b8e89f1d33f9a4b83367f4b74cd5d42bfa2c6d4f68c362e64d44ea664244c03c7fd2b7bc167a489fe7129c91156
104. }
106. }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79
• 80
• 81
• 82
• 83
• 84
• 85
• 86
• 87
• 88
• 89
• 90
• 91
• 92
• 93
• 94
• 95
• 96
• 97
• 98
• 99
• 100
• 101
• 102
• 103
• 104
• 105
• 106
• 107

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79
• 80
• 81
• 82
• 83
• 84
• 85
• 86
• 87
• 88
• 89
• 90
• 91
• 92
• 93
• 94
• 95
• 96
• 97
• 98
• 99
• 100
• 101
• 102
• 103
• 104
• 105
• 106
• 107

---

MAC方式

---

1. package MAC;
3. import javax.crypto.KeyGenerator;
4. import javax.crypto.Mac;
5. import javax.crypto.SecretKey;
6. import javax.crypto.spec.SecretKeySpec;
8. import org.apache.commons.codec.binary.Hex;
9. import org.junit.Test;
11. /**
12. * MAC: Message Authentication Code ——消息验证码
13. */
15. public class Demo {
17. private static String targetString = “I am Summer!”;
19. /**
20. * 使用MAC算法以MD5方式加密
21. * @throws Exception
22. */
23. @Test
24. public void hmacMD5() throws Exception {
25. //初始化KeyGeerator
26. KeyGenerator keyGenerator = KeyGenerator.getInstance(“HmacMD5”);
27. //产生密钥
28. SecretKey secretKey = keyGenerator.generateKey();
29. //获得密钥
30. byte[] key = secretKey.getEncoded();
32. key = Hex.decodeHex(new char[]{‘S’,‘u’,‘m’,‘m’,‘e’,‘r’,‘!’});
35. //还原密钥
36. SecretKey restoreSecretKey = new SecretKeySpec(key, “HmacMD5”);
37. //实例化Mac
38. Mac mac = Mac.getInstance(restoreSecretKey.getAlgorithm());
39. //初始化Mac
40. mac.init(restoreSecretKey);
41. byte[] hmacMD5Bytes = mac.doFinal(targetString.getBytes());
42. //转成十六进制并进行输出
43. String result = Hex.encodeHexString(hmacMD5Bytes);
44. System.out.println(result);
45. //8371828a9e53f04977b4d2ceb73ff506
46. }
49. }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50

---

小总结

---

对于MAC方式，其原理其实和前两个差不多，只不过是加入了key加密的权限。详见代码哦！