devbox
很楠不爱3
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

kafka教程大全

作者：很楠不爱3 | 2024-03-24 12:45:22

赞

踩

kafka教程

目录

什么是kafka?

kafka的核心概念

kafka高效文件特点

kafka的消息模型

安装kafka

启动 kafka

停止 kafka

kafka 命令行语句

创建 topic

查看 topic 列表

生产者生产消息（发送消息）

消费者消费消息（接收消息）

删除 topic

查看 topic 详细信息

配置点对点消息模型

配置发布/订阅消息模型

Java整合kafka

kafka topic API

kafka生产者API

kafka的顺序消费

常用参数说明

生产者发送消息

自定义生产者的 partition 分区

kafka 消费者 API

消费者消费机制和分区策略说明

kafka消息分配策略

常用参数说明

消费者接收消息

SpringBoot整合

kafka集群环境搭建

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

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

什么是kafka?

Kafka是最初由Linkedin公司开发，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目，也是一个开源【分布式流处理平台】，由Scala和Java编写，（也当做MQ系统，但不是纯粹的消息系统）

核心：一种高吞吐量的分布式流处理平台，它可以处理消费者在网站中的所有动作流数据。比如网页浏览，搜索和其他用户的行为等，应用于大数据实时处理领域。

kafka的核心概念

  • Broker

Kafka的服务端程序，可以认为一个mq节点就是一个broker，broker存储topic的数据

  • Producer生产者

创建消息Message，然后发布到MQ中。该角色将消息发布到Kafka的topic中

  • Consumer消费者

消费队列中的消息

  • ConsumerGroup消费者组

同一个topic, 可以广播发送给不同的group，一个group中只有一个consumer可以消费此消息

  • Topic

每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic，主题的意思

  • Partition分区

kafka数据存储的基本单元，topic中的数据分割为一个或多个partition，每个topic至少有一个partition，是有序的一个Topic的多个partitions, 被分布在kafka集群中的多个server上。

partition是一个有序的队列，以文件夹的形式存储在Broker本机上。
消费者数量 <=小于或者等于Partition数量。

Kafka 采取了分片和索引机制，将每个partition分为多个segment，每个segment对应2个文件 log 和 index，log默认大小配置log.segment.bytes

  • Replication副本（备份）

同个Partition会有多个副本replication ，多个副本的数据是一样的，当其他broker挂掉后，系统可以主动用副本提供服务。
默认每个topic的副本都是1（默认是没有副本，节省资源），也可以在创建topic的时候指定如果当前kafka集群只有3个broker节点，则replication-factor最大就是3了，如果创建副本为4，则会报错。

  • ReplicationLeader、ReplicationFollower

Partition有多个副本，但只有一个replicationLeader负责该Partition和生产者消费者交互。
ReplicationFollower只是做一个备份，从replicationLeader进行同步

  • ReplicationManager

负责Broker所有分区副本信息，Replication副本状态切换

  • offset偏移量
  • 每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中。
  • partition中的每个消息都有一个连续的序列号叫做offset，用于partition唯一标识一条消息（记录自己消费到哪里了的标记）。
  • 可以认为offset是partition中Message的id。kafka把offset保存在消费端的消费者组里
  • LEO（LogEndOffset）

表示每个partition的log最后一条Message的位置。

  • HW（HighWatermark）
  • 表示partition各个replicas数据间同步且一致的offset位置，即表示allreplicas已经commit的位置
  • HW之前的数据才是Commit后的，对消费者才可见
  • ISR集合里面最小leo

HW的作用：保证消费数据的一致性和副本数据的一致性

  • Follower故障
    • Follower发生故障后会被临时踢出ISR（动态变化），待该follower恢复后，follower会读取本地的磁盘记录的上次的HW，并将该log文件高于HW的部分截取掉，从HW开始向leader进行同步，等该follower的LEO大于等于该Partition的hw，即follower追上leader后，就可以重新加入ISR。
  • Leader故障
    • Leader发生故障后，会从ISR中选出一个新的leader，为了保证多个副本之间的数据一致性，其余的follower会先将各自的log文件高于hw的部分截掉（新leader自己不会截掉），然后从新的leader同步数据。

  • Segment
  • 每个topic可以有多个partition，而每个partition又由多个segment file组成。
  • segment file 由2部分组成，分别为index file和data file（log file），两个文件是一一对应的，后缀”.index”和”.log”分别表示索引文件和数据文件。
  • 命名规则：partition的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset+1。
  • ISR (in-sync replica set )
  • leader会维持一个与其保持同步的replica集合，该集合就是ISR，每一个leader partition都有一个ISR，leader动态维护, 要保证kafka不丢失message，就要保证ISR这组集合存活（至少有一个存活），并且消息commit成功
  • Partition leader 保持同步的 Partition Follower 集合, 当 ISR 中的Partition Follower 完成数据的同步之后，就会给 leader 发送 ack
  • 如果Partition follower长时间(replica.lag.time.max.ms) 未向leader同步数据，则该Partition Follower将被踢出ISR
  • Partition Leader 发生故障之后，就会从 ISR 中选举新的 Partition Leader。
  • OSR （out-of-sync-replica set）

与leader副本分区 同步滞后过多的副本集合

  • AR（Assign Replicas）

分区中所有副本统称为AR

