Kafka 使用java实现,快速入门

这篇具有很好参考价值的文章主要介绍了Kafka 使用java实现,快速入门。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、kafka的生产者和消费者

1. 生产者发送消息的流程

java kafka的使用,kafka,spring boot,java

 2. 消费者接收消息的流程

java kafka的使用,kafka,spring boot,java

 二、 java 代码实现

1. 添加依赖:

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.12</artifactId>
        </dependency>

2. 实现生产者

public class NormalProducer {

    public static void main(String[] args) {
        Properties properties = new Properties();
        //	1.配置生产者启动的关键属性参数

        //	1.1	BOOTSTRAP_SERVERS_CONFIG:连接kafka集群的服务列表,如果有多个,使用"逗号"进行分隔
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.31.101:9092");
        //	1.2	CLIENT_ID_CONFIG:这个属性的目的是标记kafkaclient的ID
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "quickstart-producer");
        //	1.3 KEY_SERIALIZER_CLASS_CONFIG VALUE_SERIALIZER_CLASS_CONFIG
        //	Q: 对 kafka的 key 和 value 做序列化,为什么需要序列化?
        //	A: 因为KAFKA Broker 在接收消息的时候,必须要以二进制的方式接收,所以必须要对KEY和VALUE进行序列化
        //	字符串序列化类:org.apache.kafka.common.serialization.StringSerializer
        //	KEY: 是kafka用于做消息投递计算具体投递到对应的主题的哪一个partition而需要的
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //	VALUE: 实际发送消息的内容
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //	2.创建kafka生产者对象 传递properties属性参数集合
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        for(int i = 0; i <10; i ++) {
            //	3.构造消息内容
            User user = new User("00" + i, "张三");
            ProducerRecord<String, String> record =
                    //	arg1:topic , arg2:实际的消息体内容,quick_start 是 topic 名称
                    new ProducerRecord<String, String>("quick_start",
                            JSON.toJSONString(user));

            //	4.发送消息
            producer.send(record);
        }


        //	5.关闭生产者
        producer.close();

    }
}

其中的 User 对象为:

public class User {

	private String id;
	
	private String name;

	public User() {
	}

	public User(String id, String name) {
		this.id = id;
		this.name = name;
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}
}

3. 实现消费者

public class NormalConsumer {

    public static void main(String[] args) {

        //	1. 配置属性参数
        Properties properties = new Properties();

        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.31.101:9092");

        //	org.apache.kafka.common.serialization.StringDeserializer
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //	非常重要的属性配置:与我们消费者订阅组有关系
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "quickstart-group");
        //	常规属性:会话连接超时时间
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
        //	消费者提交offset: 自动提交 & 手工提交,默认是自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

        //	2. 创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        //	3. 订阅你感兴趣的主题:quick_start
        consumer.subscribe(Collections.singletonList("quick_start"));

        System.err.println("quickstart consumer started...");

