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

引入依赖：

<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();
    }
}

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();
    }
}