说明

  • Topic可以指定副本数量，多个副本位于多台机器上。
  • kafka使用zookeeper在多个副本中选出一个leader，其他副本作为follower。
  • leader主要负责读写消息，也就是和生产者、消费者打交道，同时将消息同步写到其他副本中。
  • 如果某个broker宕机时，topic没有了leader，则会重新选举出新的leader。
  • 一个topic的多个partitions，被分在kafka集群中的多个broker上。
  • kafka保证同一个partition的多个repllcation一定不会分配在同一台broker上。

总结

  • 多订阅者
    • 一个topic可以有一个或者多个订阅者。
    • 每个订阅者都要有一个partition，所以订阅者数量要少于等于partition数量。
  • 高吞吐量、低延迟: 每秒可以处理几十万条消息。
  • 高并发：几千个客户端同时读写。
  • 容错性：多副本、多分区，允许集群中节点失败，如果副本数据量为n,则可以n-1个节点失败。
  • 扩展性强：支持热扩展

kafka高效文件特点

  • Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。
  • index索引文件查找，利用分段和稀疏索引。
  • producer生产数据，要写入到log文件中，写的过程中一直追加到文件末尾，为顺序写，官网数据表明。同样的磁盘，顺序写能到600M/S，而随机写只有100K/S。
  • 异步操作少阻塞sender和main线程，批量操作(batch)。
  • 零拷贝ZeroCopy。

kafka的消息模型

  • 点对点（point to point)
  • 消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息
  • 消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。 Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费
  • 发布/订阅（publish/subscribe）
  • 消息生产者（发布）将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息。
  • 和点对点方式不同，发布到topic的消息会被所有订阅者消费。

安装kafka

下载地址：Apache Kafka

安装需要的环境

  • JDK1.8+
  • Zookeeper    (ZooKeeper安装教程)
  • kafka-xxx-yyy：xxx是scala版本，yyy是kafka版本（scala是基于jdk开发，需要安装jdk环境）

将下载的 kafka 安装包上传到 linux 环境的 /usr/local 目录，进行tar -zxvf 解压缩

解压缩完成后进入到 kafka/config目录

配置 server.properties 文件

# brokerid，配置集群时每个服务器上的brokerid要配置成不一样的

broker.id=0       

# 监听地址，一般配置成局域网地址

listeners=PLAINTEXT://:9092            # 示例：listeners=PLAINTEXT://192.168.189.82:9092

# 监听地址，一般配置成外网地址，如果没有配置将使用listeners地址

advertised.listeners=PLAINTEXT://your.host.name:9092

# 持久化存储路径，多个地址的话用逗号分割,多个目录分布在不同磁盘上可以提高读写性能

log.dirs=/usr/local/kafka/data/logs

# 指定 partition 的数量
num.partitions=1

# 配置 zookeeper 连接的地址，集群连接使用英式逗号分隔

zookeeper.connect=localhost:2181

# 启动启用日志清理，true启用，false关闭

log.cleaner.enable=true

# 清理日志线程数

log.cleaner.threads=2

# 清理日志使用的策略，delete:删除，compact:压缩，根据key进行整理，有相同key不同

# value值，只保留最后一个

log.cleanup.policy=delete

# 定时任务检测删除日志间隔时间，默认5分钟

log.retention.check.interval.ms=300000

# 删除超过指定时间的消息,默认是168小时，7天
# 还有log.retention.ms, log.retention.minutes, log.retention.hours，优先级从高到低
log.retention.hours=168

# 超过指定大小后，删除旧的消息，下面是1G的字节数，-1就是没限制，log.retention.bytes

# 和log.retention.hours任意一个达到要求，都会执行删除
log.retention.bytes=1073741824

# 更多参数说明请查看 https://blog.csdn.net/lizhitao/article/details/25667831

启动 kafka

  1. # cd 到 bin 目录下 
  2. # 启动 kafka 之前要先启动 zookeeper
  3.  
  4. ./kafka-server-start.sh ../config/server.properties 
  5.  
  6. # 守护进程启动
  7. ./kafka-server-start.sh -daemon ../config/server.properties  &
  8.

停止 kafka

./kafka-server-stop.sh

kafka 命令行语句

创建 topic

  1. # --bootstrap-server：指定kafka地址，注意要和server.properties配置文件里的listeners地址保持一致，集群的方式使用,连接多个地址
  2. # --replication-factor：指定副本数量，副本数量不能超过 broker 数量
  3. # --partitions：指定分区数量
  4. # --topic：指定 topic 名称
  5. ./kafka-topics.sh --create --bootstrap-server 192.168.189.82:9092 --replication-factor 1 --partitions 1 --topic mytopic

创建成功的 topic 会在 log.dirs 目录下创建和分区数量一致的的文件夹（有几个分区就创建几个文件夹）

查看 topic 列表

  1. # --bootstrap-server：指定kafka地址，注意要和server.properties配置文件里的listeners地址保持一致
  2. ./kafka-topics.sh --list --bootstrap-server 192.168.189.82:9092

生产者生产消息（发送消息）

  1. # --broker-list ： 指定 kafka 地址，注意要和server.properties配置文件里的listeners地址保持一致
  2. # --topic ： 指定要向哪个主题发送消息，不存在的主题将被创建
  3. ./kafka-console-producer.sh --broker-list 192.168.189.82:9092 --topic mytopic

