「kafka依赖java」kafka依赖haproxy组件
今天给各位分享kafka依赖java的知识，其中也会对kafka依赖haproxy组件进行解释，如果能碰巧解决你现在面临的问题，别忘了关注本站，现在开始吧！
本文目录一览：
- 1、Kafka相关内容总结（Kafka集群搭建手记）
- 2、Springboot集成Kafka
- 3、linux 安装kafka需要配置java环境吗
- 4、kafka简介
- 5、3分钟带你彻底搞懂 Kafka
Kafka相关内容总结（Kafka集群搭建手记）
Kafka is a distributed,partitioned,replicated commit logservice。它提供了类似于JMS的特性，但是在设计实现上完全不同，此外它并不是JMS规范的实现。kafka对消息保存时根据Topic进行归类，发送消息者成为Producer,消息接受者成为Consumer,此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。无论是kafka集群，还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。
入门请参照：
在此不再赘述。
这部分不是本文的重点，但是kafka需要用到kafka集群，所以先搭建kafka集群。
从kafka官方文档看到，kafka似乎在未来的版本希望抛弃zookeep集群，自己维护集群的一致性，拭目以待吧。
我们搭建集群使用的是三台同机房的机器，因为zookeeper不怎么占资源也不怎么占空间（我们的业务目前比较简单），所以三台机器上都搭建了zookeeper集群。
搭建zookeeper集群没什么难度，参考文档：
下面列一下我的配置并解析：
一共用三台物理机器，搭建一个Kafka集群。
每台服务器的硬盘划分都是一样的，每个独立的物理磁盘挂在一个单独的分区里面，这样很方便用于Kafka多个partition的数据读写与冗余。
/data1比较小，为了不成为集群的瓶颈，所以/data1用于存放kafka以及Zookeeper
每台机器的磁盘分布如下：
下面是kafka的简单配置，三台服务器都一样，如有不一致的在下文有说明。
kafka安装在目录/usr/local/kafka/下，下面的说明以10.1.xxx.57为例。
最重要的配置文件server.properties，需要配置的信息如下：
从上面的配置看到，kafka集群不需要像hadoop集群那样，配置ssh通讯，而且一个kafka服务器（官方文档称之为broker，下面统一使用这个称呼）并不知道其他的kafka服务器的存在，因此你需要逐个broker去启动kafka。各个broker根据自己的配置，会自动去配置文件上的zk服务器报到，这就是一个有zk服务器粘合起来的kafka集群。
我写了一个启动脚本，放在 /usr/local/kafka/bin 下面。启动脚本每个broker都一样：
如同kafka集群里面每一个broker都需要单独启动一样，kafka集群里面每一个broker都需要单独关闭。
官方给出的关闭脚本是单独运行 bin/kafka-server-stop.sh
但是我运行的结果是无法关闭。打开脚本一看，才发现是最简单的办法，发一个TERM信号到kafka的java进程，官方脚本给出的grep有点问题。
发信号之后，一直tail着kafka日志，看到正常关闭。
指定zookeeper服务器，topic名称是LvsKafka（注意topic名称不能有英文句号(.)和下划线(_)，否则会通不过，理由是名称会冲突，下文对此略有解析）
replication-factor指出重复因子是2，也就是每条数据有两个拷贝，可靠性考虑。
partitions 指出需要多少个partition，数据量大的多一点，无论生产和消费，这是负载均衡和高并发的需要。
可以看到刚才新建的24个partition，比如partition 5， 他的leader是broker 59，也就是10.1.xxx.59这台机器。
建立topic时我们指出需要2个拷贝，从上面的输出的Replicas字段看到，这两个拷贝放在59,58两个机器，也就是10.1.xxx.59和10.1.xxx.58.
Isr表示当前partition的所有拷贝所在的机器中，哪些是还活着（可以提供服务）的。现在是59和58都还存活。
这个命令另外还会看到一些类似于下面的内容：
__consumer_offsets到底是什么呢？其实就是客户端的消费进度，客户端会定时上报到kafka集群，而kafka集群会把每个客户端的消费进度放入一个自己内部的topic中，这个topic就是__consumer_offsets。我查看过__consumer_offsets的内容，其实就是每个客户端的消费进度作为一条消息，放入__consumer_offsets这个topic中。
这里给了我们两个提示：
1、kafka自己管理客户端的消费进度，而不是依靠zk，这就是kafka官方文档说的kafka未来会抛弃zk的底气之一；
2、留意到这个kafka自己的topic是带下划线的，也就是，kafka担心我们自己建的topic如果带下划线的话会跟这些内部自用的topic冲突；
Springboot集成Kafka
最近陆续收到不少朋友反馈，他们在计划要在springboot项目引入kafka中间件。在网上找过很多资料，但都比较零散，按照其说明进行操作后让项目正常跑起来仍是比较坎坷，听说我对kafka比较了解，希望给予一起分享。刚好最近因为疫情，实际相对于平常稍宽松，也想借此写点东西，一来作为自己的总结，二来可以给予有需要的朋友一些引导，针对此文期望对各位遇到问题的朋友有一定的帮助。
1. 安装JDK ，具体版本可以根据项目实际情况选择，目前使用最多的为jdk8
2. 安装Zookeeper ，具体版本可以根据项目实际情况选择，本项目使用的是3.5.8
3. 安装Kafka ，具体版本可以根据项目实际情况选择，本项目使用的是3.5.1
4. 安装Kafka Manage （非必要：安装主要是了对kafka项目操作提供图形化界面操作），具体版本可以根据项目实际情况选择，本项目使用的是1.3.3.7
parent
groupIdorg.springframework.boot/groupId
artifactIdspring-boot-starter-parent/artifactId
version2.3.12.RELEASE/version
relativePath/ !-- lookup parent from repository --
/parent
dependencies
!--springboot web依赖 --
dependency
groupIdorg.springframework.boot/groupId
artifactIdspring-boot-starter-web/artifactId
/dependency
!--kafka依赖 --
dependency
groupIdorg.springframework.kafka/groupId
artifactIdspring-kafka/artifactId
/dependency
/dependencies
spring:
kafka:
bootstrap-servers: 127.0.0.1:9092
producer: #生产者
# 发生错误后，消息重发的次数。重试1次,此值需要结合业务场景，重试与否各有千秋(重试，好处：尽可能的确保生产者写入block成功；坏处：有可能时带有顺序写入的数据打乱顺序
#比如：依次写入数据 1/2/3，但写入1时因网络异常，进行了重写，结果落到block的数据成了2/3/1)
retries: 1
#当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。模式时16k
batch-size: 16384 #16k
# 设置生产者内存缓冲区的大小
buffer-memory: 33554432
acks: 1
# 键的序列化方式
key-serializer: org.apache.kafka.common.serialization.StringSerializer
# 值的序列化方式
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
#group-id: default-group
# 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D.此属性只有在enable-auto-commit：true时生效
auto-commit-interval: 1S
enable-auto-commit: false
# 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
# latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
# earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
auto-offset-reset: earliest
# 键的反序列化方式
key-deserializer: org.apache.kafka.common.serialization.StringSerializer
# 值的反序列化方式
value-deserializer: org.apache.kafka.common.serialization.StringSerializer
listener:
# 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
# RECORD
# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
# BATCH
# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
# TIME
# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
# COUNT
# TIME | COUNT 有一个条件满足时提交
# COUNT_TIME
# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
# MANUAL
# 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种
# MANUAL_IMMEDIATE
ack-mode: manual_immediate
# 在侦听器容器中运行的线程数
concurrency: 5
bin/kafka-topics.sh --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic test
-- 127.0.0.1:2181 zookeeper 服务器地址
-- replication-factor partitions 副本数量
--partitions partition数量
点击【Cluster】【Add Cluster】打开如下添加集群的配置界面：
输入集群的名字（如Kafka-Cluster-1）和 Zookeeper 服务器地址（如localhost:2181），选择最接近的Kafka版本（如0.10.1.0）
package com.charlie.cloudconsumer.service.impl.kafka;
import com.charlie.cloudconsumer.common.utils.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import java.util.UUID;
/**
* @Author: charlie
* @CreateTime: 2022/4/9
* @Description : kafka消息生产端,为实践业务提供向kafka block发现消息的API
*/
@Component
@Slf4j
public class QueueProducer {
@Autowired
private KafkaTemplatekafkaTemplate;
public void sendQueue(String topic,Object msgContent) {
String obj2String =JSON.toJSONString(msgContent);
log.info("kafka service 准备发送消息为：{}",obj2String);
//发送消息
ListenableFuturefuture =kafkaTemplate.send(topic,UUID.randomUUID().toString(),obj2String);
future.addCallback(new ListenableFutureCallback() {
//消息发送成功
@Override
public void onSuccess(SendResult stringObjectSendResult) {
log.info("[kafka service-生产成功]topic：{}，结果{}",topic, stringObjectSendResult.toString());
}
//消息发送失败
@Override
public void onFailure(Throwable throwable) {
//发送失败的处理，本处只是记录了错误日志,可结合实际业务做处理
log.info("[kafka service-生产失败]topic：{}，失败原因{}",topic, throwable.getMessage());
}
});
}
}
package com.charlie.cloudconsumer.service.impl.kafka;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.stereotype.Component;
/**
* @Author: charlie
* @CreateTime: 2022/4/9
* @Description : 消费端实际的业务处理对象
*/
@Component //添加此注解的原因是因为消费端在项目启动时就会初始化，消费端需要用到此类，故也让此类在项目启动时进行注册
public class QueueDataProcess {
public boolean doExec(Object obj) {
// todu 具体的业务逻辑
if (ObjectUtils.isNotEmpty(obj)) {
return true;
}else {
return false;
}
}
}
package com.charlie.cloudconsumer.service.impl.kafka;
import com.charlie.cloudconsumer.common.utils.JSON;
import com.charlie.cloudconsumer.model.Order;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import java.util.Optional;
/**
* @Author: charlie
* @CreateTime: 2022/4/9
* @Description : kafka消息消费端，负责消费特定topic消息
*/
@Component
@Slf4j
@SuppressWarnings("all")
public class QueueConsumer {
@Autowired
private QueueDataProcess queueDataProcess;
/**
*
*/
@KafkaListener(topics ="test", groupId ="consumer")
public void doConsumer(ConsumerRecord record,Acknowledgment ack,@Header(KafkaHeaders.RECEIVED_TOPIC)String topic) {
Optional message =Optional.ofNullable(record.value());
if (message.isPresent()) {
try {
Object msg =message.get();
log.info("[kafka-消费] doConsumer 消费了： Topic:" + topic +",Message:" +msg);
boolean res =queueDataProcess.doExec(JSON.parseObject(msg.toString(),Order.class));
if (res) {
ack.acknowledge();
}
}catch (Exception ex) {
log.error("[kafka-消费异常] doConsumer Error {} ",ExceptionUtils.getFullStackTrace(ex));
}
}
}
}
package com.charlie.cloudconsumer.controller;
import com.alibaba.fastjson.JSON;
import com.charlie.cloudconsumer.common.utils.AjaxResult;
import com.charlie.cloudconsumer.common.utils.BuildResponseUtils;
import com.charlie.cloudconsumer.model.Order;
import com.charlie.cloudconsumer.service.impl.kafka.QueueProducer;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
/**
* @Author: charlie
* @CreateTime: 2022/4/9
* @Description : kafka消息发送控制器，负责接受用户的发送的队列消息
*/
@RestController
@RequestMapping(value ="/kafka",produces =MediaType.APPLICATION_JSON_VALUE)
public class KafkaController {
@Autowired
private QueueProducer queueProducer;
@RequestMapping(value = "/send",method = RequestMethod.POST)
public AjaxResult? sendMsg(@RequestBody Order order) {
AjaxResult? ajaxResult= null;
if (ObjectUtils.isNotEmpty(order)) {
queueProducer.sendQueue("test",order);
ajaxResult = BuildResponseUtils.success(0,"发送消息："+ JSON.toJSONString(order) + "成功!");
} else {
ajaxResult = BuildResponseUtils.success(1,"发送消息："+ JSON.toJSONString(order) + "失败!");
}
return ajaxResult;
}
}
linux 安装kafka需要配置java环境吗
纯java开发的软件在linux下面也可以应用自如。那么首先就需要配置好linux下的java环境，具体说来，就是配置jdk环境变量。 介绍在linux下配置jdk环境变量的几种常用方法。 首先在linux下安装jdk，如果出现提示权限不够(且root下也提示权限不够)，...
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的顶级项目。
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依赖java和kafka依赖haproxy组件的介绍到此就结束了，不知道你从中找到你需要的信息了吗 ？如果你还想了解更多这方面的信息，记得收藏关注本站。
发布于：2022-12-15，除非注明，否则均为原创文章，转载请注明出处。