连阿里都在用它处理亿万级数据统计，论其对Java程序员的重要性！

一．了解淘宝Kafka架构

在ActiveMQ、RabbitMQ、RocketMQ、Kafka消息中间件之间，咱们为何要选择Kafka?下面详细介绍一下，2012年9月份我在支付宝作余额宝研发，2013年6月支付宝正式推出余额宝，2013年8月担任支付宝淘宝彩票项目经理带领兄弟们一块儿作研发，期间须要与淘宝和500万对接竞彩接口数据，业余时间与淘宝的同事沟通，了解天猫在电商节如何处理这些大数据的？技术架构上采用了哪些策略呢？

1、应用无状态(淘宝session框架)

2、有效使用缓存(Tair)

3、应用拆分(HSF)

4、数据库拆分(TDDL)

5、异步通讯(Notify)

6、非结构化数据存储 ( TFS,NOSQL)

7、监控、预警系统

8、配置统一管理

天猫的同事把大体的架构跟我描述了一番，心有感悟。我们来看一下2018年双11当天的成交额。

 

二．kafka实现天猫亿万级数据统计架构

 

Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统，Flume支持在日志系统中定制各种数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各类数据接受方（可定制）的能力

• Data Access：数据通道
• Computing：计算
• Persistence：执行保存方式
• spout：表示一个流的源头，产生tuple
• bolt：处理输入流并产生多个输出流，能够作简单的数据转换计算，复杂的流处理通常须要通过多个bolt进行处理

为何不能用分布式文件HDFS集群？

一、实时性：hdfs的实时性没有kafka高。

二、消费量的记录：hdfs不会记录你这个块文件消费到了哪里，而基于zookeeper的kafka会记录你消费的点。

三、并发消费：hdfs不支持并发消费，而kafka支持并发消费，即多个consumer.

四、弹性且有序：当数据量会很大，并且处理完以后就能够删除时，频繁的读写会对hdfs中NameNode形成很大的压力。而kafka的消费点是记录在zookeeper的，而且kafka的每条数据都是有“坐标”的，因此消费的时候只要这个“坐标”向后移动就好了，并且删除的时候只要把这个“坐标”以前的数据删掉便可。

三．什么是Kafka?

 

经过上图就能够了解到，生产者Producers(农民和厨师)，消费主题top(鱼，骨头，草，香蕉),消费者Comsumer(猫，狗，老牛，猴子)，生产者根据消费主题获取本身想要的食物

四．Kafka架构原理

 

五．Kafka能帮咱们解决什么问题？

请高手指明一下kafka解决了什么问题，什么场景下使用？消息订阅和发布吗，好像redis也支持，功能是否有重叠？

一．消息队列

假设你意气风发，要开发新一代的互联网应用，以期在互联网事业中一展宏图。借助云计算，很容易开发出以下原型系统：

• Web应用：部署在云服务器上，为我的电脑或者移动用户提供的访问体验。
• SQL数据库：为Web应用提供数据持久化以及数据查询。

 

这套架构简洁而高效，很快可以部署到百度云等云计算平台，以便快速推向市场。互联网不就是讲究小步快跑嘛！

好景不长。随着用户的迅速增加，全部的访问都直接经过SQL数据库使得它不堪重负，不得不加上缓存服务以下降SQL数据库的荷载；为了理解用户行为，开始收集日志并保存到Hadoop上离线处理，同时把日志放在全文检索系统中以便快速定位问题；因为须要给投资方看业务情况，也须要把数据汇总到数据仓库中以便提供交互式报表。此时的系统的架构已经盘根错节了，考虑未来还会加入实时模块以及外部数据交互，真是痛并快乐着……

 

这时候，应该跑慢一些，让灵魂跟上来。

本质上，这是一个数据集成问题。没有任何一个系统可以解决全部的事情，因此业务数据根据不一样用途存而放在不一样的系统，好比归档、分析、搜索、缓存等。数据冗余自己没有任何问题，可是不一样系统之间像意大利面条同样复杂的数据同步倒是挑战。

这时候就轮到Kafka出场了。