消费者消费消息（接收消息）

  1. # --bootstrap-server：指定kafka地址，注意要和server.properties配置文件里的listeners地址保持一致
  2. # --from-beginning：会把主题中之前未消费的消息进行消费，不带这个参数是从当前进行消费
  3. # --topic：指定主题
  4. ./kafka-console-consumer.sh --bootstrap-server 192.168.189.82:9092 --from-beginning --topic mytopic

删除 topic

  1. # --bootstrap-server：指定kafka地址，要和server.properties配置文件里的listeners地址保持一致
  2. # --delete：删除
  3. # --topic：指定要删除的主题名称
  4. ./kafka-topics.sh --bootstrap-server 192.168.189.82:9092 --delete --topic mytopic

查看 topic 详细信息

./kafka-topics.sh --describe --bootstrap-server 192.168.189.82:9092 --topic mytopic

配置点对点消息模型

1.配置 config/consumer.properties 文件

# 配置 kafka服务地址，多个服务使用逗号分隔

bootstrap.servers=localhost:9092

# 消费者组id

group.id=test-consumer-group

2.创建一个topic

./kafka-topics.sh --create --bootstrap-server 192.168.189.82:9092 --replication-factor 1 --partitions 2 --topic t1

3.开启一个生产者，两个消费者，进行测试

只有一个消费者可以消费消息

生产者

./kafka-console-producer.sh --broker-list 192.168.189.82:9092 --topic t1

消费者一

  1. # --consumer.config：指定consumer.properties配置文件的路径
  2. ./kafka-console-consumer.sh --bootstrap-server 192.168.189.82:9092 --from-beginning --topic t1 --consumer.config ../config/consumer.properties

消费者二        // 命令和消费者一一样

  1. # --consumer.config：指定consumer.properties配置文件的路径
  2. ./kafka-console-consumer.sh --bootstrap-server 192.168.189.82:9092 --from-beginning --topic t1 --consumer.config ../config/consumer.properties

注意：只有在有多个分区的时候，消费策略才会生效，如果测试发现没有负载均衡或者范围，请查看分区数量是不是多个

配置发布/订阅消息模型

1.将 config/consumer.properties 配置文件，拷贝2份，保证 group.id 不同

consumer-1.properties 配置文件

bootstrap.servers=192.168.189.82:9092
group.id=consumer-group-1

consumer-2.properties 配置文件

bootstrap.servers=192.168.189.82:9092
group.id=consumer-group-2

2.创建一个topic

./kafka-topics.sh --create --bootstrap-server 192.168.189.82:9092 --replication-factor 1 --partitions 2 --topic t2

3.开启一个生产者，两个消费者，进行测试

多个消费者都可以消费相同的消息，效果如下

生产者

./kafka-console-producer.sh --broker-list 192.168.189.82:9092 --topic t2

消费者一

  1. # --consumer.config：指定consumer.properties配置文件的路径
  2. ./kafka-console-consumer.sh --bootstrap-server 192.168.189.82:9092 --from-beginning --topic t2 --consumer.config ../config/consumer-1.properties

消费者二

  1. # --consumer.config：指定consumer.properties配置文件的路径
  2. ./kafka-console-consumer.sh --bootstrap-server 192.168.189.82:9092 --from-beginning --topic t2 --consumer.config ../config/consumer-2.properties

Java整合kafka

pom 文件中添加依赖

  1. <dependency>
  2. 	<groupId>org.apache.kafka</groupId>
  3. 	<artifactId>kafka-clients</artifactId>
  4. 	<version>3.4.0</version>
  5. </dependency>

