# Kafka_深入探秘者（1）：初识 kafka

Kafka_深入探秘者（1）：初识 kafka

一、kafka 特性

1、Kafka ：最初是由 Linkedln 公司采用 Scala 语言开发的一个多分区、多副本并且基于 ZooKeeper 协调的分布式消息系统，现在已经捐献给了 Apache 基金会。目前 Kafka 已经定位为一个分布式流式处理平台，它以高吞吐、可持久化、可水平扩展、支持流处理等多种特性而被广泛应用。

2、Apache Kafka 是一个分布式的发布-订阅消,息系统，能够支撑海量数据的数据传递。在离线和实时的消息处理业务系统中，Kafka 都有广泛的应用。Kafka 将消息持久化到磁盘中，并对消息创建了备份保证了数据的安全。Kafka 在保证了较高的处理速度的同时，又能保证数据处理的低延迟和数据的零丢失。

3、示例图：

在这里插入图片描述

4、kafka 特性

  • (1)高吞吐量、低延迟: kafka 每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个主题可以分多个分区,消费组对分区进行消费操作;

  • (2)可扩展性: kafka 集群支持热扩展;

  • (3)持久性、可靠性: 消息被持久化到本地磁盘，并且支持数据备份防止数据丢失;

  • (4)容错性: 允许集群中节点失败(若副本数量为n,则允许n-1个节点失败);

  • (5)高并发: 支持数千个客户端同时读写;

5、kafka 使用场景

  • (1)日志收集:

一个公司可以用 Kafka 可以收集各种服务的 l0g，通过kafka以统一接口服务的方式开放给各种 consumer，例如 Hadoop、Hbase、Solr 等;

  • (2)消息系统:

解耦和生产者和消费者、缓存消息等:

  • (3)用户活动跟踪:

Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动,这些活动信息被各个服务器发布到 kafka 的 topic 中，然后订阅者通过订阅这些 topic 来做实时的监控分析，或者装载到 Hadoop、数据仓库中做离线分析和挖掘;

  • (4)运营指标:

Kafka 也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

  • (5)流式处理:

比如 spark streaming 和 storm。

6、技术优势

1）可伸缩性: Kafka 的两个重要特性造就了它的可伸缩性。

  • 1、Kafka 集群在运行期间可以轻松地扩展或收缩(可以添加或删除代理)，而不会宕机。
  • 2、可以扩展一个 Kafka 主题来包含更多的分区。由于一个分区无法扩展到多个代理，所以它的容量受到代理磁盘空间的限制。能够增加分区和代理的数量意味着单个主题可以存储的数据量是没有限制的。

2）容错性和可靠性:

Kafka 的设计方式使某个代理的故障能够被集群中的其他代理检测到。由于每个主题都可以在多个代理上复制，所以集群可以在不中断服务的情况下从此类故障中恢复并继续运行。

3）吞吐量: 代理能够以超快的速度有效地存储和检索数据

二、kafka 概念详解

1、kafka 官网： https://kafka.apache.org/

2、kafka 概念 流程图：

2.png

3、Producer

生产者即数据的发布者，该角色将消息发布到 Kafka 的 topic 中。broker 接收到生产者发送的消息后，broker 将该消息追加到当前用于追加数据的 segment 文件中。生产者发送的消息，存储到一个 partition 中，生产者也可以指定数据存储的 partition。

4、Consumer

消费者可以从 broker 中读取数据。消费者可以消费多个 topic 中的数据。

5、Topic

在 Kafka 中，使用一个类别属性来划分数据的所属类，划分数据的这个类称为 topic。如果把Kafka看做为一个数据库，topic 可以理解为数据库中的一张表，topic 的名字即为表名。

6、Partition

topic 中的数据分割为一个或多个 partition。每个 topic 至少有一个 partition。每个 partition 中的数据使用多个 segment 文件存储。partition 中的数据是有序的，partition 间的数据丢失了数据的顺序。如果 topic 有多个 partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将 partition 数目设为 1。

7、Partition offset

每条消息都有一个当前 Partition 下唯一的64字节的 offset，它指明了这条消息的起始位置。

8、Replicas of partition

副本是一个分区的备份。副本不会被消费者消费，副本只用于防止数据丢失，即消费者不从为 folower 的 partition 中消费数据，而是从为 leader 的 partition 中读取数据。副本之间是一主多从的关系。