Kafka可让合适的数据以合适的形式出如今合适的地方。Kafka的作法是提供消息队列，让生产者单往队列的末尾添加数据，让多个消费者从队列里面依次读取数据而后自行处理。以前链接的复杂度是O(N^2)，而如今下降到O(N)，扩展起来方便多了：

 

在Kafka的帮助下，你的互联网应用终于可以支撑飞速增加的业务，成为下一个BAT指日可待。

以上故事说明了Kafka主要用途是数据集成，或者说是流数据集成，以Pub/Sub形式的消息总线形式提供。可是，Kafka不只仅是一套传统的消息总线，本质上Kafka是分布式的流数据平台，由于如下特性而著名：

• 提供Pub/Sub方式的海量消息处理。
• 以高容错的方式存储海量数据流。
• 保证数据流的顺序。

二．日志采集

随着互联网的不断发展，用户所产生的行为数据被愈来愈多的网站重视，如何对于用户信息进行采集则愈来愈受到重视，下面就为你们介绍基于Kafka的服务端用户行为日志采集方式。

1. 技术选型

服务端日志采集主要经过在Controller的接口中进行埋点，而后经过AOP技术、Kafka消息系统以及logback对用户行为进行采集。

之因此使用AOP技术是由于AOP的如下重要特定：

• 代码的侵入性小。对于业务代码的侵入性小，只须要在Controller的接口上添加注解，而后在其余模块对用户行为进行采集。
• 重用性。对于相同做用的代码能够进行重用。
• 扩展性。可以很好的对系统进行扩展。

因为使用异步方式对用户行为信息进行收集，所以须要使用消息中间件。目前消息中间件很是多，比较流行的有ActiveMQ、ZeroMQ、RabbitMQ、Kafka等。每一个消息中间件都有各类的优点劣势，之因此使用Kafka消息中间件，是由于如下几点因素：

• 高性能。每秒钟能够处理数以千计生产者生成的消息。
• 高扩展性。能够经过简单的增长服务器横向扩展Kafka集群的容量。
• 分布式。消息来自数以千计的服务，使用分布式来解决单机处理海量数据的瓶颈。
• 持久性。Kafka中的消息能够持久化到硬盘上，这样能够防止数据的丢失。

由于用户的行为数据最终是以日志的形式持久化的，所以使用logback对日志持久化到日志服务器中。

2.整体架构

图1 整体架构图

服务端日志采集系统主要由两个工程组成：陆金所-bi-core和lu-bi-service。因为中国平安陆金所使用dubbo框架，所以有服务提供方和服务消费方。lu-bi-core被web、wap和mainsite服务消费方依赖。此外，lu-bi-service也依赖于lu-bi-core，主要是依赖于其中的一些实体类及工具类。

lu-bi-core工程为Kafka消息的生产者，主要封装实现切面的具体逻辑，其主要职责以下：

• 解析用户请求的Request信息：从Request中提取用户的基本信息，如设备型号、用户的供应商、ip、设备的分辨率、设备平台、设备的操做系统、设备id、app渠道等。
• 接口对应的参数：经过切面能够提取接口的参数值，从而知道用户的业务信息。
• 应用层返回的结果信息：由于切面使用AfterReturning方式，所以能够获取用层的返回结果，从返回结果中能够提取有用的信息。
• 用户的基本信息：用户的id信息。
• 信息格式化：将信息转化成JSON字符串。
• 发送消息：将最终须要发送的消息放入本地阻塞队列中，经过另外一个线程异步从阻塞队列中获取消息并发送到Kafka Broker中。

lu-bi-service工程为Kafka消息的消费者，其主要职责以下：

• 实时从Kafka中拉取最新的数据。
• 将JSON字符串转化成，方便进一步对用信息进行加工。
• 对用户的ip进行解析，获取ip对应的地区以及经纬度信息。
• 将加工好的最终信息持久化到log文件中。

3.部署图

图2 部署图

上图为陆金所与日志系统系统相关的部署图，App、Wap和Mainsite服务器集群分别对应不一样终端的应用。Kafka集群使用杭研的集群，目前有10个Broker。日志服务器有两台，经过Kafka的均衡策略对日志进行消费。

4.日志采集的流程

