热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

Java实现Kafka的生产者、消费者

Java实现Kafka的生产者、消费者,Go语言社区,Golang程序员人脉社

一、生产者

import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import java.util.Properties; public class SimpleKafkaProducer { private static KafkaProducer producer; private final static String TOPIC = "adienTest2"; public SimpleKafkaProducer(){ Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", 33554432); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //设置分区类,根据key进行数据分区 producer = new KafkaProducer(props); } public void produce(){ for (int i = 30;i<40;i++){ String key = String.valueOf(i); String data = "hello kafka message:"+key; producer.send(new ProducerRecord(TOPIC,key,data)); System.out.println(data); } producer.close(); } public static void main(String[] args) { new SimpleKafkaProducer().produce(); } }

二、消费者

import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.log4j.Logger; import java.util.Arrays; import java.util.Properties; public class SimpleKafkaConsumer { private static KafkaConsumer consumer; private final static String TOPIC = "adienTest2" public SimpleKafkaConsumer(){ Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); //每个消费者分配独立的组号 props.put("group.id", "test2"); //如果value合法,则自动提交偏移量 props.put("enable.auto.commit", "true"); //设置多久一次更新被消费消息的偏移量 props.put("auto.commit.interval.ms", "1000"); //设置会话响应的时间,超过这个时间kafka可以选择放弃消费或者消费下一条消息 props.put("session.timeout.ms", "30000"); //自动重置offset props.put("auto.offset.reset","earliest"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); cOnsumer= new KafkaConsumer(props); } public void consume(){ consumer.subscribe(Arrays.asList(TOPIC)); while (true) { ConsumerRecords records = consumer.poll(100); for (ConsumerRecord record : records){ System.out.printf("offset = %d, key = %s, value = %s",record.offset(), record.key(), record.value()); System.out.println(); } } } public static void main(String[] args) { new SimpleKafkaConsumer().consume(); } }

三、运行
1、该生产者、消费者是运行本地搭建的kafka,所以先用命令行启动zookeeper,再启动kafka服务,还要创建本例中使用的topic:adienTest2
2、运行生产者代码,运行结果如下:
这里写图片描述
3、运行消费者代码,运行结果如下:
这里写图片描述


推荐阅读
  • Explore a common issue encountered when implementing an OAuth 1.0a API, specifically the inability to encode null objects and how to resolve it. ... [详细]
  • 优化ListView性能
    本文深入探讨了如何通过多种技术手段优化ListView的性能,包括视图复用、ViewHolder模式、分批加载数据、图片优化及内存管理等。这些方法能够显著提升应用的响应速度和用户体验。 ... [详细]
  • 优化局域网SSH连接延迟问题的解决方案
    本文介绍了解决局域网内SSH连接到服务器时出现长时间等待问题的方法。通过调整配置和优化网络设置,可以显著缩短SSH连接的时间。 ... [详细]
  • HBase运维工具全解析
    本文深入探讨了HBase常用的运维工具,详细介绍了每种工具的功能、使用场景及操作示例。对于HBase的开发人员和运维工程师来说,这些工具是日常管理和故障排查的重要手段。 ... [详细]
  • Explore how Matterverse is redefining the metaverse experience, creating immersive and meaningful virtual environments that foster genuine connections and economic opportunities. ... [详细]
  • 本文详细介绍了如何在Linux系统上安装和配置Smokeping,以实现对网络链路质量的实时监控。通过详细的步骤和必要的依赖包安装,确保用户能够顺利完成部署并优化其网络性能监控。 ... [详细]
  • 本文介绍如何利用动态规划算法解决经典的0-1背包问题。通过具体实例和代码实现,详细解释了在给定容量的背包中选择若干物品以最大化总价值的过程。 ... [详细]
  • 深入理解Java中的volatile、内存屏障与CPU指令
    本文详细探讨了Java中volatile关键字的作用机制,以及其与内存屏障和CPU指令之间的关系。通过具体示例和专业解析,帮助读者更好地理解多线程编程中的同步问题。 ... [详细]
  • MQTT技术周报:硬件连接与协议解析
    本周开发笔记重点介绍了在新项目中使用MQTT协议进行硬件连接的技术细节,涵盖其特性、原理及实现步骤。 ... [详细]
  • 本文详细介绍了Java中org.w3c.dom.Text类的splitText()方法,通过多个代码示例展示了其实际应用。该方法用于将文本节点在指定位置拆分为两个节点,并保持在文档树中。 ... [详细]
  • 本文详细介绍了 Apache Jena 库中的 Txn.executeWrite 方法,通过多个实际代码示例展示了其在不同场景下的应用,帮助开发者更好地理解和使用该方法。 ... [详细]
  • 本文详细介绍如何使用Samba软件配置CIFS文件共享服务,涵盖安装、配置、权限管理及多用户挂载等关键步骤。通过具体示例和命令行操作,帮助读者快速搭建并优化Samba服务器。 ... [详细]
  • MySQL 数据库迁移指南:从本地到远程及磁盘间迁移
    本文详细介绍了如何在不同场景下进行 MySQL 数据库的迁移,包括从一个硬盘迁移到另一个硬盘、从一台计算机迁移到另一台计算机,以及解决迁移过程中可能遇到的问题。 ... [详细]
  • 本文介绍如何在Java项目中使用Log4j库进行日志记录。我们将详细说明Log4j库的引入、配置及简单应用,帮助开发者快速上手。 ... [详细]
  • 本文探讨了高质量C/C++编程的最佳实践,并详细分析了常见的内存错误及其解决方案。通过深入理解内存管理和故障排除技巧,开发者可以编写更健壮的程序。 ... [详细]
author-avatar
横刀2502934567
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有