java 操作 kafka 最基本 demo

数据流通道

2018-08-01

102

0

技术:java8 + kafka client

运行环境:IDEA 15.2 + jdk8 + windows 10

demo功能:提供一个java 操作kafka最基本demo

kafka下载:https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/

pom


    org.apache.kafka
    kafka-clients
    0.11.0.0

生产者

public class KafkaProducerSender {
    public static Properties properties = new Properties();

    static {
        properties.put("bootstrap.servers", "10.32.8.5:9092,10.32.0.116:9092,10.32.8.10:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    }

    public static String topic = "HelloWorld";


    public static void sendDemo() {
        System.out.println("start produce...");
        Producer<String, String> producer = null;
        try {
            producer = new KafkaProducer<>(properties);
            for (int i = 0; i < 100; i++) {
                String msg = "Message " + i;
                producer.send(new ProducerRecord<>(topic, msg));
                System.out.println("Sent:" + msg);
            }
        } catch (Exception e) {
            e.printStackTrace();

        } finally {
            producer.close();
        }
    }
}

消费者

public class KafkaConsumerReciever {
    public static void recieve() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", KafkaProducerSender.properties.get("bootstrap.servers"));
        properties.put("group.id", "group-1");
        properties.put("enable.auto.commit", "true");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "30000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.subscribe(Arrays.asList(KafkaProducerSender.topic));
        while (true) {
            System.out.println("i'am alive...");
            ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
            if (records.isEmpty()) {
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, value = %s", record.offset(), record.value());
                System.out.println();
            }
        }
    }
}

测试

public class Application {
    public static void main(String[] args) {
        KafkaProducerSender.sendDemo();
        KafkaConsumerReciever.recieve();
    }
}

欢迎添加微信,互相学习↑↑↑ -_-

发表评论

全部评论:0条

白老虎

programming is not only to solve problems, ways to think