9、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 集群数据不均衡。

10、Leader

每个 partition 有多个副本，其中有且仅有一个作为 Leader，Leader 是当前负责数据的读写的 partition。

11、Follower

Follower 跟随 Leader，所有写请求都通过 Leader 路由，数据变更会广播给所有 Follower，Follower 与 Leader 保持数据同步。如果 Leader 失效，则从 Follower 中选举出一个新的Leader。当 Follower 与 Leader 挂掉、卡住或者同步太慢，leader 会把这个 follower 从 “in syncreplicas"(ISR) 列表中删除，重新创建一个 Follower。

12、Zookeeper

Zookeeper 负责维护和协调 broker。当 Kafka 系统中新增了 broker 或者某个 broker 发生故障失效时，由 ZooKeeper 通知生产者和消费者。生产者和消费者依据 Zookeeper 的 broker 状态信息与 broker 协调数据的发布和订阅任务。

13、AR(Assigned Replicas)

分区中所有的副本统称为 AR。

14、ISR(In-Sync Replicas)

所有与 Leader 部分保持一定程度的副(包括 Leader 副本在内)本组成 ISR。

15、OSR(Out-of-Sync-Replicas)

与 Leader 副本同步滞后过多的副本。

16、HW(High Watermark)

高水位，标识了一个特定的 offset，消费者只能拉取到这个 offset 之前的消息。

17、LEO(Log End offset)

即日志末端位移(log end offset)，记录了该副本底层日志(l0g)中下一条消息的位移值。注意是下一条消息!也就是说，如果 LEO=10，那么表示该副本保存了10条消息，位移值范围是[0,9]。

在这里插入图片描述

三、kafka 环境配置 jdk、zookeeper 下载安装

1、 jdk-12 下载安装

2、jdk-12 环境变量配置


sudo vim /etc/profile # 添加 jdk 环境变量配置
export JAVA_HOME=/opt/java/jdk-12.0.1
export JRE_HOME-${JAVA_HOME}/jre
#export MAVEN_HOME=/opt/maven/apache-maven-3.5.0
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
#export PATH=S{JAVA_HOME}/bin:${MAVEN_HOME}/bin:$PATH
export PATH=S{JAVA_HOME}/bin:$PATH# 测试 jdk 是否安装配置成功
java -version

3、zookeeper 下载安装

1）Zookeeper 是安装 Kafka 集群的必要组件，Kafka 通过 Zookeeper 来实施对元数据信息的管理，包括集群、主题分区等内容。

2）同样在官网下载安装包到指定目录解压缩，步骤如下:

ZooKeeper 官网: http://zookeeper.apache.org

3）把 zookeeper 上传至服务器，并解压


# 切换目录：
cd /usr/local/zookeeper/
# 解压即安装
tar -zxvf apache-zookeeper-3.6.3-bin.tar.gz -C /usr/local/zookeeper/

4）修改 Zookeeper 的配置文件，首先进入安装路径 conf 目录，并将 zoo_sample.cfg 文件修改为 zoo.cfg，并对核心参数进行配置。文件内容如下:


# 切换目录
cd /usr/local/zookeeper/apache-zookeeper-3.6.3-bin/conf/# 将 zoo_sample.cfg 文件修改为 zoo.cfg
mv zoo_sample.cfg zoo.cfg# 对核心参数进行配置
vim zoo.cfg# The number of milliseconds of each tick
#zk服务器的心跳时间
tickTime-2080
# The number of ticks that the initial
#synchronization phase can take
#投票选举新Leader的初始化时间
initlimit=10
# The number of ticks that can pass between
# sendidg a request and getting an acknowledgementsyncLimit-5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.# 数据目录(需要新建此目录)
dataDir=/usr/local/zookeeper/apache-zookeeper-3.6.3-bin/data# 日志目录
dataLogDir=/usr/local/zookeeper/apache-zookeeper-3.6.3-bin/log# the port at which the clients will connect
#Zookeeper对外服务端口，保持默认
clientPort-2181

4、启动 Zookeeper 命令: bin/zkServer.sh start


# 切换目录
cd /usr/local/zookeeper/apache-zookeeper-3.6.3-bin/
# 启动 Zookeeper
bin/zkServer.sh start
# 查询 zookeeper 是否启动成功
ps -ef | grep zookeeper 
# 或者 
jps -l

