详细安装访问： >
在kafka的 config 目录下找到 serverproperties 配置文件
把 listeners 和 advertisedlisteners 两处配置的注释去掉，可以根据需要配置连接的服务器 外网IP 和 端口号 ，我这里演示选择的是本地 localhost 和默认端口 9092
KafkaTemplate 这个类包装了个生产者 Producer ，来提供方便的发送数据到 kafka 的主题 topic 里面。
send() 方法的源码， KafkaTemplate 类中还重载了很多 send() 方法，有需要可以看看源码
通过 KafkaTemplate 模板类发送数据。
kafkaTemplatesend(String topic, K key, V data) ,第一个入参是主题，第二个入参是发送的对象，第三个入参是发送的数据。通过 @KafkaListener 注解配置用户监听 topics
bootstrap-servers ：kafka服务器地址(可以多个)
consumergroup-id :指定一个默认的组名
不指定的话会报
1 earliest ：当各分区下有已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，从头开始消费
2 latest ：当各分区下有已提交的 offset 时，从提交的 offset 开始消费；无提交的 offset 时，消费新产生的该分区下的数据
3 none ： topic 各分区都存在已提交的 offset 时，从 offset 后开始消费；只要有一个分区不存在已提交的 offset ，则抛出异常
这个属性也是必须配置的，不然也是会报错的
在使用Kafka发送接收消息时，生产者 producer 端需要序列化，消费者 consumer 端需要反序列化，由于网络传输过来的是 byte[] ，只有反序列化后才能得到生产者发送的真实的消息内容。这样消息才能进行网络传输
consumerkey-deserializer 和 consumervalue-deserializer 是消费者 key/value 反序列化
producerkey-deserializer 和 producervalue-deserializer 是生产者 key/value 序列化
StringDeserializer 是内置的字符串反序列化方式
StringSerializer 是内置的字符串序列化方式
在 orgapachekafkacommonserialization 源码包中还提供了多种类型的序列化和反序列化方式
要自定义序列化方式，需要实现接口 Serializer
要自定义反序列化方式，需要实现接口 Deserializer
详细可以参考
>随着信息技术的快速发展及互联网用户规模的急剧增长，计算机所存储的信息量正呈爆炸式增长，目前数据量已进入大规模和超大规模的海量数据时代， 如何高效地存储、分析、处理和挖掘海量数据 已成为技术研究领域的热点和难点问题。而 如何采集和运营管理、分析这些数据 也是大数据处理中一个至关重要的组成环节，这就需要相应的基础设施对其提供支持。针对这个需求，当前业界已有很多开源的消息系统应运而生，kafka就是一款当然非常流行的消息系统。
Kafka是一款开源的、轻量级的、分布式、可分区和具有复制备份的（Replicated）、基于ZooKeeper协调管理的分布式流平台的功能强大的消息系统。作为一个流式处理平台，必须具备以下3个关键特性：
1) 能够允许发布和订阅流数据。
2) 存储流数据时提供相应的容错机制。
3) 当流数据到达时能够被及时处理。
消息流系统kafka的基本结构包括生产者和消费者，以及kafka集群。
生产者负责生产消息，将消息写入Kafka集群；消费者从Kafka集群中拉取消息。
消息是Kafka通信的基本单位 ，由一个 固定长度的消息头 和一个 可变长度的消息体 构成。
Kafka将 一组消息 抽象归纳为一个主题（Topic），也就是说，一个主题是对消息的一个分类。 生产者将消息指定主题发送到kafka集群，消费者订阅主题或主题的某些分区进行消费。
Kafka将一组消息归纳为一个主题，而 每个主题又被分成一个或多个分区(Partition) 。每个分区由一系列有序、不可变的消息组成，是一个有序队列。 每个分区在物理上对应为一个文件夹 ，分区的命名规则为主题名称后接“—”连接符，之后再接分区编号，分区编号从0开始，编号最大值为分区的总数减1。
分区使得Kafka在并发处理上变得更加容易，理论上来说，分区数越多吞吐量越高，但这要根据集群实际环境及业务场景而定。同时，分区也是Kafka保证消息被顺序消费以及对消息进行负载均衡的基础。
疑问和答案 ：分区如何保证消息被顺序消费？每个分区内的消息是有序的，但不同分区间如何保证？猜测是分区从存储空间上比较大，分区个数少。顺序消费的主要因素在分区内的消息，分区间的可以忽略。高吞吐率顺序写磁盘估计也是这个原因。
Kafka只能保证一个分区之内消息的有序性，并不能保证跨分区消息的有序性。 每条消息被追加到相应的分区中，是顺序写磁盘，因此效率非常高，这是Kafka高吞吐率的一个重要保证 。同时与传统消息系统不同的是，Kafka并不会立即删除已被消费的消息，由于磁盘的限制消息也不会一直被存储，因此 Kafka提供两种删除老数据的策略 ，一是基于消息已存储的时间长度，二是基于分区的大小。这两种策略都能通过配置文件进行配置。
每个分区又有一至多个副本（Replica），分区的副本分布在集群的不同代理上，以提高可用性。
从存储角度上分析，分区的每个副本在逻辑上抽象为一个日志（Log）对象，即分区的副本与日志对象是一一对应的。每个主题对应的 分区数 可以在Kafka启动时所加载的配置文件中配置，也可以在创建主题时指定。当然，客户端还可以在主题创建后修改主题的分区数。
为什么副本要分Leader和Follower？ 如果没有Leader副本，就需要所有的副本都同时负责读/写请求处理，同时还得保证这些副本之间数据的一致性，假设有n个副本则需要有n×n条通路来同步数据，这样数据的一致性和有序性就很难保证。
为解决这个问题，Kafka选择分区的一个副本为Leader，该分区其他副本为Follower，只有 Leader副本 才负责处理客户端 读/写请求 ，Follower副本从Leader副本同步数据。
引入Leader副本后客户端只需与Leader副本进行交互，这样数据一致性及顺序性就有了保证。Follower副本从Leader副本同步消息，对于n个副本只需n-1条通路即可，这样就使得系统更加简单而高效。
副本Follower与Leader的角色并不是固定不变的，如果Leader失效，通过相应的选举算法将从其他Follower副本中选出新的Leader副本。
疑问 ：leader副本和follower副本是如何选出来的？通过zookeeper选举的嘛？
Kafka在ZooKeeper中动态维护了一个 ISR（In-sync Replica） ，即保存同步的副本列表，该列表中保存的是与Leader副本保持消息同步的所有副本对应的代理节点id。 如果一个Follower副本宕机或是落后太多 ，则该Follower副本节点将 从ISR列表中移除 。 本书用宕机 来特指某个代理失效的情景，包括但不限于代理被关闭，如代理被人为关闭或是发生物理故障、心跳检测过期、网络延迟、进程崩溃等。
任何发布到分区的消息会被直接追加到日志文件的尾部（分区目录下以“log”为文件名后缀的数据文件），而每条 消息 在日志文件中的位置都会对应一个按序递增的 偏移量 。偏移量是一个分区下严格有序的 逻辑值 ，它并不表示消息在磁盘上的物理位置。由于Kafka几乎不允许对消息进行随机读写，因此Kafka并没有提供额外索引机制到存储偏移量。
消费者可以通过控制消息偏移量来对消息进行消费 ，如消费者可以指定消费的起始偏移量。 为了保证消息被顺序消费，消费者已消费的消息对应的偏移量也需要保存 。需要说明的是，消费者对消息偏移量的 *** 作并不会影响消息本身的偏移量。旧版消费者将消费偏移量保存到ZooKeeper当中， 而新版消费者是将消费偏移量保存到Kafka内部一个主题当中。 当然，消费者也可以自己在外部系统保存消费偏移量，而无需保存到Kafka中。
推测 ：一个主题有多个分区，一个分区有多个副本。一个主题(一类消息)有多个分区(消息被分段)，一个分区(每段消息)有多个副本(每段消息的副本数)。消息一旦发给kafka，就会分配一个偏移量，在多个副本中的偏移量是一样的。这样的话，消费者通过偏移量消费时对于多个副本就没有差异性。
Kafka集群由一个或多个Kafka实例构成，每一个Kafka实例称为代理（Broker），通常也称代理为Kafka服务器（KafkaServer）。在生产环境中Kafka集群一般包括一台或多台服务器，我们可以在一台服务器上配置一个或多个代理。 每一个代理都有唯一的标识id，这个id是一个非负整数 。在一个Kafka集群中，每增加一个代理就需要为这个代理配置一个与该集群中其他代理不同的id, id值可以选择任意非负整数即可，只要保证它在整个Kafka集群中唯一，这个id就是代理的名字，也就是在启动代理时配置的brokerid对应的值。
生产者（Producer）负责将消息发送给代理，也就是向Kafka代理发送消息的客户端。
消费者（Comsumer）以拉取（pull）方式拉取数据，它是消费的客户端。在Kafka中 每一个消费者都属于一个特定消费组 （ConsumerGroup），可以为每个消费者指定一个消费组，以groupId代表消费组名称，通过groupid配置设置。 如果不指定消费组 ，则该消费者属于默认消费组test-consumer-group。
每个消费者有一个全局唯一的id ，通过配置项clientid指定， 如果客户端没有指定消费者的id， Kafka会自动为该消费者生成一个全局唯一的id，格式为${groupId}-${hostName}-${timestamp}-${UUID前8位字符}。 同一个主题的一条消息只能被同一个消费组下某一个消费者消费 ，但不同消费组的消费者可同时消费该消息。 消费组是Kafka用来实现对一个主题消息进行广播和单播的手段 ，实现消息广播只需指定各消费者均属于不同的消费组，消息单播则只需让各消费者属于同一个消费组。
推论： kafka消息是按照消息类型(主题)，在一个消费者组中只能消费一次。也就是一个消费者组只消费一类型的消息。如果某个服务要消费一类消息，必须将自己置为不同的消费者组。
Kafka利用ZooKeeper保存相应元数据信息， Kafka元数据信息包括如代理节点信息、Kafka集群信息、旧版消费者信息及其消费偏移量信息、主题信息、分区状态信息、分区副本分配方案信息、动态配置信息等。 Kafka在启动或运行过程当中会在ZooKeeper上创建相应节点 来保存元数据信息， Kafka通过监听机制在这些节点注册相应监听器来监听节点元数据的变化 ，从而由ZooKeeper负责管理维护Kafka集群，同时通过ZooKeeper我们能够很方便地对Kafka集群进行水平扩展及数据迁移。
springkafkabootstrap-servers=101257041:9092,101257035:9092,101257036:9092
#client-id
springkafkaclient-id=group1
生产者参数
# acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
# acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
# acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应
springkafkaproduceracks=1
#当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
springkafkaproducerbatch-size=16384
# 发生错误后，消息重发的次数。
springkafkaproducerretries=3
# 设置生产者内存缓冲区的大小。
springkafkaproducerbuffer-memory=33554432
springkafkaproducerkey-serializer=orgapachekafkacommonserializationStringSerializer
springkafkaproducervalue-serializer=orgapachekafkacommonserializationStringSerializer
消费者参数
# 自动提交的时间间隔
springkafkaconsumerauto-commit-interval=1000
# offset的消费位置
springkafkaconsumerauto-offset-reset=latest
# 是否自动提交
springkafkaconsumerenable-auto-commit=false
# 最大拉取间隔时间
springkafkaconsumermaxpollintervalms =600000
# 会话超时时间
springkafkaconsumersessiontimeoutms =10000
springkafkaconsumerkey-deserializer=orgapachekafkacommonserializationStringDeserializer
springkafkaconsumervalue-deserializer=orgapachekafkacommonserializationStringDeserializer
# 消费组名称
springkafkaconsumergroupId=dmsdecision
# 最大拉取条数
springkafkaconsumermax-poll-records=30
# 心跳时间
springkafkaconsumerheartbeat-interval=3000
# kafka springkafkapropertiesparsefileContainerFactory_concurrency监听线程数未设置时，本参数生效
springkafkalistenerconcurrency=30
#MANUAL 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgmentacknowledge()后提交
#MANUAL_IMMEDIATE 手动调用Acknowledgmentacknowledge()后立即提交
#RECORD 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
#BATCH 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
#TIME 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
#COUNT 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
#COUNT_TIME TIME或COUNT 有一个条件满足时提交
# ack_mode为COUNT/COUNT_TIME 时配置
springkafkalistenerack-mode=manual_immediate
# ack_mode为COUNT/COUNT_TIME 时配置
springkafkalistenerack-count=
# ack_mode为/COUNT_TIME 时配置
springkafkalistenerack-time=
# poll拉取数据超时时间
springkafkalistenerpoll-timeout=我觉得应该是需要的，一般来说任何的电子设备进行了一些重大的更改都是需要重启的，特别是你的这个还是更改了设备的这个样子的。下面是关于配置的扩展资料。
硬件方面
1．CPU，这个主要取决于频率和二级缓存，三级缓存，核心数量。频率越高、二级缓存越大，三级缓存越大，核心越多，运行速度越快。速度越快的CPU只有三级缓存影响响应速度。
2．内存，内存的存取速度取决于接口、颗粒数量多少与储存大小（包括内存的接口，如：SDRAM133，DDR333，DDR2-533，DDR2-800,DDR3-1333、DDR3-1600、DDR4-2133），一般来说，内存越大，处理数据能力越强，而处理数据的速度主要看内存属于哪种类型（如DDR就没有DDR3处理得快）。一般大型游戏（PUBG、战地5、俄罗斯钓鱼、使命召唤16等）与大型软件（pr、ae等）都会占用很多内存，因此，对于游戏玩家来说，越高的内存可以玩越多的游戏。
3．主板，主要还是处理芯片，如：笔记本i965比i945芯片处理能力更强，i945比i910芯片在处理数据的能力又更强些，依此类推。并且更好的主板还可以适配更强大的CPU（中央处理器）。
4．硬盘，硬盘分为固态硬盘（SSD）、机械硬盘（HDD）、混合硬盘（SSHD），固态硬盘速度最快，混合硬盘次之，机械硬盘最差。越大的硬盘存的文件就多，（如存放，音乐等）首先硬盘的数据读取与写入的速度和硬盘的转速（分：高速硬盘和低速硬盘，高速硬盘一般用在大型服务器中，如：10000转，15000转；低速硬盘用在一般电脑中，包括笔记本电脑），台式机电脑一般用7200转，笔记本电脑一般用5400转，这主要是考虑到高速硬盘在笔记本电脑中由于电脑移动振动意外刮伤硬盘盘片以及功耗和散热原因。
硬盘速度又因接口不同，速率不同，一般而言，分IDE和SATA（也就是常说的串口）接口，早前的硬盘多是IDE接口，相比之下，存取速度比SATA接口的要慢些。
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 服务
到官网 >
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）