「javakafa」java卡方检验

本篇文章给大家谈谈javakafa，以及java卡方检验对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

• 1、kafka简介
• 2、kafka消费者java版本读取不到消息怎么办
• 3、3分钟带你彻底搞懂 Kafka
• 4、kafka是干嘛的
• 5、kafka——消费者原理解析
• 6、Java使用kafka发送消息没有生效

kafka简介

一、kafka定义

二、kafka的优势

三、kafka的原理

四、kafka起源

一、Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

二、kafka的优势

高吞吐量、低延迟：kafka美妙之处是可以处理几十万条信息，它的延迟最低只有几毫秒，每个topic可以分多个partition，consumer

group对partition进行consume操作。

可扩展性：kafka集群支持热扩展

持久化、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失

容错性：允许集群中节点失败（若副本数量为n，则允许n-1个节点失败）

高并发：支持数千个客户端同时读写

三、kafka的原理

kafka是如何实现以上所述这几点，我们逐一说明：

1.高吞吐量、低延迟

kafka在设计之初就是为了针对大数据量的传输处理，高吞吐量、低延迟最主要看的就是单位时间内所能读写的数据总量，我们先来看生产端。

kafka采取了一定量的批处理机制，即当生产数据达到一定数量或者达到时间窗口后，将所收集到的数据一批次的提交到服务器，我们假设处理一次数据的时间为1ms，那每秒钟能处理1000条，延时为1ms，如果此时将处理间隔变成9ms，即每10ms处理一批数据，假设这段时间接收到100条处理，那每秒则能处理10000条，但是延时变成了10ms。为了获得最大的吞吐量，需要牺牲一定的延迟，但是这样的牺牲是值得的。当确定了这种小批量方式之后，高速的写则取决于kafka自身写磁盘的速度了。而由于kafka本身对数据不做任何的处理，只管写入数据，保管数据，分发数据，因此会是一种批量顺序写入数据的情况，而磁盘的读写速度大量消耗在寻址上，也就是随机读写，但是对于顺序写入的速度是非常快的，甚至能媲美内存的随机写入速度。有人做过一个对比，普通磁盘顺序写入每秒能达到53.2M/s，SSD的顺序写入速度为42.2M/s，内存的顺序写入速度为358.2M/s。kafka正是利用了这个特性，顺序写入，速度相对较快。而kafka本身虽然也是写入磁盘持久化数据，但实际上kafka是将数据顺序写入页缓存中（page cache），然后由操作系统自行决定何时写到磁盘上，因此kafka的写操作能在每秒轻轻松松达到写入数十万条记录。并且基于kafka的动态扩展，这个数字还能不断增大。

kafka在消费端也有着高吞吐量，由于kafka是将数据写入到页缓存中，同时由于读写相间的间隔并不大，很大可能性会在缓存中命中，从而保证高吞吐量。另外kafka由于本身不对数据做任何的修改，完全使用零拷贝技术，大大提升数据的读取能力。

2.kafka每个节点叫做broker，而每一个broker都是独立运行的，可以随时加入kafka集群，集群的心跳管理是由zookeeper负责，新加入的broker只要broker id不与原有的冲突就能顺利的加入集群中，实现动态扩展。

3.kafka的持久化在上面已经提到，kafka绕过了java的堆处理数据，直接将数据写入页缓存，然后由操作系统来管理页缓存写入磁盘，实现持久化。kafka每一个主题topic是一个业务数据，他可由多个partition组成，而每个partition可以有多个replica副本，用于保证数据的可靠性。replica分为两个角色，一个是leader，一个是追随者，同一时间，每一个partition只能有一个leader，其他都是追问随者，laeder负责接收数据并写入log，而追随者不能被用户写入数据，只是从leader角色的replica副本中同步log写入自己的log，保持数据同步。kafka中有一个概念，ISR，全称是in-sync

