Java客户端访问Kafka

Posted 2022-02-03 java_wxid

我是🌟廖志伟🌟，一名🌕Java开发工程师🌕、📝Java领域优质创作者📝、🎉CSDN博客专家🎉、🌹幕后大佬社区创始人🌹。拥有多年一线研发经验，研究过各种常见框架及中间件的底层源码，对于大型分布式、微服务、三高架构（高性能、高并发、高可用）有过实践架构经验。

🍊博主：java_wxid
🍊博主：Java廖志伟
🍊社区：幕后大佬

---

文章目录

• • Java客户端访问Kafka
  • • 引入maven依赖
    • 消息发送端代码
    • 消息接收端代码

本文内容：

Java客户端访问Kafka

---

引入maven依赖

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>2.4.1</version>
</dependency>

消息发送端代码

package com.sky.kafka.kafkaDemo;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class MsgProducer 

    private final static String TOPIC_NAME = "my-replicated-topic";

    public static void main(String[] args) throws InterruptedException, ExecutionException 
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
         /*
         发出消息持久化机制参数
        （1）acks=0： 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。
        （2）acks=1： 至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一
             条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
        （3）acks=-1或all： 需要等待 min.insync.replicas(默认为1，推荐配置大于等于2) 这个参数配置的副本个数都成功写入日志，这种策略会保证
            只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。
         */
        /*props.put(ProducerConfig.ACKS_CONFIG, "1");
         *//*
        发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造成消息重复发送，比如网络抖动，所以需要在
        接收者那边做好消息接收的幂等性处理
        *//*
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        //重试间隔设置
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
        //设置发送消息的本地缓冲区，如果设置了该缓冲区，消息会先发送到本地缓冲区，可以提高消息发送性能，默认值是33554432，即32MB
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        *//*
        kafka本地线程会从缓冲区取数据，批量发送到broker，
        设置批量发送消息的大小，默认值是16384，即16kb，就是说一个batch满了16kb就发送出去
        *//*
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        *//*
        默认值是0，意思就是消息必须立即被发送，但这样会影响性能
        一般设置10毫秒左右，就是说这个消息发送完后会进入本地的一个batch，如果10毫秒内，这个batch满了16kb就会随batch一起被发送出去
        如果10毫秒内，batch没满，那么也必须把消息发送出去，不能让消息的发送延迟时间太长
        *//*
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);*/
        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        int msgNum = 5;
        final CountDownLatch countDownLatch = new CountDownLatch(msgNum);
        for (int i = 1; i <= msgNum; i++) 
            Order order = new Order(i, 100 + i, 1, 1000.00);
            //指定发送分区
            /*ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , 0, order.getOrderId().toString(), JSON.toJSONString(order));*/
            //未指定发送分区，具体发送的分区计算公式：hash(key)%partitionNum
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , order.getOrderId().toString(), JSON.toJSONString(order));

            //等待消息发送成功的同步阻塞方法
            /*RecordMetadata metadata = producer.send(producerRecord).get();
            System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                    + metadata.partition() + "|offset-" + metadata.offset());*/

            //异步回调方式发送消息
            producer.send(producerRecord, new Callback() 
                public void onCompletion(RecordMetadata metadata, Exception exception) 
                    if (exception != null) 
                        System.err.println("发送消息失败：" + exception.getStackTrace());

                    
                    if (metadata != null) 
                        System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                                + metadata.partition() + "|offset-" + metadata.offset());
                    
                    countDownLatch.countDown();
                
            );

            //送积分 TODO

        

        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.close();
    

消息接收端代码