kafka topic API

  1. 	// 主题名称
  2. 	private static final String TOPIC_NAME = "mytopic_client";
  3.  
  4. 	// 创建 kafka 客户端
  5. 	public static AdminClient initAdminClient() {
  6. 		Properties properties = new Properties();
  7. 		// 指定 kafka 服务器地址，集群使用英式逗号分隔 "192.168.189.83:9092,192.168.189.84:9092,192.168.189.85:9092"
  8. 		properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.82:9092");
  9. 		return AdminClient.create(properties);
  10. 	}
  11.  
  12. 	// 创建 topic 主题
  13. 	@Test
  14. 	public void createTopicTest() throws ExecutionException, InterruptedException {
  15. 		AdminClient adminClient = initAdminClient();
  16. 		// 创建 topic主题对象，指定分区数量，副本数量
  17. 		NewTopic newTopic = new NewTopic(TOPIC_NAME, 2, (short) 1);
  18. 		// 创建topic，阻塞返回创建结果
  19. 		CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));
  20. 		// 创建结果，创建失败会报错
  21. 		result.all().get();
  22. 	}
  23.  
  24. 	// 查询 topic 列表
  25. 	@Test
  26. 	public void listTopicTest() throws ExecutionException, InterruptedException {
  27. 		AdminClient adminClient = initAdminClient();
  28. 		// 是否查询系统生成的 topic
  29. 		// ListTopicsResult listTopicsResult = adminClient.listTopics(new ListTopicsOptions().listInternal(true));
  30. 		ListTopicsResult listTopicsResult = adminClient.listTopics();
  31. 		Set<String> topics = listTopicsResult.names().get();
  32. 		for (String topic : topics) {
  33. 			System.err.println(topic);
  34. 		}
  35. 	}
  36.  
  37. 	// 删除 topic
  38. 	@Test
  39. 	public void deleteTopicTest() throws ExecutionException, InterruptedException {
  40. 		AdminClient adminClient = initAdminClient();
  41. 		DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Collections.singletonList(TOPIC_NAME));
  42. 		// 删除失败会报错
  43. 		deleteTopicsResult.all().get();
  44. 	}
  45.  
  46. 	// 查看 topic 详情
  47. 	@Test
  48. 	public void detailTopicTest() throws ExecutionException, InterruptedException {
  49. 		AdminClient adminClient = initAdminClient();
  50. 		DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(TOPIC_NAME));
  51. 		Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.allTopicNames().get();
  52. 		Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
  53. 		entries.forEach((map) -> System.err.println("name: " + map.getKey() + ", desc: " + map.getValue()));
  54. 	}
  55.  
  56. 	// 增加 topic 分区数量，分区只能增加，不能减少
  57. 	@Test
  58. 	public void incrPartitionTopicTest() throws ExecutionException, InterruptedException {
  59. 		HashMap<String, NewPartitions> infoMap = new HashMap<>(1);
  60.  
  61. 		AdminClient adminClient = initAdminClient();
  62. 		// 指定分区数量
  63. 		NewPartitions newPartitions = NewPartitions.increaseTo(3);
  64. 		// 给指定的topic设置分区数量
  65. 		infoMap.put(TOPIC_NAME, newPartitions);
  66.  
  67. 		CreatePartitionsResult clientPartitionsResult = adminClient.createPartitions(infoMap);
  68. 		clientPartitionsResult.all().get();
  69. 	}

kafka生产者API

生产者配置官网：Apache Kafka

ProducerRecord类的作用？

生产者发送到broker的流程步骤，一个 topic 有多个partition分区，每个分区又有多个副本。

  • 如果指定Partition ID,则PR被发送至指定Partition (ProducerRecord)。
  • 如果未指定Partition ID,但指定了Key, PR会按照hash(key)发送至对应Partition。
  • 如果未指定Partition ID也没指定Key，PR会按照默认 round-robin轮训模式发送到每个Partition。
    • 消费者消费partition分区默认是range模式。
  • 如果同时指定了Partition ID和Key, PR只会发送到指定的Partition (Key不起作用，代码逻辑决定)。
  • 注意：Partition有多个副本，但只有一个replicationLeader负责该Partition和生产者消费者交互。

Kafka的客户端发送数据到服务器，不是来一条就发一条，会经过内存缓冲区（默认是16KB），通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里，然后把很多消息收集到的Batch里面，再一次性发送到Broker上去的，从而实现高性能。

kafka的顺序消费

  • ProducerRecord 如果key为空，kafka使用默认的partitioner，使用RoundRobin算法将消息均衡地分布在各个partition上。
  • ProducerRecord  如果key不为空，kafka使用自己实现的hash方法对key进行散列，决定消息该被写到Topic的哪个partition，拥有相同key的消息会被写到同一个partition，实现顺序消息。

常用参数说明

# broker/kafka地址

bootstrap.servers

# 当producer向leader发送数据时，可以通过request.required.acks参数来设置数据可靠性

# ，数据不丢失。级别,分别是0, 1，all。

# 0：producer发送一次就不再发送了，不管是否发送成功，发送出去的消息还在半路，或者还没写入磁盘， Partition Leader所在Broker就直接挂了，客户端认为消息发送成功了，此时就会导致这条消息就丢失。

# 1：只要Partition Leader接收到消息而且写入本地磁盘，就认为成功了，不管他其他的Follower有没有同步过去这条消息了。问题：万一Partition Leader刚刚接收到消息，Follower还没来得及同步过去，结果Leader所在的broker宕机了就会导致消息丢失。

# all/-1：producer只有收到分区内所有副本的成功写入全部落盘的通知才认为推送消息成功。leader会维持一个与其保持同步的replica集合，该集合就是ISR，leader副本也在isr里面。

问题：如果在follower同步完成后，broker发送ack之前，leader发生故障，那么会造成数据重复。

问题：acks=all 就可以代表数据一定不会丢失了吗？

Partition只有一个副本，也就是一个Leader，任何Follower都没有。接收完消息后宕机，也会导致数据丢失，acks=all，必须跟ISR列表里至少有2个以上的副本配合使用。在设置request.required.acks=-1的同时，也要min.insync.replicas这个参数设定 ISR中的最小副本数是多少，默认值为1，改为 >=2，如果ISR中的副本数少于min.insync.replicas配置的数量时，客户端会返回异常。

acks

# 请求失败，生产者会自动重试，指定是0次，如果启用重试，则会有重复消息的可能性。
retries

# 每个分区未发送消息总字节大小,单位：字节，超过设置的值就会提交数据到服务端，默认

# 值是16KB
batch.size

# 默认值是0，消息是立刻发送的，即便batch.size缓冲空间还没有满，如果想减少请求的数

# 量，可以设置 linger.ms 大于0，即消息在缓冲区保留的时间，超过设置的值就会被提交到

# 服务端，通俗解释是，本该早就发出去的消息被迫至少等待了linger.ms时间，相对于这时