replica，即所有可用的replica副本，这里的ISR数量只要大于1，这个partition就能正常运作，因此容错性非常好，假设n个replica，那最多可以坏n-1个replica的情况下，还能保持系统正常运行。当replica迟滞到一定时间后，会被kafka从ISR中剔除，当再次同步后，可以再次加入ISR，如果这时候leader出现问题，会从ISR中重新选举一个leader，原先的leader再次同步成功后会重新加入ISR，成为一个flower。

4.上面提到了kafka的ISR机制，kafka的容错性就是由ISR的机制来保证的。

5.kafka集群可以动态扩展broker，多个partition同时写入消费数据，实现真正的高并发。

四、kafka的起源

kafka起源于LinkedIn公司，当时领英公司需要收集两大类数据，一是业务系统和应用程序的性能监控指标数据，而是用户的操作行为数据。当时为了收集这两类数据，领英自研了两套相应的数据收集系统，但是这两套系统都存在一些弊端，无法实现实时交互、实时性差、维护成本高。因此领英的工程师希望找到一个统一的组件来收集分发消费这些大批量的数据，ActiveMQ由于扩展性不足，不能支撑大数据量而被抛弃，从而决定自研一套满足需求的系统组件，也就是kafka。

kafka的设计之初主要有三个目标：

1.为生产者和消费者提供一套简单的API

2.降低网络传输和磁盘存储开销

3.具有高伸缩性架构

目前kafka可以算是超额完成了目标。

kafka的名称由来也很有意思，因为kafka系统的写操作性能特别强，因此想使用一个作家的名字来命名kafka，而Jay Kreps，kafka的三位作者之一，在上大学的时候很喜欢Franz Kafka，因此起来这样一个名字。

kafka在2010年开源，2011年7月正式进入Apache进行孵化，2012年10月顺利毕业，后成为Apache的顶级项目。

kafka消费者java版本读取不到消息怎么办

3. 启动服务

3.1 启动zookeeper

启动zk有两种方式，第一种是使用kafka自己带的一个zk。

bin/zookeeper-server-start.sh config/zookeeper.properties

另一种是使用其它的zookeeper，可以位于本机也可以位于其它地址。这种情况需要修改config下面的sercer.properties里面的zookeeper地址

。例如zookeeper.connect=10.202.4.179:2181

3.2 启动 kafka

bin/kafka-server-start.sh config/server.properties

4.创建topic

bin/kafka-topics.sh --create --zookeeper 10.202.4.179:2181 --replication-factor 1 --partitions 1 --topic test

创建一个名为test的topic，只有一个副本，一个分区。

通过list命令查看刚刚创建的topic

bin/kafka-topics.sh -list -zookeeper 10.202.4.179:2181

5.启动producer并发送消息启动producer

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

启动之后就可以发送消息了

比如

test

hello boy

按Ctrl+C退出发送消息

6.启动consumer

bin/kafka-console-consumer.sh --zookeeper 10.202.4.179:2181 --topic test --from-beginning

启动consumer之后就可以在console中看到producer发送的消息了

可以开启两个终端，一个发送消息，一个接受消息。

如果这样都不行的话，查看zookeeper进程和kafka的topic，一步步排查原因吧。

3分钟带你彻底搞懂 Kafka

Kafka到底是个啥？用来干嘛的？

官方定义如下：

翻译过来，大致的意思就是，这是一个实时数据处理系统，可以横向扩展，并高可靠！

实时数据处理 ，从名字上看，很好理解，就是将数据进行实时处理，在现在流行的微服务开发中，最常用实时数据处理平台有 RabbitMQ、RocketMQ 等消息中间件。

这些中间件，最大的特点主要有两个：

在早期的 web 应用程序开发中，当请求量突然上来了时候，我们会将要处理的数据推送到一个队列通道中，然后另起一个线程来不断轮训拉取队列中的数据，从而加快程序的运行效率。

