使用Java实现数据的生产和消费

【Kafka】Java实现数据的生产和消费Kafka介绍Kafka 是由LinkedIn公司开发的它是一个分布式的支持多分区、多副本基于 Zookeeper 的分布式消息流平台它同时也是一款开源的基于发布订阅模式的消息引擎系统。Kafka术语Broker消息中间件处理节点一个Kafka节点就是一个Broker一个或者多个Broker可以组成一个Kafka集群Topic每条发布到Kafka集群的消息都有一个类别这个类别被称为Topic。物理上不同Topic的消息分开存储逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处PartitionPartition是物理上的概念每个Topic包含一个或多个PartitionProducer负责发布消息到Kafka BrokerConsumer消息消费者向Kafka Broker读取消息的客户端Consumer Group每个Consumer属于一个特定的Consumer Group可为每个Consumer指定Groupname若不指定Groupname则属于默认的GroupConsumer Offset消费者在消费消息的过程中记录消费者在分区中消费进度的字段就是消息位移它是一个偏移量随着消费者不断消费分区中的消息而递增ReplicaKafka 中消息的备份又叫做副本Replica副本的数量是可以配置的Kafka 定义了两类副本领导者副本Leader Replica 和 追随者副本Follower Replica前者对外提供服务后者只是被动跟随Rebalance当 Kafka 的某个主题的消费者组中有一个消费者不可用后其他消费者会自动重新分配订阅的主题分区这个过程叫做 Rebalance是 Kafka 实现消费者端高可用的重要手段。Kafka特性高吞吐、低延迟kakfa 最大的特点就是收发消息非常快kafka 每秒可以处理几十万条消息它的最低延迟只有几毫秒高伸缩性每个主题(topic) 包含多个分区(partition)主题中的分区可以分布在不同的主机(broker)中持久性、可靠性Kafka 能够允许数据的持久化存储消息被持久化到磁盘并支持数据备份防止数据丢失Kafka 底层的数据存储是基于 Zookeeper 存储的Zookeeper 的数据能够持久存储容错性允许集群中的节点失败某个节点宕机Kafka 集群能够正常工作高并发支持数千个客户端同时读写。Kafka应用场景活动跟踪Kafka 可以用来跟踪用户行为比如你经常回去App购物你打开App的那一刻你的登陆信息登陆次数都会作为消息传输到 Kafka 当你浏览购物的时候你的浏览信息你的搜索指数你的购物爱好都会作为一个个消息传递给 Kafka 这样就可以生成报告可以做智能推荐购买喜好等传递消息Kafka 另外一个基本用途是传递消息应用程序向用户发送通知就是通过传递消息来实现的这些应用组件可以生成消息而不需要关心消息的格式也不需要关心消息是如何发送的度量指标Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据生产各种操作的集中反馈比如报警和报告日志记录Kafka 的基本概念来源于提交日志比如可以把数据库的更新发送到 Kafka 上用来记录数据库的更新时间通过Kafka以统一接口服务的方式开放给各种consumer例如hadoop、Hbase、Solr等流式处理流式处理是有一个能够提供多种应用程序的领域限流削峰Kafka 多用于互联网领域某一时刻请求特别多的情况下可以把请求写入Kafka 中避免直接请求后端程序导致服务崩溃。以上介绍参考Kafka官方文档。Kafka核心APIKafka有4个核心API应用程序使用Producer API发布消息到1个或多个Topics中应用程序使用ConsumerAPI来订阅1个或多个Topics并处理产生的消息应用程序使用Streams API充当一个流处理器从1个或多个Topics消费输入流并产生一个输出流到1个或多个Topics有效地将输入流转换到输出流Connector API允许构建或运行可重复使用的生产者或消费者将Topic链接到现有的应用程序或数据系统。Kafka为何如此之快Kafka 实现了零拷贝原理来快速移动数据避免了内核之间的切换。Kafka 可以将数据记录分批发送从生产者到文件系统Kafka 主题日志到消费者可以端到端的查看这些批次的数据。批处理能够进行更有效的数据压缩并减少 I/O 延迟Kafka 采取顺序写入磁盘的方式避免了随机磁盘寻址的浪费。总结一下其实就是四个要点顺序读写零拷贝消息压缩分批发送。案例项目创建Dependencies构建工具为MavenMaven的依赖如下代码语言javascriptAI代码解释dependency groupIdorg.apache.kafka/groupId artifactIdkafka_2.12/artifactId version1.0.0/version scopeprovided/scope /dependency dependency groupIdorg.apache.kafka/groupId artifactIdkafka-clients/artifactId version1.0.0/version /dependency dependency groupIdorg.apache.kafka/groupId artifactIdkafka-streams/artifactId version1.0.0/version /dependencyKafka Producer代码语言javascriptAI代码解释package cn.com.codingce.module; import java.util.Properties; import java.util.Random; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; public class Producer { // 定义主题 public static String topic codingce_test; public static void main(String[] args) throws InterruptedException { Properties p new Properties(); // bootstrap.servers: kafka的地址, 多个地址用逗号分割 p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.31.150:9092); // acks:消息的确认机制默认值是0. acks0: 如果设置为0生产者不会等待kafka的响应; acks1: 这个配置意味着kafka会把这条消息写到本地日志文件中但是不会等待集群中其他机器的成功响应 // acksall: 这个配置意味着leader会等待所有的follower同步完成. 这个确保消息不会丢失, 除非kafka集群中所有机器挂掉. 这是最强的可用性保证. p.put(acks, all); // retries: 配置为大于0的值的话, 客户端会在消息发送失败时重新发送. p.put(retries, 0); // batch.size: 当多条消息需要发送到同一个分区时生产者会尝试合并网络请求. 这会提高client和生产者的效率. p.put(batch.size, 16384); // key.serializer: 键序列化默认org.apache.kafka.common.serialization.StringDeserializer. p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); // value.deserializer:值序列化默认org.apache.kafka.common.serialization.StringDeserializer. p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); KafkaProducerString, String kafkaProducer new KafkaProducer(p); try { do { String msg 后端码匠, new Random().nextInt(100); ProducerRecordString, String record new ProducerRecord(topic, msg); kafkaProducer.send(record); System.out.println(消息发送成功: msg ); Thread.sleep(1000L); } while (true); } finally { kafkaProducer.close(); } } }output代码语言javascriptAI代码解释消息发送成功: 后端码匠, 97 消息发送成功: 后端码匠, 35 消息发送成功: 后端码匠, 81 消息发送成功: 后端码匠, 46 消息发送成功: 后端码匠, 62 消息发送成功: 后端码匠, 53 消息发送成功: 后端码匠, 42 消息发送成功: 后端码匠, 56 消息发送成功: 后端码匠, 99 消息发送成功: 后端码匠, 46 消息发送成功: 后端码匠, 49 消息发送成功: 后端码匠, 35 消息发送成功: 后端码匠, 17 消息发送成功: 后端码匠, 78 消息发送成功: 后端码匠, 66 消息发送成功: 后端码匠, 4 消息发送成功: 后端码匠, 9 消息发送成功: 后端码匠, 69 消息发送成功: 后端码匠, 52 消息发送成功: 后端码匠, 2 消息发送成功: 后端码匠, 8 消息发送成功: 后端码匠, 86 消息发送成功: 后端码匠, 12 消息发送成功: 后端码匠, 67 消息发送成功: 后端码匠, 91 消息发送成功: 后端码匠, 8 消息发送成功: 后端码匠, 56 消息发送成功: 后端码匠, 89 消息发送成功: 后端码匠, 37 消息发送成功: 后端码匠, 39 消息发送成功: 后端码匠, 71 Kafka Consumer代码语言javascriptAI代码解释package cn.com.codingce.module; 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; import org.apache.kafka.common.serialization.StringDeserializer; import java.util.Collections; import java.util.Properties; public class Consumer { private static final String GROUPID codingce_consumer_a; public static void main(String[] args) { Properties p new Properties(); // bootstrap.servers: kafka的地址, 多个地址用逗号分割 p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.31.150:9092); // 消费者所属的分组id, 组名 不同组名可以重复消费.例如你先使用了组名A消费了Kafka的1000条数据, 但是你还想再次进行消费这1000条数据, // 并且不想重新去产生, 那么这里你只需要更改组名就可以重复消费了. p.put(ConsumerConfig.GROUP_ID_CONFIG, GROUPID); // 是否自动提交, 默认为true. p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); // 从poll(拉)的回话处理时长 p.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000); // 超时时间 p.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000); // 一次最大拉取的条数 p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000); // 消费规则, 默认earliest p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, earliest); // key.serializer: 键序列化, 默认org.apache.kafka.common.serialization.StringDeserializer. p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // value.deserializer:值序列化, 默认org.apache.kafka.common.serialization.StringDeserializer. p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); KafkaConsumerString, String kafkaConsumer new KafkaConsumer(p); // 订阅消息 kafkaConsumer.subscribe(Collections.singletonList(Producer.topic)); do { // 订阅之后, 再从kafka中拉取数据 ConsumerRecordsString, String records kafkaConsumer.poll(100); for (ConsumerRecordString, String record : records) { System.out.printf(-----topic:%s, offset:%d, 消息:%s-----\n, record.topic(), record.offset(), record.value()); } } while (true); } }output代码语言javascriptAI代码解释-----topic:codingce_test, offset:289, 消息:后端码匠, 97----- -----topic:codingce_test, offset:290, 消息:后端码匠, 35----- -----topic:codingce_test, offset:291, 消息:后端码匠, 81----- -----topic:codingce_test, offset:292, 消息:后端码匠, 46----- -----topic:codingce_test, offset:293, 消息:后端码匠, 62----- -----topic:codingce_test, offset:294, 消息:后端码匠, 53----- -----topic:codingce_test, offset:295, 消息:后端码匠, 42----- -----topic:codingce_test, offset:296, 消息:后端码匠, 56----- -----topic:codingce_test, offset:297, 消息:后端码匠, 99----- -----topic:codingce_test, offset:298, 消息:后端码匠, 46----- -----topic:codingce_test, offset:299, 消息:后端码匠, 49----- -----topic:codingce_test, offset:300, 消息:后端码匠, 35----- -----topic:codingce_test, offset:301, 消息:后端码匠, 17----- -----topic:codingce_test, offset:302, 消息:后端码匠, 78----- -----topic:codingce_test, offset:303, 消息:后端码匠, 66----- -----topic:codingce_test, offset:304, 消息:后端码匠, 4----- -----topic:codingce_test, offset:305, 消息:后端码匠, 9----- -----topic:codingce_test, offset:306, 消息:后端码匠, 69----- -----topic:codingce_test, offset:307, 消息:后端码匠, 52----- -----topic:codingce_test, offset:308, 消息:后端码匠, 2----- -----topic:codingce_test, offset:309, 消息:后端码匠, 8----- -----topic:codingce_test, offset:310, 消息:后端码匠, 86----- -----topic:codingce_test, offset:311, 消息:后端码匠, 12----- -----topic:codingce_test, offset:312, 消息:后端码匠, 67----- -----topic:codingce_test, offset:313, 消息:后端码匠, 91----- -----topic:codingce_test, offset:314, 消息:后端码匠, 8----- -----topic:codingce_test, offset:315, 消息:后端码匠, 56----- -----topic:codingce_test, offset:316, 消息:后端码匠, 89----- -----topic:codingce_test, offset:317, 消息:后端码匠, 37----- -----topic:codingce_test, offset:318, 消息:后端码匠, 39----- -----topic:codingce_test, offset:319, 消息:后端码匠, 71-----