带你逆袭kafka之路
1. kafka概述
1.1 kafka简介
Apache Kafka 是一个快速、可扩展的、高吞吐的、可容错的分布式“发布-订阅”消息系统， 使用 Scala 与 Java 语言编写，可以将消息从一个端点传递到另外一个端点，较之传统的消息中 间件(例如 ActiveMQ、RabbitMQ)，Kafka 具备高吞吐量、内置分区、支持消息副本和高容 错的特性，很是适合大规模消息处理应用程序。java
Kafka 官网: http://kafka.apache.org/算法
Kafka主要设计目标以下：spring
- 以时间复杂度为O(1)的方式提供消息持久化能力，即便对TB级以上数据也能保证常数时间的访问性能。
- 高吞吐率。即便在很是廉价的商用机器上也能作到单机支持每秒100K条消息的传输。
- 支持Kafka Server间的消息分区，及分布式消费，同时保证每一个partition内的消息顺序传输。
- 同时支持离线数据处理和实时数据处理。
- 支持在线水平扩展
Kafka一般用于两大类应用程序：数据库
- 创建实时流数据管道，以可靠地在系统或应用程序之间获取数据
- 构建实时流应用程序，以转换或响应数据流
要了解Kafka如何执行这些操做，让咱们从头开始深刻研究Kafka的功能。apache
首先几个概念：bootstrap
- Kafka在一个或多个能够跨越多个数据中心的服务器上做为集群运行。
- Kafka集群将记录流存储在称为主题的类别中。
- 每一个记录由一个键，一个值和一个时间戳组成
1.2 kafka架构体系
1.3 kafka的应用场景
kafka的应用场景很是多, 下面咱们就来举几个咱们最多见的场景api
1.3.1 用户的活动跟踪
用户在网站的不一样活动消息发布到不一样的主题中心，而后能够对这些消息进行实时监测、实时处理。固然，也能够加载到Hadoop或离线处理数据仓库，对用户进行画像。像淘宝、天猫、京东这些大型电商平台，用户的全部活动都要进行追踪的。安全
1.3.2 日志收集
1.3.3 限流削峰
1.3.4 高吞吐率实现
Kafka与其余MQ相比，最大的特色就是高吞吐率。为了增长存储能力，Kafka将全部的消息都写入到了低速大容量的硬盘。按理说，这将致使性能损失，但实际上，Kafka仍然能够保持超高的吞吐率，而且其性能并未受到影响。其主要采用以下方式实现了高吞吐率。springboot
- 顺序读写：Kafka将消息写入到了分区partition中，而分区中的消息又是顺序读写的。顺序读写要快于随机读写。
- 零拷贝：生产者、消费者对于Kafka中的消息是采用零拷贝实现的。
- 批量发送：Kafka容许批量发送模式。
- 消息压缩：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参数的值进行设置。
- 0值
异步发送。生产者向 kafka 发送消息而不须要 kafka 反馈成功 ack。该方式效率最高，但可靠性最低。其可能会存在消息丢失的状况。
- 在传输过程当中会出现消息丢失。
- 在broker内部会出现消息丢失。
- 会出现写入到kafka中的消息的顺序与生产顺序不一致的状况。
- 1值
同步发送。生产者发送消息给 kafka，broker 的 partition leader 在收到消息后立刻发送 成功 ack(无需等等 ISR 中的 Follower 同步)，生产者收到后知道消息发送成功，而后会再发送消息。若是一直未收到 kafka 的 ack，则生产者会认为消息发送失败，会重发消息。
该方式对于 Producer 来讲，若没有收到 ACK，必定能够确认消息发送失败了，而后能够 重发;可是，即便收到了 ACK，也不能保证消息必定就发送成功了。故，这种状况，也可能 会发生消息丢失的状况。
- -1值
同步发送。生产者发送消息给 kafka，kafka 收到消息后要等到 ISR 列表中的全部副本都 同步消息完成后，才向生产者发送成功 ack。若是一直未收到 kafka 的 ack，则认为消息发送 失败，会自动重发消息。该方式会出现消息重复接收的状况。
3.5 消费者消费过程解析
生产者将消息发送到topitc中, 消费者便可对其进行消费, 其消费过程以下:
- consumer向broker提交链接请求，其所链接上的broker都会向其发送broker controller的通讯URL，即配置文件中的listeners地址；
- 当consumer指定了要消费的topic后，会向broker controller发送消费请求；
- broker controller会为consumer分配一个或几个partition leader，并将该partition的当前offset发送给consumer；
- consumer会按照broker controller分配的partition对其中的消息进行消费；
- 当consumer消费完该条消息后，consumer会向broker发送一个消息已经被消费反馈，即该消息的offset；
- 在broker接收到consumer的offset后，会更新相应的__consumer_offset中；
- 以上过程会一直重复，知道消费者中止请求消费；
- Consumer能够重置offset，从而能够灵活消费存储在broker上的消息。
3.6 Partition Leader选举范围
当leader宕机后，broker controller会从ISR中挑选一个follower成为新的leader。若是ISR中没有其余副本怎么办？能够经过unclean.leader.election.enable的值来设置leader选举范围。
- false
必须等到ISR列表中全部的副本都活过来才进行新的选举。该策略可靠性有保证，但可用性低。
-
true
在ISR列表中没有副本的状况下，能够选择任意一个没有宕机的主机做为新的leader，该策略可用性高，但可靠性没有保证。
3.7 重复消费问题的解决方案
- 同一个consumer重复消费
当Consumer因为消费能力低而引起了消费超时，则可能会造成重复消费。
在某数据恰好消费完毕，可是正准备提交offset时候，消费时间超时，则broker认为这条消息未消费成功。这时就会产生重复消费问题。
其解决方案：延长offset提交时间。
- 不一样的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.下载并解压
cd /usr/local/src
wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.4.0/kafka_2.11-2.4.0.tgz
mkdir /data/servers
tar xzvf kafka_2.11-2.4.0.tgz -C /data/servers/
cd /data/servers/kafka_2.11-2.4.0
2.修改配置文件
kafka的配置文件$KAFKA_HOME/config/server.properties, 主要修改一下下面几项
确保每一个机器上的id不同 broker.id=0 配置服务端的监控地址 listeners=PLAINTEXT://192.168.51.128:9092 kafka 日志目录 log.dirs=/data/servers/kafka_2.11-2.4.0/logs #kafka设置的partitons的个数 num.partitions=1 zookeeper的链接地址, 若是有本身的zookeeper集群, 请直接使用本身搭建的zookeeper集群 zookeeper.connect=192.168.51.128:2181
由于我本身是本机作实验, 全部使用的是一个主机的不一样端口, 在线上, 就是不一样的机器,你们参考便可
咱们这里使用kafka的zookeeper, 只启动一个节点, 可是正真的生产过程当中, 是须要zookeeper集群, 本身搭建就好, 后期咱们也会出zookeeper的教程, 你们请关注就行了.
3.拷贝3份配置文件
#建立对应的日志目录 mkdir -p /data/servers/kafka_2.11-2.4.0/logs/9092 mkdir -p /data/servers/kafka_2.11-2.4.0/logs/9093 mkdir -p /data/servers/kafka_2.11-2.4.0/logs/9094 #拷贝三份配置文件 cp server.properties server_9092.properties cp server.properties server_9093.properties cp server.properties server_9094.properties
修改不一样端口对应的文件
#9092的id为0, 9093的id为1, 9094的id为2 broker.id=0 # 配置服务端的监控地址, 分别在不通的配置文件中写入不一样的端口 listeners=PLAINTEXT://192.168.51.128:9092 # kafka 日志目录, 目录也是对应不一样的端口 log.dirs=/data/servers/kafka_2.11-2.4.0/logs/9092 # kafka设置的partitons的个数 num.partitions=1 # zookeeper的链接地址, 若是有本身的zookeeper集群, 请直接使用本身搭建的zookeeper集群 zookeeper.connect=192.168.51.128:2181
修改zookeeper的配置文件
dataDir=/data/servers/zookeeper server.1=192.168.51.128:2888:3888
而后建立zookeeper的myid文件
echo "1"> /data/servers/zookeeper/myid
5.启动zookeeper
使用kafka内置的zookeeper
cd /data/servers/kafka_2.11-2.4.0/bin zookeeper-server-start.sh -daemon ../config/zookeeper.properties netstat -anp |grep 2181
启动kafka
./kafka-server-start.sh -daemon ../config/server_9092.properties ./kafka-server-start.sh -daemon ../config/server_9093.properties ./kafka-server-start.sh -daemon ../config/server_9094.properties
4.2 kafka的操做
- 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
to. In case of providing this, a direct Zookeeper connection won't be 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的分区的个数
示例:
cd /data/servers/kafka_2.11-2.4.0/bin # 建立topic test1 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 # 建立topic test2 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 # 查看topic 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
4. 发送消息
他们能够经过客户端的命令生产消息
先来看看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
- 消费消息
咱们也仍是先来看看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.消费者自动提交:
定义本身的生产者
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Properties;
/**
* @ClassName MyKafkaProducer
* @Description TODO
* @Author lingxiangxiang
* @Date 3:37 PM
* @Version 1.0
**/
public class MyKafkaProducer {
private org.apache.kafka.clients.producer.KafkaProducer<Integer, String> producer;
public MyKafkaProducer() {
Properties properties = new Properties();
properties.put("bootstrap.servers", "192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094");
properties.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 设置批量发送
properties.put("batch.size", 16384);
// 批量发送的等待时间50ms, 超过50ms, 不足批量大小也发送
properties.put("linger.ms", 50);
this.producer = new org.apache.kafka.clients.producer.KafkaProducer<Integer, String>(properties);
}
public boolean sendMsg() {
boolean result = true;
try {
// 正常发送, test2是topic, 0表明的是分区, 1表明的是key, hello world是发送的消息内容
final ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>("test2", 0, 1, "hello world");
producer.send(record);
// 有回调函数的调用
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
System.out.println(recordMetadata.topic());
System.out.println(recordMetadata.partition());
System.out.println(recordMetadata.offset());
}
});
// 本身定义一个类
producer.send(record, new MyCallback(record));
} catch (Exception e) {
result = false;
}
return result;
}
}
定义生产者发送成功的回调函数
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
* @ClassName MyCallback
* @Description TODO
* @Author lingxiangxiang
* @Date 3:51 PM
* @Version 1.0
**/
public class MyCallback implements Callback {
private Object msg;
public MyCallback(Object msg) {
this.msg = msg;
}
@Override
public void onCompletion(RecordMetadata metadata, Exception e) {
System.out.println("topic = " + metadata.topic());
System.out.println("partiton = " + metadata.partition());
System.out.println("offset = " + metadata.offset());
System.out.println(msg);
}
}
生产者测试类:
在生产者测试类中,本身遇到一个坑, 就是最后本身没有加sleep, 就是怎么检查本身的代码都没有问题, 可是最后就是无法发送成功消息, 最后加了一个sleep就能够了, 由于主函数main已经执行完退出, 可是消息并无发送完成, 须要进行等待一下.固然, 你在生产环境中可能不会遇到这样问题, 呵呵, 代码以下
import static java.lang.Thread.sleep;
/**
* @ClassName MyKafkaProducerTest
* @Description TODO
* @Author lingxiangxiang
* @Date 3:46 PM
* @Version 1.0
**/
public class MyKafkaProducerTest {
public static void main(String[] args) throws InterruptedException {
MyKafkaProducer producer = new MyKafkaProducer();
boolean result = producer.sendMsg();
System.out.println("send msg " + result);
sleep(1000);
}
}
消费者类:
import kafka.utils.ShutdownableThread;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
/**
* @ClassName MyKafkaConsumer
* @Description TODO
* @Author lingxiangxiang
* @Date 4:12 PM
* @Version 1.0
**/
public class MyKafkaConsumer extends ShutdownableThread {
private KafkaConsumer<Integer, String> consumer;
public MyKafkaConsumer() {
super("KafkaConsumerTest", false);
Properties properties = new Properties();
properties.put("bootstrap.servers", "192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094");
properties.put("group.id", "mygroup");
properties.put("enable.auto.commit", "true");
properties.put("auto.commit.interval.ms", "1000");
properties.put("session.timeout.ms", "30000");
properties.put("heartbeat.interval.ms", "10000");
properties.put("auto.offset.reset", "earliest");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
this.consumer = new KafkaConsumer<Integer, String>(properties);
}
@Override
public void doWork() {
consumer.subscribe(Arrays.asList("test2"));
ConsumerRecords<Integer, String>records = consumer.poll(1000);
for (ConsumerRecord record : records) {
System.out.println("topic = " + record.topic());
System.out.println("partition = " + record.partition());
System.out.println("key = " + record.key());
System.out.println("value = " + record.value());
}
}
}
消费者的测试类:
/**
* @ClassName MyConsumerTest
* @Description TODO
* @Author lingxiangxiang
* @Date 4:23 PM
* @Version 1.0
**/
public class MyConsumerTest {
public static void main(String[] args) {
MyKafkaConsumer consumer = new MyKafkaConsumer();
consumer.start();
System.out.println("==================");
}
}
2. 消费者同步手动提交
前面的消费者都是以自动提交 offset 的方式对 broker 中的消息进行消费的，但自动提交 可能会出现消息重复消费的状况。因此在生产环境下，不少时候须要对 offset 进行手动提交， 以解决重复消费的问题。
手动提交又能够划分为同步提交、异步提交，同异步联合提交。这些提交方式仅仅是 doWork()方法不相同，其构造器是相同的。因此下面首先在前面消费者类的基础上进行构造 器的修改，而后再分别实现三种不一样的提交方式。
- 同步提交方式是，消费者向 broker 提交 offset 后等待 broker 成功响应。若没有收到响 应，则会从新提交，直到获取到响应。而在这个等待过程当中，消费者是阻塞的。其严重影响 了消费者的吞吐量。
修改前面的MyKafkaConsumer.java, 主要修改下面的配置
import kafka.utils.ShutdownableThread;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
/**
* @ClassName MyKafkaConsumer
* @Description TODO
* @Author lingxiangxiang
* @Date 4:12 PM
* @Version 1.0
**/
public class MyKafkaConsumer extends ShutdownableThread {
private KafkaConsumer<Integer, String> consumer;
public MyKafkaConsumer() {
super("KafkaConsumerTest", false);
Properties properties = new Properties();
properties.put("bootstrap.servers", "192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094");
properties.put("group.id", "mygroup");
// 这里要修改为手动提交
properties.put("enable.auto.commit", "false");
// properties.put("auto.commit.interval.ms", "1000");
properties.put("session.timeout.ms", "30000");
properties.put("heartbeat.interval.ms", "10000");
properties.put("auto.offset.reset", "earliest");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
this.consumer = new KafkaConsumer<Integer, String>(properties);
}
@Override
public void doWork() {
consumer.subscribe(Arrays.asList("test2"));
ConsumerRecords<Integer, String>records = consumer.poll(1000);
for (ConsumerRecord record : records) {
System.out.println("topic = " + record.topic());
System.out.println("partition = " + record.partition());
System.out.println("key = " + record.key());
System.out.println("value = " + record.value());
//手动同步提交
consumer.commitSync();
}
}
}
3. 消费者异步手工提交
手动同步提交方式须要等待 broker 的成功响应，效率过低，影响消费者的吞吐量。异步提交方式是，消费者向 broker 提交 offset 后不用等待成功响应，因此其增长了消费者的吞吐量。
import kafka.utils.ShutdownableThread;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
/**
* @ClassName MyKafkaConsumer
* @Description TODO
* @Author lingxiangxiang
* @Date 4:12 PM
* @Version 1.0
**/
public class MyKafkaConsumer extends ShutdownableThread {
private KafkaConsumer<Integer, String> consumer;
public MyKafkaConsumer() {
super("KafkaConsumerTest", false);
Properties properties = new Properties();
properties.put("bootstrap.servers", "192.168.51.128:9092,192.168.51.128:9093,192.168.51.128:9094");
properties.put("group.id", "mygroup");
// 这里要修改为手动提交
properties.put("enable.auto.commit", "false");
// properties.put("auto.commit.interval.ms", "1000");
properties.put("session.timeout.ms", "30000");
properties.put("heartbeat.interval.ms", "10000");
properties.put("auto.offset.reset", "earliest");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
this.consumer = new KafkaConsumer<Integer, String>(properties);
}
@Override
public void doWork() {
consumer.subscribe(Arrays.asList("test2"));
ConsumerRecords<Integer, String>records = consumer.poll(1000);
for (ConsumerRecord record : records) {
System.out.println("topic = " + record.topic());
System.out.println("partition = " + record.partition());
System.out.println("key = " + record.key());
System.out.println("value = " + record.value());
//手动同步提交
// consumer.commitSync();
//手动异步提交
// consumer.commitAsync();
// 带回调公共的手动异步提交
consumer.commitAsync((offsets, e) -> {
if(e != null) {
System.out.println("提交次数, offsets = " + offsets);
System.out.println("exception = " + e);
}
});
}
}
}
5.2 springboot使用kafka
如今你们的开发过程当中, 不少都用的是springboot的项目, 直接启动了, 若是仍是用原生的API, 就是有点low了啊, 那kafka是如何和springboot进行联合的呢?
- maven配置
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.1.1</version>
</dependency>
- 添加配置文件
在application.properties中加入以下配置信息:
kafka 链接地址
spring.kafka.bootstrap-servers = 192.168.51.128:9092,10.231.128.96:9093,192.168.51.128:9094
生产者
spring.kafka.producer.acks = 0 spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.retries = 3 spring.kafka.producer.batch-size = 4096 spring.kafka.producer.buffer-memory = 33554432 spring.kafka.producer.compression-type = gzip
消费者
spring.kafka.consumer.group-id = mygroup spring.kafka.consumer.auto-commit-interval = 5000 spring.kafka.consumer.heartbeat-interval = 3000 spring.kafka.consumer.key-deserializer = org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.auto-offset-reset = earliest spring.kafka.consumer.enable-auto-commit = true # listenner, 标识消费者监听的个数 spring.kafka.listener.concurrency = 8 # topic的名字 kafka.topic1 = topic1
生产者
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
@Service
@Slf4j
public class MyKafkaProducerServiceImpl implements MyKafkaProducerService {
@Resource
private KafkaTemplate<String, String> kafkaTemplate;
// 读取配置文件
@Value("${kafka.topic1}")
private String topic;
@Override
public void sendKafka() {
kafkaTemplate.send(topic, "hell world");
}
}
消费者
@Component
@Slf4j
public class MyKafkaConsumer {
@KafkaListener(topics = "${kafka.topic1}")
public void listen(ConsumerRecord<?, ?> record) {
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
if (kafkaMessage.isPresent()) {
log.info("----------------- record =" + record);
log.info("------------------ message =" + kafkaMessage.get());
}
- 1. 逆袭之路
- 2. 小白的逆袭之路
- 3. 背锅侠逆袭之路
- 4. 编程小白的逆袭之路
- 5. 一个小白的逆袭之路
- 6. 一个学渣的 CTO 逆袭之路
- 7. 零基础学员逆袭之路
- 8. 反射型XSS的逆袭之路
- 9. 土八路的逆袭——死或生
- 10. 从0到1的逆袭之路，Python改变你的学习方式
- 更多相关文章...
- • Eclipse Java 构建路径 - Eclipse 教程
- • Spring JDK动态代理（附带实例） - Spring教程
- • 再有人问你分布式事务，把这篇扔给他
- • 适用于PHP初学者的学习线路和建议
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 逆袭之路
- 2. 小白的逆袭之路
- 3. 背锅侠逆袭之路
- 4. 编程小白的逆袭之路
- 5. 一个小白的逆袭之路
- 6. 一个学渣的 CTO 逆袭之路
- 7. 零基础学员逆袭之路
- 8. 反射型XSS的逆袭之路
- 9. 土八路的逆袭——死或生
- 10. 从0到1的逆袭之路，Python改变你的学习方式