但是随着请求量不断的增大，并且队列通道的数据一致处于高负载，在这种情况下，应用程序的内存占用率会非常高，稍有不慎，会出现内存不足，造成程序内存溢出，从而导致服务不可用。

随着业务量的不断扩张，在一个应用程序内，使用这种模式已然无法满足需求，因此之后，就诞生了各种消息中间件，例如 ActiveMQ、RabbitMQ、RocketMQ等中间件。

采用这种模型，本质就是将要推送的数据，不在存放在当前应用程序的内存中，而是将数据存放到另一个专门负责数据处理的应用程序中，从而实现服务解耦。

消息中间件 ：主要的职责就是保证能接受到消息，并将消息存储到磁盘，即使其他服务都挂了，数据也不会丢失，同时还可以对数据消费情况做好监控工作。

应用程序 ：只需要将消息推送到消息中间件，然后启用一个线程来不断从消息中间件中拉取数据，进行消费确认即可！

引入消息中间件之后，整个服务开发会变得更加简单，各负其责。

Kafka 本质其实也是消息中间件的一种，Kafka 出自于 LinkedIn 公司，与 2010 年开源到 github。

LinkedIn 的开发团队，为了解决数据管道问题，起初采用了 ActiveMQ 来进行数据交换，大约是在 2010 年前后，那时的 ActiveMQ 还远远无法满足 LinkedIn 对数据传递系统的要求，经常由于各种缺陷而导致消息阻塞或者服务无法正常访问，为了能够解决这个问题，LinkedIn 决定研发自己的消息传递系统， Kafka 由此诞生 。

在 LinkedIn 公司，Kafka 可以有效地处理每天数十亿条消息的指标和用户活动跟踪，其强大的处理能力，已经被业界所认可，并成为大数据流水线的首选技术。

先来看一张图， 下面这张图就是 kafka 生产与消费的核心架构模型 ！

如果你看不懂这些概念没关系，我会带着大家一起梳理一遍！

简而言之，kafka 本质就是一个消息系统，与大多数的消息系统一样，主要的特点如下：

与 ActiveMQ、RabbitMQ、RocketMQ 不同的地方在于，它有一个**分区 Partition **的概念。

这个分区的意思就是说，如果你创建的 topic 有5个分区，当你一次性向 kafka 中推 1000 条数据时，这 1000 条数据默认会分配到 5 个分区中，其中每个分区存储 200 条数据。

这样做的目的，就是方便消费者从不同的分区拉取数据，假如你启动 5 个线程同时拉取数据，每个线程拉取一个分区，消费速度会非常非常快！

这是 kafka 与其他的消息系统最大的不同！

和其他的中间件一样，kafka 每次发送数据都是向 Leader 分区发送数据，并顺序写入到磁盘，然后 Leader 分区会将数据同步到各个从分区 Follower ，即使主分区挂了，也不会影响服务的正常运行。

那 kafka 是如何将数据写入到对应的分区呢？kafka中有以下几个原则：

与生产者一样，消费者主动的去kafka集群拉取消息时，也是从 Leader 分区去拉取数据。

这里我们需要重点了解一个名词： 消费组 ！

考虑到多个消费者的场景，kafka 在设计的时候，可以由多个消费者组成一个消费组，同一个消费组者的消费者可以消费同一个 topic 下不同分区的数据，同一个分区只会被一个消费组内的某个消费者所消费，防止出现重复消费的问题！

但是不同的组，可以消费同一个分区的数据！

你可以这样理解，一个消费组就是一个客户端，一个客户端可以由很多个消费者组成，以便加快消息的消费能力。

但是，如果一个组下的消费者数量大于分区数量，就会出现很多的消费者闲置。

如果分区数量大于一个组下的消费者数量，会出现一个消费者负责多个分区的消费，会出现消费性能不均衡的情况。

因此，在实际的应用中，建议消费者组的 consumer 的数量与 partition 的数量保持一致！

光说理论可没用，下面我们就以 centos7 为例，介绍一下 kafka 的安装和使用。

