Kafka实战（Scala操作）

Kafka基础讲解部分

Kafka基础讲解部分

Kafka实战（Scala操作）

1、引入依赖

版本：

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<spark.scala.version>2.12</spark.scala.version>
<spark.version>3.1.2</spark.version>
<spark.kafka.version>3.5.1</spark.kafka.version>
<kafka.version>2.8.0</kafka.version>

具体依赖：

<!-- spark-core -->
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_${spark.scala.version}</artifactId><version>${spark.version}</version>
</dependency><!-- spark-sql -->
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_${spark.scala.version}</artifactId><version>${spark.version}</version>
</dependency><!-- kafka -->
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>${kafka.version}</version>
</dependency>

2、创建生产者（Producer）

一：生产者相关配置讲解：

• BATCH_SIZE_CONFIG = "batch.size"：批处理数量，消息为batch.size大小，生产者才会发送消息

• LINGER_MS_CONFIG = "linger.ms"：延迟时间，如果消息大小迟迟不为batch.size大小，则可以在指定的时间linger.ms后发送

• RETRIES_CONFIG= "retry.count"：重试次数，消息发送失败时，生产者可以再重试retry.count次数

• ACKS_CONFIG= "acks"：ack机制，生产者需要等待aks个副本成功写入消息后，才认为消息发送成功

• acks一共有三个选项

  • acks=0：生产者不需要等待来自服务器的任何确认。一旦消息被发送出去，生产者就认为消息已经成功发送。
  • acks=1（默认值）：生产者需要等待分区中的leader副本成功写入消息并返回确认后，才认为消息发送成功。
  • acks=all 或 acks=-1：生产者需要等待ISR（In-Sync Replicas，同步副本集）中的所有副本都成功写入消息后，才认为消息发送成功。

• KEY_SERIALIZER_CLASS_CONFIG：键序列化

• VALUE_SERIALIZER_CLASS_CONFIG：值序列化

二：ProducerRecord讲解：

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {this(topic, partition, timestamp, key, value, null);
}
基本讲解：Topic: 指定消息要发送到的目标主题（topic）的名称。Partition: 可选的分区号。如果未指定分区号，则Kafka会根据键（如果有的话）或者使用默认的分区策略来决定消息被发送到哪个分区。Timestamp: 可选的时间戳。指定消息的时间戳，通常用来记录消息的产生时间Key: 消息的键（optional）。在Kafka中，消息可以有一个可选的键，用于分区（partition）消息。键通常是一个字符串或者字节数组。Value: 消息的实际内容，可以是任何序列化的数据。

异步发送的普通生产者

在异步发送模式下，生产者调用send()方法发送消息后，不会立即等待服务器的响应，而是继续执行后续操作。

import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerConfig, ProducerRecord, RecordMetadata}import java.util.Properties
import scala.util.Random// kafka生产者:将数据导入kafka中
object KafkaProducer {def main(args: Array[String]): Unit = {// 生产者相关配置val producerConf = new Properties()// 设置连接kafka(集群)配置【必配】producerConf.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"single:9092")// 批处理数量（每10条处理一次）producerConf.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"10")// 延迟时间:如果消息大小迟迟不为batch.size大小，则可以在指定的时间（50ms）后发送producerConf.setProperty(ProducerConfig.LINGER_MS_CONFIG,"50")// 消息发送失败时，生产者可以重试的次数producerConf.setProperty(ProducerConfig.RETRIES_CONFIG,"2")// ack机制：生产者需要等待1个副本成功写入消息后，才认为消息发送成功producerConf.setProperty(ProducerConfig.ACKS_CONFIG,"1")// 键值序列化producerConf.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.IntegerSerializer")producerConf.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")// 构造生产者producerval producer:KafkaProducer[Int,String] = new KafkaProducer(producerConf)// 发送消息val topic = "test02" // 指定主题（需存在）val rand = new Random()for(i<- 1 to 100000){// 封装待发送的消息val record: ProducerRecord[Int, String] = new ProducerRecord[Int, String](topic, 0, System.currentTimeMillis(), i, "test-"+i)// 向生产者发送记录producer.send(record)Thread.sleep(5+rand.nextInt(20))}// 强制将所有待发送的消息立即发送到 Kafka 集群，而不需要等待缓冲区填满或达到任何延迟阈值producer.flush()// 关闭并释放资源producer.close()}
}

异步发送的带回调函数生产者

在异步发送模式下，生产者调用send()方法发送消息后，不会立即等待服务器的响应，而是继续执行后续操作，带回调函数生产者可以在控制台中看到发送成功或失败信息。

import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerConfig, ProducerRecord, RecordMetadata}import java.util.Properties
import scala.util.Randomimport java.util.Properties// kafka生产者:将数据导入kafka中
object KafkaProducer {def main(args: Array[String]): Unit = {// 生产者相关配置val producerConf = new Properties()// 设置连接kafka(集群)配置【必配】producerConf.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"single:9092")// 批处理数量（每10条处理一次）producerConf.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"10")// 延迟时间:如果消息大小迟迟不为batch.size大小，则可以在指定的时间（50ms）后发送producerConf.setProperty(ProducerConfig.LINGER_MS_CONFIG,"50")// 消息发送失败时，生产者可以重试的次数producerConf.setProperty(ProducerConfig.RETRIES_CONFIG,"2")// ack机制：生产者在发送消息后需要等待来自服务器的确认级别（服务器给出一次确认，才算写入成功）producerConf.setProperty(ProducerConfig.ACKS_CONFIG,"1")// 键值序列化producerConf.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.IntegerSerializer")producerConf.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")// 构造生产者producerval producer:KafkaProducer[Int,String] = new KafkaProducer(producerConf)// 发送消息val topic = "test02" // 指定主题（需存在）val rand = new Random()for(i<- 1 to 100000){// 封装待发送的消息val record: ProducerRecord[Int, String] = new ProducerRecord[Int, String](topic, 0, System.currentTimeMillis(), i, "test-"+i)// 用生产者发送记录producer.send(record,new Callback {override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {if (exception == null){// 发送信息成功println("发送信息成功，topic："+metadata.topic()+",partition:"+metadata.partition()+",offset:"+metadata.offset())}else{// 发送信息失败exception.printStackTrace()}}})Thread.sleep(5+rand.nextInt(20))}// 强制将所有待发送的消息立即发送到 Kafka 集群，而不需要等待缓冲区填满或达到任何延迟阈值producer.flush()// 关闭并释放资源producer.close()}
}

3、创建消费者（Consumer）

一：生产者相关配置讲解：

• GROUP_ID_CONFIG：消费者组的标识符,有还几个消费者共用这个组进行消费同一个主题消息
• GROUP_INSTANCE_ID_CONFIG：消费者组中消费者实例的标识符，当前消费者自己在消费者组中的唯一标识
• MAX_POLL_RECORDS_CONFIG="records.count"：控制每次从 Kafka 主题中拉取的最大记录数为records.count
• MAX_POLL_INTERVAL_MS_CONFIG：控制两次连续拉取消息之间的最大时间间隔，若消费者在此时间间隔内没有发送心跳，则会被认为死亡
• HEARTBEAT_INTERVAL_MS_CONFIG="time"：控制消费者发送心跳的频率,每time毫秒发一次心跳
• ENABLE_AUTO_COMMIT_CONFIG：消费者自动提交偏移量（offset）给 Kafka
• AUTO_COMMIT_INTERVAL_MS_CONFIG="time"：控制自动提交偏移量的频率，即time毫秒提交一次偏移量
• PARTITION_ASSIGNMENT_STRATEGY_CONFIG="org.apache.kafka.clients.consumer.RoundRobinAssignor"：轮询分区分配策略：将所有可用分区依次分配给消费者，以平衡负载
• AUTO_OFFSET_RESET_DOC="latest"：偏移量策略：从分区的最新偏移量开始消费

二：消费者策略（读取数据方式）

Kafka为消费者提供了三种类型的订阅消费模式：subscribe（订阅模式）、SubscribePattern（正则订阅模式）、assign（指定模式）。

• subscribe（订阅模式）：