日志采集流程图以下所示：

图3 日志打点流程图

上图为消息生产者和消息消费者共同组成的流程图。

• 消息生产者的具体步骤以下：
• 经过切面拦截用户的请求。
• 从切面中提取请求头的基本信息，如设备信息，cookie信息，ip信息等。
• 提取请求的接口参数信息。
• 从接口返回值中提取相关信息，如id，pvid等。
• 将提取的信息封装成JSON字符串，放到阻塞队列中，假如阻塞队列溢出会有三次重试机制。
• 异步线程从本地阻塞队列中获取数据，并将信息组装发送到Kafka的Broker中，此时消息生产者结束。

消息消费者的具体步骤以下：

• 实时从Kafka Broker中批量拉取消息。
• 将拉取的消息转化成对象。
• 解析ip对应的国家、省份、城市、经纬度信息。
• 对不一样业务场景的信息进一步解析。
• 将日志信息转化成JSON字符串，持久化到log文件中。

5. 相关配置

• application-XXX.properties：该配置放Kafka的相关属性，包括topic、groupId、server等信息。
• lu-log-msg.xml：该配置放在app-web，mainsite-web，wap-web的src/main/resources目录下，主要是初始化kafka生产者的信息。
• lu-bi-service.xml：该配置放在lu-bi-service工程的src/main/resources目录下，主要用于加载kafka消费者的配置信息，而且启动kafka消费者服务。
• logback.xml：该配置放在lu-bi-service工程的src/main/resources目录下，主要用于声明日志文件存放的目录，须要持久化的日志的package路径，以及日志持久化的格式。
• ip_conf.txt：该配置放在lu-bi-service工程的src/main/resources目录下，用于解析ip对应的地域、经纬度等信息。

六．关于面试问题

1.Redis和Kafka区别？

做者跟你们举个例子：

老板有个好消息要告诉你们，公司要发放年终奖，有两个办法:

1.到会议室每一个座位上挨个儿告诉每一个人。什么？张三去上厕所了？那张三就只能错过好消息了！

2.老板把消息写到会议上的黑板报上，谁想知道就来看一下，什么？张三请假了？不要紧，我一周以后才擦掉，总会看见的！什么张三请假两周？那就算了，我反正只保留一周，否则其余好消息没地方写了

redis用第一种办法，kafka用第二种办法，知道什么区别了吧

Redis PUB/SUB使用场景：

1. 消息持久性需求不高

2. 吞吐量要求不高

3. 能够忍受数据丢失

4. 数据量不大

Kafka使用场景：

上面之外的其余场景：）

1. 高可靠性

2. 高吞吐量

3. 持久性高

Kafka、RabbitMQ、RocketMQ等消息中间件的对比

有关测试结论

Kafka的吞吐量高达17.3w/s，不愧是高吞吐量消息中间件的行业老大。这主要取决于它的队列模式保证了写磁盘的过程是线性IO。此时broker磁盘IO已达瓶颈。

RocketMQ也表现不俗，吞吐量在11.6w/s，磁盘IO %util已接近100%。RocketMQ的消息写入内存后即返回ack，由单独的线程专门作刷盘的操做，全部的消息均是顺序写文件。

RabbitMQ的吞吐量5.95w/s，CPU资源消耗较高。它支持AMQP协议，实现很是重量级，为了保证消息的可靠性在吞吐量上作了取舍。咱们还作了RabbitMQ在消息持久化场景下的性能测试，吞吐量在2.6w/s左右。

在服务端处理同步发送的性能上，Kafka>RocketMQ>RabbitMQ

写在最后

现在都在谈论寒冬有多可怕，笔者做为一个过来人，却有不一样的见解：寒冬不可怕，在寒冬里没有生存能力，才是最可怕的。

所以小编总结了这几年在阿里的工做经验并结合目前互联网最主流的Java架构技术，最后录制了七大Java架构技术专题视频（源码阅读、分布式架构、微服务、性能优化、阿里项目实战、Devops、并发编程）分享在个人裙537775426中，而且每晚我都会在群内直播讲解这些架构技术的底层实现原理，感兴趣的程序员们能够加群找管理员获取。