kafka 需要 zookeeper 来保存服务实例的元信息，因此在安装 kafka 之前，我们需要先安装 zookeeper。

zookeeper 安装环境依赖于 jdk，因此我们需要事先安装 jdk

下载zookeeper，并解压文件包

创建数据、日志目录

配置zookeeper

重新配置 dataDir 和 dataLogDir 的存储路径

最后，启动 Zookeeper 服务

到官网 下载想要的版本，我这里下载是最新稳定版 2.8.0 。

按需修改配置文件 server.properties （可选）

server.properties 文件内容如下：

其中有四个重要的参数：

可根据自己需求修改对应的配置！

启动 kafka 服务

创建一个名为 testTopic 的主题，它只包含一个分区，只有一个副本：

运行 list topic 命令，可以看到该主题。

输出内容：

Kafka 附带一个命令行客户端，它将从文件或标准输入中获取输入，并将其作为消息发送到 Kafka 集群。默认情况下，每行将作为单独的消息发送。

运行生产者，然后在控制台中键入一些消息以发送到服务器。

输入两条内容并回车：

Kafka 还有一个命令行使用者，它会将消息转储到标准输出。

输出结果如下：

本文主要围绕 kafka 的架构模型和安装环境做了一些初步的介绍，难免会有理解不对的地方，欢迎网友批评、吐槽。

由于篇幅原因，会在下期文章中详细介绍 java 环境下 kafka 应用场景！

kafka是干嘛的

Kafka是由Apache软件基金会开发的一个开源流处理平台，Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。这种动作是在现代网络上的许多社会功能的一个关键因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

kafka——消费者原理解析

kafka采用发布订阅模式：一对多。发布订阅模式又分两种：

Kafka为这两种模型提供了单一的消费者抽象模型： 消费者组 （consumer group）。 消费者用一个消费者组名标记自己。 一个发布在Topic上消息被分发给此消费者组中的一个消费者。 假如所有的消费者都在一个组中，那么这就变成了队列模型。 假如所有的消费者都在不同的组中，那么就完全变成了发布-订阅模型。 一个消费者组中消费者订阅同一个Topic，每个消费者接受Topic的一部分分区的消息，从而实现对消费者的横向扩展，对消息进行分流。

注意：当单个消费者无法跟上数据生成的速度，就可以增加更多的消费者分担负载，每个消费者只处理部分partition的消息，从而实现单个应用程序的横向伸缩。但是不要让消费者的数量多于partition的数量，此时多余的消费者会空闲。此外，Kafka还允许多个应用程序从同一个Topic读取所有的消息，此时只要保证每个应用程序有自己的消费者组即可。

消费者组的概念就是：当有多个应用程序都需要从Kafka获取消息时，让每个app对应一个消费者组，从而使每个应用程序都能获取一个或多个Topic的全部消息；在每个消费者组中，往消费者组中添加消费者来伸缩读取能力和处理能力，消费者组中的每个消费者只处理每个Topic的一部分的消息，每个消费者对应一个线程。

在同一个群组中，无法让一个线程运行多个消费者，也无法让多线线程安全地共享一个消费者。按照规则，一个消费者使用一个线程，如果要在同一个消费者组中运行多个消费者，需要让每个消费者运行在自己的线程中。最好把消费者的逻辑封装在自己的对象中，然后使用java的ExecutorService启动多个线程，使每个消费者运行在自己的线程上,可参考

一个 consumer group 中有多个 consumer，一个 topic 有多个 partition，所以必然会涉及到 partition 的分配问题，即确定哪个 partition 由哪个 consumer 来消费。

关于如何设置partition值需要考虑的因素

Kafka 有两种分配策略，一个是 RoundRobin，一个是 Range，默认为Range，当消费者组内消费者发生变化时，会触发分区分配策略（方法重新分配）。

以上三种现象会使partition的所有权在消费者之间转移，这样的行为叫作再均衡。

再均衡的优点 ：