# 间内积累了更多消息，批量发送 减少请求。如果batch被填满或者linger.ms达到上限，满足

# 其中一个就会被发送
linger.ms

# buffer.memory是用来约束Kafka Producer能够使用的内存缓冲的大小的，默认值32MB。

# 如果buffer.memory设置的太小，可能导致消息快速的写入内存缓冲里，但Sender线程来不

# 及把消息发送到Kafka服务器。会造成内存缓冲很快就被写满，而一旦被写满，就会阻塞用

# 户线程，不让继续往Kafka写消息了。buffer.memory要大于batch.size，否则会报申请内存

# 不足的错误，不要超过物理内存，根据实际情况调整。
buffer.memory

# key的序列化器，将用户提供的 key和value对象ProducerRecord 进行序列化处理，

# key.serializer必须被设置，即使消息中没有指定key，序列化器必须是一个实现

# org.apache.kafka.common.serialization.Serializer接口的类，将key序列化成字节数组。

key.serializer
value.serializer  

生产者发送消息

  1. 	private static final String TOPIC_NAME = "mytopic_client";
  2.  
  3. 	public static Properties getProperties() {
  4. 		Properties props = new Properties();
  5.  
  6.         // 集群使用英式逗号分隔 "192.168.189.83:9092,192.168.189.84:9092,192.168.189.85:9092"
  7. 		props.put("bootstrap.servers", "192.168.189.82:9092");
  8. //      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.82:9092");
  9.  
  10. 		props.put("acks", "all");
  11. //		props.put(ProducerConfig.ACKS_CONFIG, "all");
  12.  
  13. 		props.put("retries", 0);
  14. //		props.put(ProducerConfig.RETRIES_CONFIG, 0);
  15.  
  16. 		props.put("batch.size", 16384);
  17. //		props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
  18.  
  19. 		props.put("linger.ms", 5);
  20. //      props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
  21.  
  22. 		props.put("buffer.memory", 33554432);
  23. //      props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
  24.  
  25.  
  26. 		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  27. //		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  28. 		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  29. //      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
  30.  
  31. 		return props;
  32. 	}
  33.  
  34. 	// 发送消息
  35. 	@Test
  36. 	public void sendTest() throws ExecutionException, InterruptedException {
  37. 		Properties properties = getProperties();
  38. 		Producer<String, String> producer = new KafkaProducer<>(properties);
  39. 		for (int i = 0; i < 3; i++) {
  40. 			Future<RecordMetadata> future = producer.send(new ProducerRecord<>(TOPIC_NAME, "my is key-" + i, "my is value-" + i));
  41. 			// 获取发送结果
  42. 			RecordMetadata recordMetadata = future.get();
  43. 			// topic - 分区编号@offset
  44. 			System.err.println("发送状态：" + recordMetadata.toString());
  45. 		}
  46. 		producer.close();
  47. 	}
  48.  
  49. 	/**
  50. 	 * 发送消息 回调函数
  51. 	 */
  52. 	@Test
  53. 	public void sendWithCallbackTest() {
  54. 		Properties properties = getProperties();
  55. 		Producer<String, String> producer = new KafkaProducer<>(properties);
  56. 		for (int i = 0; i < 3; i++) {
  57. 			producer.send(new ProducerRecord<>(TOPIC_NAME, "my is key-" + i, "my is value-" + i), new Callback() {
  58. 				@Override
  59. 				public void onCompletion(RecordMetadata metadata, Exception exception) {
  60. 					if (null == exception) {
  61. 						System.err.println("发送状态:" + metadata.toString());
  62. 					} else {
  63. 						exception.printStackTrace();
  64. 					}
  65. 				}
  66. 			});
  67. 		}
  68. 		producer.close();
  69. 	}
  70.  
  71. 	/**
  72. 	 * 发送消息 回调函数
  73. 	 * <p>
  74. 	 * 指定某个分区, 实现顺序消息
  75. 	 */
  76. 	@Test
  77. 	public void sendWithCallbackAndPartitionTest() {
  78. 		Properties properties = getProperties();
  79. 		Producer<String, String> producer = new KafkaProducer<>(properties);
  80. 		for (int i = 0; i < 10; i++) {
  81. 			// 第二个参数，发送到指定的分区
  82. 			producer.send(new ProducerRecord<>(TOPIC_NAME, 1, "my is key-" + i, "my is value-" + i), new Callback() {
  83. 				@Override
  84. 				public void onCompletion(RecordMetadata metadata, Exception exception) {
  85. 					if (exception == null) {
  86. 						// recordMetadata.toString() 打印：topic - 分区编号@offset
  87. 						System.err.println("发送状态:" + metadata.toString());
  88. 					} else {
  89. 						exception.printStackTrace();
  90. 					}
  91. 				}
  92. 			});
  93. 		}
  94. 		producer.close();
  95. 	}