        try {
            //	4.采用拉取消息的方式消费数据
            while(true) {
                //	等待多久拉取一次消息
                //	拉取TOPIC_QUICKSTART主题里面所有的消息
                //	topic 和 partition是 一对多的关系,一个topic可以有多个partition
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                //	因为消息是在partition中存储的,所以需要遍历partition集合
                for(TopicPartition topicPartition : records.partitions()) {
                    //	通过TopicPartition获取指定的消息集合,获取到的就是当前topicPartition下面所有的消息
                    List<ConsumerRecord<String, String>> partitionRecords = records.records(topicPartition);
                    //	获取TopicPartition对应的主题名称
                    String topic = topicPartition.topic();
                    //	获取当前topicPartition下的消息条数
                    int size = partitionRecords.size();

                    System.err.println(String.format("--- 获取topic: %s, 分区位置:%s, 消息总数: %s",
                            topic,
                            topicPartition.partition(),
                            size));

                    for(int i = 0; i < size; i++) {
                        ConsumerRecord<String, String> consumerRecord = partitionRecords.get(i);
                        //	实际的数据内容
                        String value = consumerRecord.value();
                        //	当前获取的消息偏移量
                        long offset = consumerRecord.offset();
                        //	ISR : High Watermark, 如果要提交的话,比如提交当前消息的offset+1
                        //	表示下一次从什么位置(offset)拉取消息
                        long commitOffser = offset + 1;
                        System.err.println(String.format("获取实际消息 value:%s, 消息offset: %s, 提交offset: %s",
                                value, offset, commitOffser));
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }
}

4. 测试结果

生产者发送的消息在消费者端可以正常接收:

java kafka的使用,kafka,spring boot,java文章来源地址https://www.toymoban.com/news/detail-606486.html

到了这里,关于Kafka 使用java实现,快速入门的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【Spring Boot】集成Kafka实现消息发送和订阅

    最近忙着搞低代码开发,好久没新建spring项目了,结果今天心血来潮准备建个springboot项目 注意Type选Maven,java选8,其他默认 点下一步后完成就新建了一个spring boot项目,配置下Maven环境,主要是settings.xml文件,里面要包含阿里云仓库,不然可能依赖下载不下来 在maven配置没问

    2024年02月09日
    浏览(49)
  • Spring Boot中使用Kafka时遇到“构建Kafka消费者失败“的问题

    在使用Spring Boot开发应用程序时,集成Apache Kafka作为消息队列是一种常见的做法。然而,有时候在配置和使用Kafka时可能会遇到一些问题。本文将探讨在Spring Boot应用程序中使用Kafka时可能遇到的\\\"构建Kafka消费者失败\\\"错误,并提供解决方案。 错误描述: 当尝试构建Kafka消费者时

    2024年01月17日
    浏览(48)
  • Spring Boot整合Kafka+SSE实现实时数据展示

    2024年3月10日 不使用Rabbitmq或者Rocketmq是因为Kafka是Hadoop集群下的组成部分,对于大数据的相关开发适应性好,且当前业务场景下不需要使用死信队列,不过要注意Kafka对于更新时间慢的数据拉取也较慢,因此对与实时性要求高可以选择其他MQ。 使用消息队列是因为该中间件具有

    2024年04月24日
    浏览(40)
  • 使用Spring Boot集成中间件:Kafka的高级使用案例讲解

    在实际应用中,Kafka作为一种强大的分布式消息系统,广泛应用于实时数据处理和消息传递。本文将通过一个全面的使用案例,详细介绍如何使用Spring Boot集成Kafka,并展示其在实际场景中的应用。 在开始之前,我们需要确保已经完成以下准备工作: 安装并启动Kafka集群 创建

    2024年02月01日
    浏览(54)
  • kafka--技术文档--spring-boot集成基础简单使用

            查阅了很多资料了解到,使用了spring-boot中整合的kafka的使用是被封装好的。也就是说这些使用其实和在linux中的使用kafka代码的使用其实没有太大关系。但是逻辑是一样的。这点要注意! 核心配置为: 如果在下面规定了spring-boot的版本那么就不需要再使用版本号,如

    2024年02月11日
    浏览(48)
  • 入门Spring Boot:快速构建Java应用的利器

    Spring Boot是由Pivotal团队开发的开源框架,它基于Spring框架,旨在简化Java应用程序的开发过程。它提供了一种约定大于配置的方式,通过自动配置和起步依赖(Starter Dependencies)来消除繁琐的配置,从而使开发者能够更快地构建独立、可执行的、生产级的Spring应用。 与传统的

    2024年02月07日
    浏览(47)
  • Kafka增加安全验证安全认证,SASL认证,并通过spring boot-Java客户端连接配置

    公司Kafka一直没做安全验证,由于是诱捕程序故需要面向外网连接,需要增加Kafka连接验证,保证Kafka不被非法连接,故开始研究Kafka安全验证 使用Kafka版本为2.4.0版本,主要参考官方文档 官网对2.4版本安全验证介绍以及使用方式地址: https://kafka.apache.org/24/documentation.html#secu

    2024年02月01日
    浏览(66)
  • 【Spring Boot+Kafka+Mysql+HBase】实现分布式优惠券后台应用系统(附源码)

    需要全部代码请点赞关注收藏后评论区留言私信~~~ 分布式优惠券后台应用系统服务于两类用户群体,一类是商户,商户可以根据自己的实际情况即进行优惠券投放,另一类是平台消费用户,用户可以去领取商户发放的优惠券 分布式优惠券后台应用系统采用SpringBoot作为主体开

    2024年02月10日
    浏览(43)
  • Spring Boot与Apache Kafka实现高吞吐量消息处理:解决大规模数据处理问题

    现代数据量越来越庞大对数据处理的效率提出了更高的要求。Apache Kafka是目前流行的分布式消息队列之一。Spring Boot是现代Java应用程序快速开发的首选框架。综合使用Spring Boot和Apache Kafka可以实现高吞吐量消息处理。 Apache Kafka采用分布式发布-订阅模式具有高度的可扩展性和可

    2024年02月05日
    浏览(52)
  • 在Spring Boot微服务集成Kafka客户端(spring-kafka)操作Kafka

    记录 :457 场景 :在Spring Boot微服务集成Kafka客户端spring-kafka-2.8.2操作Kafka。使用Spring封装的KafkaTemplate操作Kafka生产者Producer。使用Spring封装的@KafkaListener操作Kafka的消费者Consumer。 版本 :JDK 1.8,Spring Boot 2.6.3,kafka_2.12-2.8.0,spring-kafka-2.8.2。 Kafka安装 :https://blog.csdn.net/zhangbeizhen1

    2024年02月09日
    浏览(61)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包