【Kafka】基本使用：Java操作Kafka（kafka-clients）

╰半橙微兮° 2022-12-21 04:42 213阅读 0赞

引入依赖：

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

## 1.Producer ##

### 1.1 同步调用 ###

public class MyKafKaProducer extends Thread{
        
    
        KafkaProducer<Integer, String> producer;
        String topic; // 主题
    
        public MyKafKaProducer(String topic) {
        
            // 构建连接配置
            Properties properties = new Properties();
           
            // 若要配多个服务器，用逗号隔开
            // 注：服务器要开放端口，若云服务器还要在server.properties配置内网IP和外网IP
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "39.105.136.112:9092");
            properties.put(ProducerConfig.CLIENT_ID_CONFIG, "my-producer");
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
                           IntegerSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
                           StringSerializer.class.getName());
    
            // 构造Client无非是：new 或 工厂模式
            producer = new KafkaProducer<Integer, String>(properties);
            this.topic = topic;
        }
    
        public void run() {
        
            int num = 0;
            String msg = "kafka practice msg: " + num;
            while (num < 20) {
        
                try {
        
                    // 发送消息send()！！! 同步调用
                    // Future.get()会阻塞，等待返回结果......
                    RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, msg)).get();
          			// 等上面get到结果了，才能执行这里
                    System.out.println(recordMetadata.offset() + "->" + recordMetadata.partition() + 
                                       "->" + recordMetadata.topic());
                    TimeUnit.SECONDS.sleep(2);
                    num++;
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                } catch (ExecutionException e) {
        
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) {
        
            // 传入test主题
            new MyKafKaProducer("test").start();
        }
    }

![在这里插入图片描述][20201114163857472.png_pic_center]

### 1.2 异步调用 ###

public void run() {
                
        int num=0;        
        while(num<50){
                    
            String msg="pratice test message:"+num;            
            try {
                       
                // 异步调用 
                producer.send(new ProducerRecord<>(topic, msg), new Callback() {
             
                    @Override 
                    // 回调函数
                    // 注：这里其实还能通过lamada用函数式接口(metadata, exception) -> {}
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
           
                  		System.out.println("callback: "+ recordMetadata.offset()
                                           +"->"+recordMetadata.partition());                 
                    }               
                });               
                TimeUnit.SECONDS.sleep(2);         
                num++;            
            } catch (InterruptedException e) {
                  
                e.printStackTrace();         
            }    
        }   
      }

从本质上来说，kafka都是采用异步的方式来发送消息到broker，但是kafka并不是每次发送消息都会直接发送到broker上，而是把消息放到了一个发送队列中，等积累到一定量（batch.size）或等待时间了（linger.ms），就通过一个后台线程不断从队列取出消息进行发送，发送成功后会触发callback。

*  batch.size
    
    生产者发送多个消息到broker上的同一个分区时，为了减少网络请求带来的性能开销，通过批量的方式来提交消息，可以通过这个参数来控制批量提交的字节数大小，默认大小是16384byte,也就是16kb， 意味着当一批消息大小达到指定的batch.size的时候会统一发送
 *  linger.ms
    
    Producer默认会把两次发送时间间隔内收集到的所有Requests进行一次聚合然后再发送，以此提高吞吐量，而linger.ms就是为每次发送到broker的请求增加一些delay，以此来聚合更多的Message请求。

这个有点类似TCP里面的Nagle算法，在TCP协议的传输中，为了减少大量小数据包的发送，采用了Nagle 算法，也就是基于小包的等-停协议。

只要满足上二者之一，就能将消息发送到broker上

## 2.Consumer ##

public class MykafkaConsumer extends Thread{
        
    
        KafkaConsumer<Integer, String> consumer;
        String topic;
    
        public MykafkaConsumer(String topic) {
        
            // 构建连接配置，这里是ConsumerConfig
            Properties properties = new Properties();
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "39.105.136.112:9092");
            properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "my-consumer");
            // 反序列化，这里是Deserializer
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                           IntegerDeserializer.class.getName());
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                           StringDeserializer.class.getName());
                           
            // 以下是Producer没有的配置
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-gid"); // 要加入的group
            properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000"); // 超时，心跳
            properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 自动提交（批量）
            properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 新group消费为位置
    
            consumer = new KafkaConsumer<>(properties);
            this.topic = topic;
        }
    
        public void run() {
        
            // 死循环不断消费消息
            while (true) {
        
                // 绑定订阅主题
                // 注：Collections.singleton返回一个单元素&不可修改Set集合，
                // 同样的还有singletonList，singletonMap
                consumer.subscribe(Collections.singleton(this.topic));
                // 接收消息 POLL()！！！
                ConsumerRecords<Integer, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
                // 注：一行的lamada表达式可以不用{}
                consumerRecords.forEach(record -> System.out.println(record.key() + "->" + 
                                                     record.value() + "->" + record.offset()));
            }
        }
    
        public static void main(String[] args) {
        
            // 拉取test主题的消息
            new MykafkaConsumer("test").start();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]  
Consumer中那些Producer没有的配置到底是什么意思呢？请看这篇[【Kafka】基本使用：关于Consumer的几个概念][Kafka_Consumer]

[20201114163857472.png_pic_center]: /images/20221120/81e21b74ca4e49989be458ee6117ffad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20201114172331550.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw==,size_16,color_FFFFFF,t_70#pic_center
[Kafka_Consumer]: https://blog.csdn.net/weixin_43935927/article/details/109697632