自定义生产者的 partition 分区

  1. import org.apache.kafka.clients.producer.Partitioner;
  2. import org.apache.kafka.common.Cluster;
  3. import org.apache.kafka.common.PartitionInfo;
  4. import org.apache.kafka.common.utils.Utils;
  5.  
  6. import java.util.List;
  7. import java.util.Map;
  8.  
  9. public class MyPartitioner implements Partitioner {
  10. 	@Override
  11. 	public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
  12. 		if ("my is key".equals(key)) {	// 自定义如果key相同，走分区0
  13. 			return 0;
  14. 		}
  15. 		List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
  16. 		int numPartitions = partitions.size();
  17. 		//使用hash值取模，确定分区(默认的也是这个方式)
  18. 		return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
  19. 	}
  20.  
  21. 	@Override
  22. 	public void close() {
  23. 	}
  24.  
  25. 	@Override
  26. 	public void configure(Map<String, ?> configs) {
  27. 	}
  28. }

使用自定义分区

  1.     Properties properties = getProperties();
  2.     properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.xxx.MyPartitioner");
  3.     Producer<String, String> producer = new KafkaProducer<>(properties);

kafka 消费者 API

消费者消费机制和分区策略说明

  • 消费者使用什么模式从broker获取数据的？
  • 消费者采用 pull 拉取方式，从broker的partition获取数据。
  • pull 模式则可以根据 consumer 的消费能力进行自己调整，不同的消费者性能不一样。
    • 如果broker没有数据，consumer可以配置 timeout 时间，阻塞等待一段时间之后再返回。
  • 如果是broker主动push，优点是可以快速处理消息，但是容易造成消费者处理不过来，消息堆积和延迟。
  • 消费者从哪个分区进行消费？

顶层接口：org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor

kafka消息分配策略

策略一：round-robin轮询

具体实现类是org.apache.kafka.clients.consumer.RoundRobinAssignor

消费者在同一个消费者组的，所订阅的消息不同的情况下：

轮询分配策略是基于所有可用的消费者和所有可用的分区的，然后按照轮询顺序给每一个消费者一个分区，如果消费者数量和分区数量一致那每一个消费者都会分配到一个分区，这没有什么问题。

如果消费者多于分区数量，那么在把分区分配完后，剩下没有分配到分区的消费者会处于空闲没有消息的状态，这就比较浪费资源了。

如果消费者少于分区数量，那么在把消费者都轮询一边后，剩下的分区都会分配给最后一个消费者，这样最后一个消费者就会承担很多消费压力，导致消息阻塞超时，性能下降。

策略二：range范围（默认策略）

具体实现类是org.apache.kafka.clients.consumer.RangeAssignor

range策略是基于主题的，把主题中的分区按照数字顺序排序，把消费者按照消费者名称字典排序，排好序后每个主题当前的分区数除以当前所有订阅该主题的消费者，除不尽的情况下（还有剩余分区）排序靠前的消费者会多消费一个分区。

如果只有一个topic，前面的消费者多消费一个分区影响不大，假如有多个topic呢，那么前面的消费者就要消费每个topic的余下的一个分区，topic越多前面的消费者消费的分区越多，则会承担较多的消费压力，导致消息阻塞超时，性能下降。

重新分配机制

例子：有10个分区，5个消费者，消费者按照顺序启动，前面启动的消费者分配了10个分区，后面启动的消费者怎么分配？

当消费者组里的消费者数量发生变化（增加或减少）或者是 topic 中的分区数量发生变化时，就会触发消费者端的 rebalance（重新分配）操作。

消费者在消费过程中宕机了，重新恢复后从哪里消费，会有什么问题？

消费者会记录offset，故障恢复后从这里继续消费，这个offset老版本的时候记录在zookeeper里面和本地，新版默认将offset保存在kafka的内置topic中，名称是 __consumer_offsets

  • __consumer_offsets
    • 该Topic默认有50个Partition，每个Partition有3个副本，分区数量由参数offset.topic.num.partition配置。
    • 通过groupId的哈希值和该参数取模的方式来确定某个消费者组已消费的offset保存到__consumer_offsets主题的哪个分区中。
    • 由 消费者组名+主题+分区，确定唯一的offset的key，从而获取对应的值。
    • 三元组：group.id+topic+分区号，而 value 就是 offset 的值。

常用参数说明

# broker/kafka地址

bootstrap.servers

# 消费者分组ID，分组内的消费者只能消费该消息一次，不同分组内的消费者可以重复消费

# 该消息
group.id
​
# 是否自动提交偏移量，true：自动提交，false：不自动提交
enable.auto.commit
​
# 自动提交offset周期
auto.commit.interval.ms
​
# 重置消费偏移量策略，消费者在读取一个没有偏移量的分区或者偏移量无效情况下（因消

# 费者长时间失效、包含偏移量的记录已经过时并被删除）该如何处理？
# 默认是latest不从头消费，如果需要从头消费partition消息，需要改为 earliest，并且需要消

# 费者组名换一个名字，才能从头消费
auto.offset.reset
​
#序列化器
key.deserializer

value.deserializer

