您的位置:首页 > 房产 > 建筑 > Kafka实战(Scala操作)

Kafka实战(Scala操作)

2024/12/23 16:25:56 来源:https://blog.csdn.net/qq_73339471/article/details/140893127  浏览:    关键词: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=allacks=-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()}")}}}
}

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com