四、kafka 环境配置 kafka 下载安装

1、 下载安装解压 kafka

官网下载地址： https://kafka.apache.org/downloads


# 切换目录
cd /usr/local/kafka/# 解压即安装
tar -zxvf kafka-2.12.2.8.0.tar.gz -C /usr/local/kafka/

2、配置 kafka 参数


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 修改 kafka 配置参数：
vim config/server.properties# server.properties 配置中需要关注以下几个参数:# 表示 broker 的编号，如果集群中有多个 broker，则每个 broker 的编号需要设置的不同
broker.id=0 # brokder 对外提供的服务入口地址（ kafka 监听地址：你的虚拟机 IP 地址）
# listeners=PLAINTEXT://127.0.0.1:9092 
listeners=PLAINTEXT://172.18.30.110:9092 # 设置存放消息日志文件的地址
log.dirs=/tmp/kafka/log # Kafka 所需 Zookeeper 集群地址
zookeeper.connect=lolalhost:2181

3、启动 kafka


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 启动 kafka
bin/kafka-server-start.sh config/server.properties

4、查询 kafka 是否启动成功：


# 重新打开一个终端# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 查询 kafka 是否启动成功
ps -ef | grep kafka 
# 或者 
jps -l

五、kafka 消息的生产与消费

1、创建一个主题

1）命令： bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic heima --partitions 2 --replication-factor 1

2）参数说明：

–zookeeper ：指定了 kafka 所连接的 zookeeper 服务地址。
–topic : 指定了所要创建主题的名称
–partitions : 指定了分区个数
–replication-factor : 指定了副本因子
–create : 创建主题的动作指令。


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 创建一个名为 heima 的 主题
bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic heima --partitions 2 --replication-factor 1# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-topics.sh --zookeeper 172.18.30.110:2181 --create --topic heima --partitions 2 --replication-factor 1

2、展示出当前所有主题


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 展示出当前所有主题
bin/kafka-topics.sh --zookeeper 1ocalhost:2181 --1ist# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-topics.sh --zookeeper 172.18.30.110:2181 --1ist

3、 查看主题详情：


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 查看主题详情
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic heima# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-topics.sh --zookeeper 172.18.30.110:2181 --describe --topic heima

4、启动消费端接收消息：

命令: bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic heima

参数说明：
–bootstrap-server 指定了连接 Kafka 集群的地址
–topic 指定了消费端订阅的主题


# 重新打开一个终端：# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 启动消费端接收消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic heima# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-console-consumer.sh --bootstrap-server 172.18.30.110:9092 --topic heima

5、生产端发送消息

命令: bin/kafka-console-producer.sh --broker-list localhost:9092 --topic heima

参数说明：
–broker-list 指定了连接的 Kafka 集群的地址
–topic 指定了发送消息时的主题


# 重新打开一个终端：# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 生产端发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic heima# 或者把 localhost 换成 填写你的 虚拟机 IP 地址（如：172.18.30.110）：
bin/kafka-console-producer.sh --broker-list 172.18.30.110:9092 --topic heima

在这里插入图片描述

在这里插入图片描述

六、kafka java 第一个程序 001

1、打开 idea 创建 artifactId 名为 kafka_learn 的 maven 工程。