  // 订阅单个主题（test02主题）
  consumer.subscribe(Collections.singletonList("test02"));
  

• SubscribePattern（正则订阅模式）：

  // 使用正则表达式订阅主题
  consumer.subscribe(Pattern.compile("topic-.*"));
  

• assign（指定模式）：

  // 手动分配特定的分区给消费者（test02主题）
  TopicPartition partition0 = new TopicPartition("test02", 0);
  consumer.assign(Collections.singletonList(partition0));
  

package cha01import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecords, KafkaConsumer}
import java.time.Duration
import java.util.{Collections, Properties}
import scala.collection.JavaConverters.iterableAsScalaIterableConverter// 消费者：用SparkStreaming来消费kafka中数据
object KafkaConsumer {def main(args: Array[String]): Unit = {// 消费者相关配置val props = new Properties()// 设置连接kafka(集群)配置【必配】props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"single:9092")// 消费者组的标识符props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"zhou-240801")// 消费者组中消费者实例的标识符(独享的)props.setProperty(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG,"zhou")// 控制每次从 Kafka 主题中拉取的最大记录数props.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"1000")// 控制两次连续拉取消息之间的最大时间间隔，若消费者在此时间间隔内没有发送心跳，则会被认为死亡props.setProperty(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,"9000") // 9000毫秒(9秒)// 控制消费者发送心跳的频率,每1秒发一次心跳props.setProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,"1000") // 心跳：1000毫秒(1秒)// 消费者组的最小会话超时时间props.setProperty("group.min.session.timeout.ms","15000") // 15000毫秒(15秒)// 消费者组的最大会话超时时间props.setProperty("group.max.session.timeout.ms","50000") // 50000毫秒(50秒)// 消费者的会话超时时间设置props.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"20000") // 20000毫秒(20秒)// 消费者自动提交偏移量（offset）给 Kafkaprops.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true")// 控制自动提交偏移量的频率，即多久提交一次偏移量props.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"500") //500毫秒// 轮询分区分配策略：将所有可用分区依次分配给消费者，以平衡负载props.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor")// 偏移量策略：从分区的最新偏移量开始消费props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_DOC,"latest")// 键值反序列化props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.IntegerDeserializer") // 反序列化props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer") // 反序列化// 创建消费者val consumer = new KafkaConsumer[Int,String](props)// 订阅者模式val topic = "test02" // 主题consumer.subscribe(Collections.singletonList(topic))// 持续监听状态中while (true) {// 获取消息，每次获取消息的间隔为100msval records: ConsumerRecords[Int, String] = consumer.poll(Duration.ofMillis(100))for (record <- records.asScala) {println(s"topic:${record.topic()},key:${record.key()}, value:${record.value()}")}}}
}