再均衡的缺点 ：

RoundRobin 轮询方式将分区所有作为一个整体进行 Hash 排序，消费者组内分配分区个数最大差别为 1，是按照组来分的，可以解决多个消费者消费数据不均衡的问题。

但是，当消费者组内订阅不同主题时，可能造成消费混乱，如下图所示，Consumer0 订阅主题 A，Consumer1 订阅主题 B。

将 A、B 主题的分区排序后分配给消费者组，TopicB 分区中的数据可能 分配到 Consumer0 中。

Range 方式是按照主题来分的，不会产生轮询方式的消费混乱问题。

但是，如下图所示，Consumer0、Consumer1 同时订阅了主题 A 和 B，可能造成消息分配不对等问题，当消费者组内订阅的主题越多，分区分配可能越不均衡。

由于 consumer 在消费过程中可能会出现断电宕机等故障，consumer 恢复后，需要从故障前的位置继续消费，所以 consumer 需要实时记录自己消费到了哪个 offset，以便故障恢复后继续消费。

consumer group +topic + partition 唯一确定一个offest

Kafka 0.9 版本之前，consumer 默认将 offset 保存在 Zookeeper 中，从 0.9 版本开始，

consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中，该 topic 为__consumer_offsets。

你如果特别好奇，实在想看看offset什么的，也可以执行下面操作：

修改配置文件 consumer.properties

再启动一个消费者

当消费者崩溃或者有新的消费者加入，那么就会触发再均衡（rebalance），完成再均衡后，每个消费者可能会分配到新的分区，而不是之前处理那个，为了能够继续之前的工作，消费者需要读取每个partition最后一次提交的偏移量，然后从偏移量指定的地方继续处理。

case1：如果提交的偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理。

case2：如果提交的偏移量大于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。

自动提交的优点是方便，但是可能会重复处理消息

不足：broker在对提交请求作出回应之前，应用程序会一直阻塞，会限制应用程序的吞吐量。

因此，在消费者关闭之前一般会组合使用commitAsync和commitSync提交偏移量。

ConsumerRebalanceListener需要实现的两个方法

下面的例子演示如何在失去partition的所有权之前通过onPartitionRevoked()方法来提交偏移量。

Consumer有个Rebalance的特性，即重新负载均衡，该特性依赖于一个协调器来实现。每当Consumer Group中有Consumer退出或有新的Consumer加入都会触发Rebalance。

之所以要重新负载均衡，是为了将退出的Consumer所负责处理的数据再重新分配到组内的其他Consumer上进行处理。或当有新加入的Consumer时，将组内其他Consumer的负载压力，重新进均匀分配，而不会说新加入一个Consumer就闲在那。

下面就用几张图简单描述一下，各种情况触发Rebalance时，组内成员是如何与协调器进行交互的。

Tips ：图中的Coordinator是协调器，而generation则类似于乐观锁中的版本号，每当成员入组成功就会更新，也是起到一个并发控制的作用。

参考：

Java使用kafka发送消息没有生效

一般消息发不出去很大可能都是配置或环境的问题

1、排查环境是否有问题，zookeeper节点是否存活，kafka节点是否存活，通过命令行的方式能否发出去消息（使用kafka-console-producer.sh）,如果通过命令行都发不出去那就是集群的问题了。

2、网络问题，调用机器和集群之间网络是否通畅

3、调用时配置的host、port和集群中配置的是否一致，是否需要使用主机名而不是ip

4、客户端api版本是否和服务端差别太大导致不兼容

5、防火墙问题，关闭集群的防火墙实时

诸如此类，可能性太多就不一 一列举了。

你这既然有打印堆栈，如果报错肯定有异常信息的，可能卡住的时间比较长，耐心等待吧，祝你早日解决bug。

javakafa的介绍就聊到这里吧，感谢你花时间阅读本站内容，更多关于java卡方检验、javakafa的信息别忘了在本站进行查找喔。