--> idea --> File --> New --> Project --> Maven Project SDK: ( 1.8(java version "1.8.0_131" ) --> Next --> Groupld : ( djh.it )Artifactld : ( kafka_learn )Version : 1.0-SNAPSHOT--> Name: ( kafka_learn )Location: ( ...\kafka_learn\ )	--> Finish

2、在 kafka_learn 工程的 pom.xml 文件中导入依赖坐标。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>djh.it</groupId><artifactId>kafka_learn</artifactId><version>1.0-SNAPSHOT</version><name>kafka_learn</name><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.8.RELEASE</version><relativePath></relativePath></parent><properties><java.version>8</java.version>
<!--        <scala.version>2.11</scala.version>--><scala.version>2.12</scala.version><slf4j.version>1.7.21</slf4j.version>
<!--        <kafka.version>2.0.0</kafka.version>--><kafka.version>2.8.0</kafka.version><lombok.version>1.18.8</lombok.version><junit.version>4.11</junit.version><gson.version>2.2.4</gson.version><protobuff.version>1.5.4</protobuff.version>
<!--        <spark.version>2.3.1</spark.version>--><spark.version>2.4.8</spark.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>${kafka.version}</version></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka_${scala.version}</artifactId><version>${kafka.version}</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>${slf4j.version}</version></dependency><dependency><groupId>com.google.code.gson</groupId><artifactId>gson</artifactId><version>${gson.version}</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>${junit.version}</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><scope>provided</scope></dependency><dependency><groupId>io.protostuff</groupId><artifactId>protostuff-core</artifactId><version>${protobuff.version}</version></dependency><dependency><groupId>io.protostuff</groupId><artifactId>protostuff-runtime</artifactId><version>${protobuff.version}</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-core</artifactId><version>2.9.4</version></dependency><dependency><groupId>com.fasterxml.jackson.module</groupId><artifactId>jackson-module-scala_2.11</artifactId><version>2.9.5</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_${scala.version}</artifactId><version>${spark.version}</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_${scala.version}</artifactId><version>${spark.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql-kafka-0-10_${scala.version}</artifactId><version>${spark.version}</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>
<!-- kafka_learn\pom.xml -->

3、在 kafka_learn 工程中，创建 生产者 ProducerFastStart.java 类

/***  kafka_learn\src\main\java\djh\it\kafka\learn\chapter1\ProducerFastStart.java**  2024-6-21 创建 生产者 ProducerFastStart.java 类*/
package djh.it.kafka.learn.chapter1;import io.protostuff.StringSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;public class ProducerFastStart {//private static final String brokerList = "localhost:9092";private static final String brokerList = "172.18.30.110:9092";private static final String topic = "heima";public static void main( String[] args ) {Properties properties = new Properties();//1）设置 key 序列化器properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");//properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//2)设置重试次数properties.put(ProducerConfig.RETRIES_CONFIG, 10);//3)设置值序列化器properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");//properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//4)设置集群地址properties.put("bootstrap.servers", brokerList);//properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);ProducerRecord<String,String> record = new ProducerRecord<>(topic, "kafka-demo", "hello-kafka-test!");try{producer.send(record);}catch (Exception e){e.printStackTrace();}producer.close();}
}

在这里插入图片描述

七、kafka java 第一个程序 002

1、在 kafka_learn 工程中，创建 消费者 ConsumerFastStart.java 类


/***  kafka_learn\src\main\java\djh\it\kafka\learn\chapter1\ConsumerFastStart.java**  2024-6-21 创建 消费者 ConsumerFastStart.java 类*/
package djh.it.kafka.learn.chapter1;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class ConsumerFastStart {//private static final String brokerList = "localhost:9092";private static final String brokerList = "172.18.30.110:9092";private static final String topic = "heima";private static final String groupId = "group.demo";public static void main( String[] args ) {Properties properties = new Properties();//1）设置 key 序列化器properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//2)设置值序列化器properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");//3)设置集群地址properties.put("bootstrap.servers", brokerList);properties.put("group.id", groupId);KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);consumer.subscribe(Collections.singletonList(topic));while (true){ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String> record : records){System.out.println(record.value());}}}
}

2、在 kafka_learn 工程中，启动 消费者 ConsumerFastStart.java 类 和 生产者 ProducerFastStart.java 类，进行测试。会发现，消费者这边接收到，生产者发送的消费。

消费者接收消息.png

八、kafka 优化与总结

1、在 kafka_learn 工程中，修改 生产者 ProducerFastStart.java 类 ，进行优化代码。


/***  kafka_learn\src\main\java\djh\it\kafka\learn\chapter1\ProducerFastStart.java**  2024-6-21 创建 生产者 ProducerFastStart.java 类*/
package djh.it.kafka.learn.chapter1;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
//注意导包，一定要导成 kafka 的序列化包
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;public class ProducerFastStart {//private static final String brokerList = "localhost:9092";private static final String brokerList = "172.18.30.110:9092";private static final String topic = "heima";public static void main( String[] args ) {Properties properties = new Properties();//1）设置 key 序列化器 -- 优化代码//properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//2)设置重试次数 -- 优化代码properties.put(ProducerConfig.RETRIES_CONFIG, 10);//3)设置值序列化器 -- 优化代码//properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//4)设置集群地址 -- 优化代码//properties.put("bootstrap.servers", brokerList);properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);ProducerRecord<String,String> record = new ProducerRecord<>(topic, "kafka-demo", "优化-2024-6-21-kafka-test!");try{producer.send(record);}catch (Exception e){e.printStackTrace();}producer.close();}
}

2、在 kafka_learn 工程中，修改 消费者 ConsumerFastStart.java 类，进行优化代码。


/***  kafka_learn\src\main\java\djh\it\kafka\learn\chapter1\ConsumerFastStart.java**  2024-6-21 创建 消费者 ConsumerFastStart.java 类*/
package djh.it.kafka.learn.chapter1;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
//注意导包，一定要导成 kafka 的序列化包
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class ConsumerFastStart {//private static final String brokerList = "localhost:9092";private static final String brokerList = "172.18.30.110:9092";private static final String topic = "heima";private static final String groupId = "group.demo";public static void main( String[] args ) {Properties properties = new Properties();//1）设置 key 序列化器 -- 优化代码//properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//2)设置值序列化器 -- 优化代码//properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//3)设置集群地址 -- 优化代码//properties.put("bootstrap.servers", brokerList);properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);//properties.put("group.id", groupId);properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(properties);consumer.subscribe(Collections.singletonList(topic));while (true){ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));for(ConsumerRecord<String,String> record : records){System.out.println(record.value());}}}
}

在这里插入图片描述

3、初识 kafka 总结：

kafka 启动前，在进行配置 kafka 参数时，即修改 kafka/kafka_2.12-2.8.0/config/server.properties 配置文件时，需要注意以下几点：

  • 1） zookeeper.connect

指明 Zookeeper 主机地址，如果 zookeeper 是集群则以逗号隔开，如:172.6.14.61:2181,172.6.14.62:2181,172.6.14.63:21B1

  • 2） listeners 监听列表 配置：

broker 对外提供服务时绑定的 IP 和端口。多个以逗号隔开，如果监听器名称不是一个安全的协议,listener.security.protocol.map 也必须设置。主机名称设置 0.0.0.0 绑定所有的接口，主机名称为空则绑定默认的接口。如:PLAINTEXT://myhost:9092,SSL:/1:9091 CLIENT://0.0.0.0:9092,REPLICATION://localhost:9093

  • 3） broker.id

broker 的唯一标识符，如果不配置则自动生成，建议配置且一定要保证集群中必须唯一，默认-11.4.4

  • 4） log.dirs

日志数据存放的目录，如果没有配置则使用 log.dir，建议此项配置。

  • 5） message.max.bytes

服务器接受单个消息的最大大小，默认 1000012 约等于 976.6KB。


# 切换目录
cd /usr/local/kafka/kafka_2.12-2.8.0/# 修改 kafka 配置参数：
vim config/server.properties# server.properties 配置中需要关注以下几个参数:# 1）Kafka 所需 Zookeeper 集群地址
zookeeper.connect=lolalhost:2181 # 2）listeners 监听列表 配置：brokder 对外提供的服务入口地址（ kafka 监听地址：你的虚拟机 IP 地址）
# listeners=PLAINTEXT://127.0.0.1:9092 
listeners=PLAINTEXT://172.18.30.110:9092 # 3）表示 broker 的编号，如果集群中有多个 broker，则每个 broker 的编号需要设置的不同
broker.id=0 # 4）log.dirs 设置存放消息日志文件的地址
log.dirs=/tmp/kafka/log

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/357452.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

CentOS系统查看版本的各个命令

CentOS系统查看版本的各个命令

cat /etc/centos-release 查看CentOS版本 uname -a 命令的结果分别代表&#xff1a;当前系统的内核名称、主机名、内核发型版本、节点名、系统时间、硬件名称、硬件平台、处理器类型以及操作系统名称 cat /proc/version 命令用于查看Linux内核的版本信息。执行该命令后&#xf…
阅读更多...
【Linux基础】SSH登录

【Linux基础】SSH登录

SSH简介 安全外壳协议&#xff08;Secure Shell Protocol&#xff0c;简称SSH&#xff09;是一种加密的网络传输协议&#xff0c;可在不安全的网络中为网络服务提供安全的传输环境。 SSH通过在网络中建立安全隧道来实现SSH客户端与服务器之间的连接。 SSH最常见的用途是远程登…
阅读更多...
为什么用excel求出的和是错误的？

为什么用excel求出的和是错误的？

Excel中求和结果错误的原因可能有几种常见的情况&#xff1a;1. **数据格式问题**&#xff1a;有时候数字可能被错误地视为文本格式。这种情况下&#xff0c;Excel 在求和时会忽略这些单元格。你可以通过将这些单元格的格式改为数值格式来解决。2. **隐藏的行或列**&#xff1a…
阅读更多...
Java程序之动物声音“模拟器”

Java程序之动物声音“模拟器”

题目&#xff1a; 设计一个“动物模拟器”&#xff0c;希望模拟器可以模拟许多动物的叫声和行为&#xff0c;要求如下&#xff1a; 编写接口Animal&#xff0c;该接口有两个抽象方法cry()和getAnimalName()&#xff0c;即要求实现该接口的各种具体的动物类给出自己的叫声和种类…
阅读更多...
Linux 特殊变量 $?

Linux 特殊变量 $?

一. 说明 在 Linux 和其他类 Unix 系统中&#xff0c;$? 是一个特殊的变量&#xff0c;用于获取上一个命令的退出状态码。 退出状态码是一个整数值&#xff0c;通常用来表示命令的执行结果。 ⏹退出状态码的含义 0&#xff1a;命令成功执行。0以外的数字&#xff1a;命令执…
阅读更多...
[实践篇]13.29 再来聊下Pass Through设备透传

[实践篇]13.29 再来聊下Pass Through设备透传

写在前面 为什么要再聊天Pass Through？ 因为在QNX + Linux Android的技术方案下，我们会遇到LA发生reboot或异常panic后，无法正常开机。而再次异常的原因确实最头疼的Memory Corruption。观察下来是由于一些DMA外设如使用UART的一些设备在重启或panic后，没有正常走Shutdow…
阅读更多...
【Ubuntu】--- 创建用户 删除用户 及其他用户操作大全 持续更新中

【Ubuntu】--- 创建用户 删除用户 及其他用户操作大全 持续更新中

在编程的艺术世界里&#xff0c;代码和灵感需要寻找到最佳的交融点&#xff0c;才能打造出令人为之惊叹的作品。而在这座秋知叶i博客的殿堂里&#xff0c;我们将共同追寻这种完美结合&#xff0c;为未来的世界留下属于我们的独特印记。 【Ubuntu】--- 创建用户 删除用户 及其他…
阅读更多...
漏洞挖掘 | 记一次src挖掘-小程序敏感信息泄露

漏洞挖掘 | 记一次src挖掘-小程序敏感信息泄露

权当是一次漏洞挖掘的思路分享 闲言 就现在的一个web漏洞挖掘强度还是非常高的&#xff0c;所以我们不妨把我们的眼光投向一个之前可能未曾涉及到的区域———小程序 是的微信小程序&#xff0c;这玩意的防范能力和过滤能力其实对比web方向是要弱小很多的 进入正题 以下就是…
阅读更多...
Linux——man帮助命令

Linux——man帮助命令

一、man 获得帮助信息 基本语法&#xff1a;man [命令或配置文件] &#xff08;功能描述&#xff1a;获得帮助信息&#xff09; 查看 ls 命令的帮助信息 [roothadoop101 ~]# man ls man [数字] [函数] 1、Standard commands &#xff08;标准命令&#xff09; 2、System…
阅读更多...
基于Vue3.0 Node.js 的 大文件切片上传、秒传、断点续传实现方案梳理

基于Vue3.0 Node.js 的 大文件切片上传、秒传、断点续传实现方案梳理

✨&#x1f4bb; 在处理大文件上传时&#xff0c;切片上传是提高效率与用户体验的关键技术之一。下面将详细介绍如何在前端利用Vue框架与Node.js后端配合&#xff0c;实现这一功能。 &#x1f446;&#x1f3fb;大体流程 &#x1f446;&#x1f3fb;一、文件切片上传 通过文件…
阅读更多...
HTTP网络协议

HTTP网络协议

1.HTTP &#xff08;1&#xff09;概念&#xff1a; Hyper Text Transfer Protocol&#xff0c;超文本传输协议规定了浏览器和服务器之间数据传输的规则。 &#xff08;2&#xff09;特点 基于TCP协议:面向连接&#xff0c;安全基于请求-响应模型的:一次请求对应一次响应HTTP协…
阅读更多...
贪心推公式——AcWing 125. 耍杂技的牛

贪心推公式——AcWing 125. 耍杂技的牛

贪心推公式 定义 贪心算法是一种在每一步选择中都采取在当前状态下最优的选择&#xff0c;希望通过局部的最优选择来得到全局最优解的算法策略。 运用情况 问题具有最优子结构&#xff0c;即一个问题的最优解包含其子问题的最优解。可以通过局部最优决策逐步推导到全局最优…
阅读更多...
华为数通——ACL

华为数通——ACL

ACL基本介绍 ACL:访问控制列表&#xff0c;通过端口对数据流进行过滤&#xff0c;ACL判别依据是五元组&#xff1a;源IP地址&#xff0c;源端口&#xff0c;目的IP地址&#xff0c;目的端口、协议。&#xff08;ACL工作于OSI模型第三层&#xff0c;是路由器和三层交换机接口的…
阅读更多...
【Golang - 90天从新手到大师】Day06 - 数组

【Golang - 90天从新手到大师】Day06 - 数组

系列文章合集 Golang - 90天从新手到大师 数组是golang中最常用的一种数据结构,数组就是同一类型数据的有序集合 定义一个数组 格式: var name [n]type n为数组长度,n>0 且无法修改,type为数组的元素类型如: var a [2]int上面的例子定义了一个长度为2,元素类型为int的数组…
阅读更多...
Dubbo快速入门

Dubbo快速入门

1. Dubbo概述 官网地址&#xff1a;https://cn.dubbo.apache.org/zh-cn/ Apache Dubbo 是一款高性能的轻量级的Java RPC框架&#xff0c;可以和Spring框架无缝集成。 本地调用&#xff1a;本机调用&#xff0c;指同个JVM内部的方法调用&#xff0c;例如三层架构之间的方法调用…
阅读更多...
虚拟机IP地址频繁变化的解决方法

虚拟机IP地址频繁变化的解决方法

勾八动态分配IP&#xff0c;让我在学习redis集群的时候&#xff0c;配置很多的IP地址&#xff0c;但是由于以下原因导致我IP频繁变动&#xff0c;报错让我烦恼&#xff01;&#xff01;&#xff01;&#xff01; 为什么虚拟机的IP地址会频繁变化&#xff1f; 虚拟机IP地址频繁…
阅读更多...
NV-Embed论文阅读笔记

NV-Embed论文阅读笔记

这是NVIDIA的一篇论文&#xff0c;LLM通常使用的是GPT的decoder范式作为一个生成模型&#xff0c;文章探讨如何利用这样的decoder生成模型来实现BERT这样的encoder的功能&#xff0c;即提取有效的embedding。现有的方法提取embedding的方式无非是 1 mean pooling&#xff1b; 2…
阅读更多...
32 - 判断三角形（高频 SQL 50 题基础版)

32 - 判断三角形（高频 SQL 50 题基础版)

32 - 判断三角形 select *,if(xy>z and xz>y and zy > x,Yes,No) triangle fromTriangle;
阅读更多...
基于simulink的PEM燃料电池控制系统建模与仿真,对比PID,积分分离以及滑模控制器

基于simulink的PEM燃料电池控制系统建模与仿真,对比PID,积分分离以及滑模控制器

目录 1.课题概述 2.系统仿真结果 3.核心程序与模型 4.系统原理简介 4.1 PID控制器 4.2 积分分离PID控制器 4.3 滑模控制器 5.完整工程文件 1.课题概述 基于simulink的PEM燃料电池控制系统建模与仿真,对比PID,积分分离以及滑模控制器。 2.系统仿真结果 &#xff08;完…
阅读更多...
2024全国高校名单发布，电子版下载！

2024全国高校名单发布，电子版下载！

今天&#xff0c;教育部网站发布了《全国高等学校名单》。截至2024年6月20日&#xff0c;全国高等学校共计3117所&#xff0c;其中&#xff1a;普通高等学校2868所&#xff0c;含本科学校1308所、高职&#xff08;专科&#xff09;学校1560所&#xff1b;成人高等学校249所。本…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023