消费者接收消息

  1. 	private static final String TOPIC_NAME = "mytopic_client";
  2.  
  3. 	public static Properties getProperties() {
  4.  
  5. 		Properties props = new Properties();
  6.  
  7. 		//broker地址
  8. 		props.put("bootstrap.servers", "192.168.189.82:9092");
  9. //		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.189.82:9092");
  10.  
  11. 		//消费者分组ID，分组内的消费者只能消费该消息一次，不同分组内的消费者可以重复消费该消息
  12. 		props.put("group.id", "test-g1");
  13.  
  14. 		//默认是latest不从头消费，如果需要从头消费partition消息，需要改为 earliest，并且需要消费者组名换一个名字，才能从头消费
  15. 		props.put("auto.offset.reset", "earliest");
  16.  
  17. 		// 配置 消费策略（负载均衡还是范围）
  18. 		props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");
  19. //		props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
  20. //		props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RangeAssignor");
  21.  
  22. 		//开启自动提交offset
  23. 		//props.put("enable.auto.commit", "true");
  24. 		props.put("enable.auto.commit", "false");
  25.  
  26. 		//自动提交offset延迟时间
  27. 		//props.put("auto.commit.interval.ms", "1000");
  28.  
  29. 		//反序列化
  30. 		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  31. 		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  32.  
  33. 		return props;
  34. 	}
  35.  
  36. 	// 消费者接收消息
  37. 	@Test
  38. 	public void simpleConsumerTest() {
  39.  
  40. 		Properties properties = getProperties();
  41.  
  42. 		KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
  43.  
  44. 		//订阅主题
  45. 		kafkaConsumer.subscribe(Collections.singletonList(TOPIC_NAME));
  46.  
  47. 		while (true) {
  48.  
  49. 			//阻塞超时时间，如果 broker 中没有数据，等到指定的时间拉取一次
  50. 			ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
  51. 			for (ConsumerRecord record : records) {
  52. 				System.err.printf("topic=%s, offset=%d,key=%s,value=%s \n", record.topic(), record.offset(), record.key(), record.value());
  53. 			}
  54.  
  55. 			// 手动提交offset
  56. 			//同步阻塞的方式提交offset，自动失败重试
  57. //		kafkaConsumer.commitSync();
  58.  
  59. 			//异步提交offset，没有失败重试，可以在失败的地方记录日志
  60. 			if (!records.isEmpty()) {
  61.  
  62. 				kafkaConsumer.commitAsync(new OffsetCommitCallback() {
  63. 					@Override
  64. 					public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
  65. 						if (exception == null) {
  66. 							System.err.println("手工提交offset成功:" + offsets.toString());
  67. 						} else {
  68. 							System.err.println("手工提交offset失败:" + offsets.toString());
  69. 						}
  70. 					}
  71. 				});
  72. 			}
  73. 		}
  74. 	}

SpringBoot整合

pom 文件中添加依赖

  1. <dependency>
  2. 	<groupId>org.springframework.kafka</groupId>
  3. 	<artifactId>spring-kafka</artifactId>
  4. </dependency>

application.yml 配置

  1. spring:
  2.   kafka:
  3.     # kafka服务地址，多个使用英式逗号分隔
  4.     bootstrap-servers: 192.168.189.82:9092
  5.     # 生产者配置
  6.     producer:
  7.       # 消息重发的次数，如果配置了事务，retries的值必须大于0。
  8.       retries: 1
  9.       #一个批次可以使用的内存大小
  10.       batch-size: 16384
  11.       # 设置生产者内存缓冲区的大小。
  12.       buffer-memory: 33554432
  13.       # 键的序列化方式
  14.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  15.       # 值的序列化方式
  16.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  17.       # ack，如果配置了事务，这个值必须是 all或者-1
  18.       acks: all
  19.  
  20.       #事务id
  21.       transaction-id-prefix: lxx-tran-
  22.  
  23.     # 消费者配置
  24.     consumer:
  25.       # 自动提交的时间间隔 在spring boot 2.X 版本是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
  26.       auto-commit-interval: 1S
  27.       # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
  28.       auto-offset-reset: earliest
  29.       # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
  30.       enable-auto-commit: false
  31.       # 键的反序列化方式
  32.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  33.       # 值的反序列化方式
  34.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  35.  
  36.     listener:
  37.       #手工ack，调用ack.acknowledge()后立刻提交offset
  38.       ack-mode: manual_immediate
  39.       #监听器容器运行的线程数
  40.       concurrency: 4
  41.       # 主题不存在时失败，防止报错，true:主题不存在时不能启动，false:主题不存在时自动创建
  42.       missing-topics-fatal: false

生产者 发送消息 - 没有事务

  1. import org.junit.jupiter.api.Test;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.boot.test.context.SpringBootTest;
  4. import org.springframework.kafka.core.KafkaTemplate;
  5.  
  6. @SpringBootTest
  7. public class KafkaTests {
  8.  
  9. 	private static final String TOPIC_NAME = "mytopic_client";
  10.  
  11. 	@Autowired
  12. 	private KafkaTemplate<String, Object> kafkaTemplate;
  13.  
  14. 	@Test
  15. 	public void sendMessage() {
  16. 		kafkaTemplate.send(TOPIC_NAME, "我是消息")
  17. 				// 回调
  18. 				.addCallback(success -> {
  19. 					System.err.println("发送消息成功，消息详情：" + success.getRecordMetadata().toString());
  20. 				}, failure -> {
  21. 					System.err.println("发送消息失败：" + failure.getMessage());
  22. 				});
  23. 	}
  24. }

