Kafka--05---java客户端代码实现

Posted 2022-01-15 高高for 循环

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Kafka--05---java客户端代码实现相关的知识，希望对你有一定的参考价值。

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

文章目录

java客户端-⽣产者的实现
java客户端-消费者的实现

java客户端-⽣产者的实现

依赖

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

1. ⽣产者的基本实现

设置参数-----Properties
创建⽣产消息的客户端 ----KafkaProducer
创建消息-----key：作⽤是决定了往哪个分区上发，value：具体要发送的消息内容
发送消息,得到消息发送的元数据并输出

package com.cy.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

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

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        //1.设置参数
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094");
        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


        //2.创建⽣产消息的客户端，传⼊参数
        Producer<String, String> producer = new KafkaProducer<String, String>(props);


        //3.创建消息
        //key：作⽤是决定了往哪个分区上发，value：具体要发送的消息内容
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC_NAME, 1,"mykeyvalue", "hellokafka");


        //4.发送消息,得到消息发送的元数据并输出
        RecordMetadata metadata = producer.send(producerRecord).get();

        System.out.println("同步⽅式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());

发送消息到指定分区上

newProducerRecord<String, String>(TOPIC_NAME, 0,
order.getOrderId().toString(), 
JSON.toJSONString(order));

ProducerRecord

topic, partition. key, value

未指定分区，则会通过业务key的hash运算，算出消息往哪个分区上发

new ProducerRecord<String, String>(TOPIC_NAME
, order.getOrderId().toString(), JSON.toJSONString(order));

2.⽣产者的----同步发送消息

⽣产者同步发消息，在收到kafka的ack告知发送成功之前⼀直处于阻塞状态

如果⽣产者发送消息没有收到ack，⽣产者会阻塞，阻塞到3s的时间，如果还没有收到消息，会进⾏重试。重试的次数3次。

3. ⽣产者的----异步发送消息

⽣产者发消息，发送完后不⽤等待broker给回复，直接执⾏下⾯的业务逻辑

broker在收到消息后异步调⽤⽣产者提供的callback回调⽅法。

package com.cy.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

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

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        //1.设置参数
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094");
        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


        //2.创建⽣产消息的客户端，传⼊参数
        Producer<String, String> producer = new KafkaProducer<String, String>(props);


        //3.创建消息
        //key：作⽤是决定了往哪个分区上发，value：具体要发送的消息内容
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(TOPIC_NAME, 1,"mykeyvalue", "hellokafka");


        //4.异步发送消息
        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());
                
            
        );

4.⽣产者中相关配置-----ProducerConfig

props.put(ProducerConfig.ACKS_CONFIG, "1");
/*
发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造
成消息重复发送，⽐如⽹络抖动，所以需要在
接收者那边做好消息接收的幂等性处理
*/
props.put(ProducerConfig.RETRIES_CONFIG, 3);
//重试间隔设置
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
//kafka默认会创建⼀个消息缓冲区，⽤来存放要发送的消息，缓冲区是32m
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
//kafka本地线程会去缓冲区中⼀次拉16k的数据，发送到broker
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
//如果线程拉不到16k的数据，间隔10ms也会将已拉到的数据发到broker
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);

ack的配置

在同步发送的前提下，⽣产者在获得集群返回的ack之前会⼀直阻塞。那么集群什么时候返回
ack呢？此时ack有3个配置：

ack = 0 kafka-cluster不需要任何的broker收到消息，就⽴即返回ack给⽣产者，最容易
丢消息的，效率是最⾼的
ack=1（默认）：多副本之间的leader已经收到消息，并把消息写⼊到本地的log中，才
会返回ack给⽣产者，性能和安全性是最均衡的
ack=-1/all。⾥⾯有默认的配置min.insync.replicas=2(默认为1，推荐配置⼤于等于2)，
此时就需要leader和⼀个follower同步完后，才会返回ack给⽣产者（此时集群中有2个
broker已完成数据的接收），这种⽅式最安全，但性能最差。

下⾯是关于ack和重试（如果没有收到ack，就开启重试）的配置

props.put(ProducerConfig.ACKS_CONFIG, "1");
/*
发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造
成消息重复发送，⽐如⽹络抖动，所以需要在
接收者那边做好消息接收的幂等性处理
*/
props.put(ProducerConfig.RETRIES_CONFIG, 3);
//重试间隔设置
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);

5. 关于消息发送的缓冲区

kafka默认会创建⼀个消息缓冲区，⽤来存放要发送的消息，缓冲区是32m
kafka本地线程会去缓冲区中⼀次拉16k的数据，发送到broker
如果线程拉不到16k的数据，间隔10ms也会将已拉到的数据发到broker

//kafka默认会创建⼀个消息缓冲区，⽤来存放要发送的消息，缓冲区是32m
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
//kafka本地线程会去缓冲区中⼀次拉16k的数据，发送到broker
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
//如果线程拉不到16k的数据，间隔10ms也会将已拉到的数据发到broker
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);

java客户端-消费者的实现

1.消费者的基本实现

设置参数-----Properties
创建⼀个消费者的客户端------KafkaConsumer
消费者订阅主题列表-------- consumer.subscribe()
拉取消息的⻓轮询---------- consumer.poll()
打印消息

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 MySimpleConsumer 
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) 
        //设置参数-----Properties
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094");
        // 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //1.创建⼀个消费者的客户端
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        //2. 消费者订阅主题列表
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) 

            // 3.poll() API 是拉取消息的⻓轮询
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) 

                //4.打印消息
                System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
                        record.offset(), record.key(), record.value());

ConsumerRecord

record.partition()
record.offset()
record.key(),
record.value()

2.⾃动提交和⼿动提交offset

1）提交的内容

消费者⽆论是⾃动提交还是⼿动提交，都需要把所属的消费组+消费的某个主题+消费的某个分区及消费的偏移量，这样的信息提交到集群的_consumer_offsets主题⾥⾯。

2）提交的位置

⾃动提交

消费者poll消息下来以后就会⾃动提交offset

// 是否⾃动提交offset，默认就是true
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// ⾃动提交offset的间隔时间
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

注意：⾃动提交可能会丢消息。因为消费者在消费前提交offset，有可能提交完后还没消费时消费者挂了。

⼿动提交

需要把⾃动提交的配置改成false

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

⼿动提交⼜分成了两种：

1）⼿动同步提交

在消费完消息后调⽤同步提交的⽅法，当集群返回ack前⼀直阻塞，返回ack后表示提交成功，执⾏之后的逻辑

        // poll() API 是拉取消息的⻓轮询
        while (true) 
            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();//=======阻塞=== 提交成功

2）⼿动异步提交

在消息消费完后提交，不需要等到集群ack，直接执⾏之后的逻辑，可以设置⼀个回调⽅
法，供集群调⽤

        // poll() API 是拉取消息的⻓轮询
        while (true) 
            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.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());
                        

                    
                );

3.⻓轮询poll消息

默认情况下，消费者⼀次会poll500条消息。

以上是关于Kafka--05---java客户端代码实现的主要内容，如果未能解决你的问题，请参考以下文章 
 [How to] 使用HBase协处理器---Endpoint客户端代码的实现
 PBFT代码实现
 Electron客户端实现本地录制功能详解附代码
 C#实现代码生成器客户端
 PHP如何实现手机定位，求代码！
 php/js获取客户端mac地址的实现代码