package com.sky.kafka.kafkaDemo;

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.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class MsgConsumer 
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) 
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094");
        // 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        // 是否自动提交offset，默认就是true
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 自动提交offset的间隔时间
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        /*
        当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费
        latest(默认) ：只消费自己启动之后发送到主题的消息
        earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
        */
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      /*
      consumer给broker发送心跳的间隔时间，broker接收到心跳如果此时有rebalance发生会通过心跳响应将
      rebalance方案下发给consumer，这个时间可以稍微短一点
      */
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        /*
        服务端broker多久感知不到一个consumer心跳就认为他故障了，会将其踢出消费组，
        对应的Partition也会被重新分配给其他consumer，默认是10秒
        */
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
        //一次poll最大拉取消息的条数，如果消费者处理速度很快，可以设置大点，如果处理速度一般，可以设置小点
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        /*
        如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太弱，
        会将其踢出消费组，将分区分配给别的consumer消费
        */
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        // 消费指定分区
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

        //消息回溯消费
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));*/

        //指定offset消费
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);*/

        //从指定时间点开始消费
        /*List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
        //从1小时前开始消费
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
        Map<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : topicPartitions) 
            map.put(new TopicPartition(topicName, par.partition()), fetchDataTime);
        
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) 
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() + "|offset-" + offset);
            System.out.println();
            //根据消费里的timestamp确定offset
            if (value != null) 
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            
        */

        while (true) 
            /*
             * poll() API 是拉取消息的长轮询
             */
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) 
                System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
                        record.offset(), record.key(), record.value());
            

            /*if (records.count() > 0) 
                // 手动同步提交offset，当前线程会阻塞直到offset提交成功
                // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
                consumer.commitSync();

                // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
                consumer.commitAsync(new OffsetCommitCallback() 
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) 
                        if (exception != null) 
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + exception.getStackTrace());
                        
                    
                );

            */
        
    

---

总结

以上就是今天要讲的内容，还希望各位读者大大能够在评论区积极参与讨论，给文章提出一些宝贵的意见或者建议📝，合理的内容，我会采纳更新博文，重新分享给大家。

🙏四连 关注🔎点赞👍收藏⭐️留言📝

感谢大家的支持，用心写博文分享给大家，你的支持（🔎点赞👍收藏⭐️留言📝）是对我创作的最大帮助。
🍊微信公众号：南北踏尘
🍊主页地址：java_wxid
🍊社区地址：幕后大佬

给读者大大的话

我本身是一个很普通的程序员，放在人堆里，除了与生俱来的🌹盛世美颜🌹、所剩不多的发量，就剩下180的大高个了。就是我这样的一个人，默默坚持写博文也有好多年了，有句老话说的好，🌕牛逼之前都是傻逼式的坚持🌕。希望自己可以通过大量的作品，时间的积累，个人魅力、运气和时机，可以打造属于自己的🌟技术影响力🌟。同时也希望自己可以成为一个🎄懂技术🎄，🎄懂业务🎄，🎄懂管理🎄的综合型人才，作为项目架构路线的总设计师，掌控全局的🌕团队大脑🌕，技术团队中的🍊绝对核心🍊是我未来几年不断前进的目标。

---

提示：以下都是资源分享，求个一键三连。

面试资料

福利大放送，🎉欢迎关注🔎点赞👍收藏⭐️留言📝，拜托了🙏，这对我真的很重要。
点击：面试资料
提取码：2021

200套PPT模板

福利大放送，🎉欢迎关注🔎点赞👍收藏⭐️留言📝，拜托了🙏，这对我真的很重要。
点击：200套PPT模板
提取码：2021

提问的智慧

福利大放送，🎉欢迎关注🔎点赞👍收藏⭐️留言📝，拜托了🙏，这对我真的很重要。
点击：提问的智慧
提取码：2021

Java开发学习路线

名称	链接
JavaSE	点击： JavaSE
mysql专栏	点击： MySQL专栏
JDBC专栏	点击： JDBC专栏
MyBatis专栏	点击： MyBatis专栏
Web专栏	点击： Web专栏
Spring专栏	点击： Spring专栏
SpringMVC专栏	点击： SpringMVC专栏
SpringBoot专栏	点击： SpringBoot专栏
SpringCould专栏	点击： SpringCould专栏
Redis专栏	点击： Redis专栏
Linux专栏	点击： Linux专栏
Maven3专栏	点击： Maven3专栏
Spring Security5专栏	点击： Spring Security5专栏
更多专栏	更多专栏，请到 java_wxid主页 查看

P5学习路线图
P6学习路线图
P7学习路线图
P8学习路线图

以上四张图详细介绍了作为Java开发工作者所需要具备的知识技能，同学们学废了嘛，有想法系统学习的同学可以私聊我，🎉欢迎关注🔎点赞👍收藏⭐️留言📝。
🍊博主：java_wxid
🍊博主：Java廖志伟
🍊社区：幕后大佬