生产者 发送消息 - 有事务 注解的方式

  1. import org.springframework.beans.factory.annotation.Autowired;
  2. import org.springframework.kafka.core.KafkaTemplate;
  3. import org.springframework.stereotype.Service;
  4. import org.springframework.transaction.annotation.Transactional;
  5.  
  6. @Service
  7. public class TestService {
  8.  
  9. 	private static final String TOPIC_NAME = "mytopic_client";
  10.  
  11. 	@Autowired
  12. 	private KafkaTemplate<String, Object> kafkaTemplate;
  13.  
  14. 	@Transactional(rollbackFor = RuntimeException.class)
  15. 	public void sendMessage(int i) {
  16. 		kafkaTemplate.send(TOPIC_NAME, "我是消息 1 i=" + i);
  17. 		if (i == 0) {
  18. 			throw new RuntimeException();
  19. 		}
  20. 		kafkaTemplate.send(TOPIC_NAME, "我是消息 2 i=" + i);
  21. 	}
  22. }

生产者 发送消息 - 有事务 声明式事务

  1. import org.springframework.beans.factory.annotation.Autowired;
  2. import org.springframework.kafka.core.KafkaOperations;
  3. import org.springframework.kafka.core.KafkaTemplate;
  4. import org.springframework.stereotype.Service;
  5.  
  6. @Service
  7. public class TestService {
  8.  
  9. 	private static final String TOPIC_NAME = "mytopic_client";
  10.  
  11. 	@Autowired
  12. 	private KafkaTemplate<String, Object> kafkaTemplate;
  13.  
  14. 	public void sendMessage(Integer i) {
  15. 		kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, Object, Object>() {
  16. 			@Override
  17. 			public Object doInOperations(KafkaOperations<String, Object> kafkaOperations) {
  18. 				kafkaOperations.send(TOPIC_NAME, "我是消息 1 i=" + i);
  19. 				if (i == 0) {
  20. 					throw new RuntimeException();
  21. 				}
  22. 				kafkaOperations.send(TOPIC_NAME, "我是消息 2 i=" + i);
  23. 				return true;
  24. 			}
  25. 		});
  26. 	}
  27. }

消费者监听队列

  1. import org.apache.kafka.clients.consumer.ConsumerRecord;
  2. import org.springframework.kafka.annotation.KafkaListener;
  3. import org.springframework.kafka.support.Acknowledgment;
  4. import org.springframework.kafka.support.KafkaHeaders;
  5. import org.springframework.messaging.handler.annotation.Header;
  6. import org.springframework.stereotype.Component;
  7.  
  8. @Component
  9. public class TestMqListener {
  10.  
  11. 	private static final String TOPIC_NAME = "mytopic_client";
  12.  
  13. 	@KafkaListener(topics = {TOPIC_NAME}, groupId = "test-g1")
  14. 	public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
  15. 		// 打印出消息内容
  16. 		System.err.printf("消费消息，主题名称：%s，分区：%s，消息内容：%s \n", record.topic(), record.partition(), record.value());
  17. 		// 确认消费
  18. 		ack.acknowledge();
  19. 		// 拒绝消费
  20. //		ack.nack(1);
  21. 	}
  22. }

kafka集群环境搭建

准备 3 台kafka服务器，ip地址分别是

  1. 192.168.189.83
  2. 192.168.189.84
  3. 192.168.189.85

准备 3 台 已经配置好 zookeeper 的服务器，ip地址分别是

  1. 192.168.189.79
  2. 192.168.189.80
  3. 192.168.189.81

1：下载kafka，下载地址：Apache Kafka

2：编辑 kafka  config目录的 server.properties 配置文件，内容如下

  1. # 第一台机器
  2. broker.id=1
  3. port=9092
  4. listeners=PLAINTEXT://192.168.189.83:9092    // 配置私网IP
  5. # advertised.listeners=PLAINTEXT://公网ip:9092    // 配置公网IP
  6. log.dirs=/usr/local/kafka/logs
  7. zookeeper.connect=192.168.189.79:2181,192.168.189.80:2181,192.168.189.81:2181
  1. # 第二台机器
  2. broker.id=2
  3. port=9092
  4. listeners=PLAINTEXT://192.168.189.84:9092      
  5. # advertised.listeners=PLAINTEXT://公网ip:9092
  6. log.dirs=/usr/local/kafka/logs
  7. zookeeper.connect=192.168.189.79:2181,192.168.189.80:2181,192.168.189.81:2181
  1. # 第三台机器
  2. broker.id=3
  3. port=9092
  4. listeners=PLAINTEXT://192.168.189.85:9092      
  5. # advertised.listeners=PLAINTEXT://公网ip:9092
  6. log.dirs=/usr/local/kafka/logs
  7. zookeeper.connect=192.168.189.79:2181,192.168.189.80:2181,192.168.189.81:2181

3：分别启动 3台 kafka 服务

  1. ./kafka-server-start.sh ../config/server.properties 
  2.  
  3. # 守护进程启动
  4. ./kafka-server-start.sh -daemon ../config/server.properties  &

4：校验集群是否启动成功，可以创建一个 topic，查看数据

创建一个 3个备份，6个分区的 topic

./kafka-topics.sh --create --bootstrap-server 192.168.189.83:9092,192.168.189.84:9092,192.168.189.85:9092 --replication-factor 3 --partitions 6 --topic cluster-topic-test

进入到 log.dirs=/usr/local/kafka/logs目录，查看分区数据是否存在

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/很楠不爱3/article/detail/302417
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号