java kafka 生产者消费者 高级API

Java中提供高级的API，相对于低级API（更小的粒度控制消费）使用起来非常方便。

 pom:

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>1.0.0</version>
        </dependency>

一、修改kafka   server.porperties的ip是你kafka服务的ip

listeners=PLAINTEXT://192.168.111.130:9092

二、生产者的例子

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaProducerDemo {
    private final Producer<String, String> kafkaProdcer;
    public final static String TOPIC = "JAVA_TOPIC";

    private KafkaProducerDemo() {
        kafkaProdcer = createKafkaProducer();
    }

    private Producer<String, String> createKafkaProducer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.111.130:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
        return kafkaProducer;
    }

    void produce() {
        for (int i = 0; i < 10; i++) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            final String key = "key" + i;
            String data = "hello kafka message:" + key;
            kafkaProdcer.send(new ProducerRecord<String, String>(TOPIC, key, data), new Callback() {
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("发送key" + key + "成功");
                }
            });
        }
    }

    public static void main(String[] args) {
        KafkaProducerDemo kafkaProducerDemo = new KafkaProducerDemo();
        kafkaProducerDemo.produce();
    }

}

用properties构造一个Producer的实例，然后调用send方法，传入数据，还有一个回调函数。

可以看到数据已经进来了。

注意：kafka producer支持同步发送、异步发送、异步发送+回调函数方式。

1、同步方式会按顺序发送，打印出来的结果是按发送的顺序：

for (int i = 0; i < 1000; i++) {
    RecordMetadata test = producer.send(new ProducerRecord<String, String>("test", Integer.toString(i), "hello world-" + i)).get();
    System.out.println(test);
}

2、回调函数里面可以对成功或者失败，分支判断，进行业务上的进一步处理。甚至可以把失败的消息存储下来。

for (int i = 0; i < 10; i++) {
    producer.send(new ProducerRecord<String, String>("test", i + "", "xxx-" + i), new Callback() {
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e != null) {
                e.printStackTrace();
            } else {
                System.out.println("发送成功");
            }
        }
    });
}

注：回调函数里面onCompletion方法其实是阻塞的! 如果进行延时，会逐个执行，不会同时并发跑，但是发送数据任然是异步的。

三、消费者例子

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerDemo {
    private final KafkaConsumer<String, String> consumer;
    private KafkaConsumerDemo(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.111.130:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "false");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(props);
    }
    void consume(){
        consumer.subscribe(Arrays.asList(KafkaProducerDemo.TOPIC));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records){
                System.out.println("I'm coming");
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
    public static void main(String[] args) {
        KafkaConsumerDemo kafkaConsumerDemo = new KafkaConsumerDemo();
        kafkaConsumerDemo.consume();
    }
}

正常启动是看不到东西的， 两个同时启动才有。消费者只看接下来有哪些生产者发来新的消息。

props.put("enable.auto.commit", "true");

这个的意思是，消费后自动改变偏移量。如果不添加这个，就会在服务器存的offset开始消费，并且不会改变offset的值。

如果为false, 可以看到不管消费几次，服务端存储的始终是offset的值都不会改变，需要手动提交offset。

如果想让consumer从头开始消费，可以设置：

props.put("auto.offset.reset", "earliest");

这个只对新建的组有效，如果一个组已经消费过，offset的值已经存在服务端了，这样设置不起作用的，只会从服务端存储的offset开始消费。不设置默认是latest，就是从最新的开始消费。