Java編寫程序將數據存入Kafka中

Kafka是一個類似於RabbitMQ的消息系統，它的主要功能是消息的發布和訂閱、處理和存儲。

1.它類似於一個消息系統，讀寫流式的數據。

2.編寫可擴展的流應用處理程序，用於實時事件響應的場景。

3.安全的將流式的數據存儲在一個分布式，有副本備份，容錯的集群。

 

本篇博文主要介紹如何使用Java編寫程序將數據寫入到Kafka中，即Kafka生產者，並不涉及Kafka消費者。另外，像Spark，Storm等都有相應的程序從Kafka消費者中獲取數據的方法，直接調用即可。

 

Kafka的運行需要Zookeeper的幫助，所以，需要先安裝Zookeeper。

 

1.先啟動Zookeeper

bin/zookeeper-server-start.sh config/zookeeper.properties

 再啟動Kafka服務器：

bin/kafka-server-start.sh config/server.properties

2.創建一個Topic：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

 顯示topic

bin/kafka-topics.sh --list --zookeeper localhost:2181

 也可以在程序中進行topic的創建。

3.發送消息：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

4.接收消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

 

 

 

下面，是本次的程序：

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

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

public class MyProducer extends Thread {
    private final KafkaProducer<Integer, String> producer;
    private final String topic;
    private final Boolean isAsync;

    public MyProducer(String topic, Boolean isAsync) {
        Properties prop = new Properties();
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<Integer, String>(prop);
        this.topic = topic;
        this.isAsync = isAsync;
    }

    public void run() {
        int messageNo = 1;
        while (true) {
            String messageStr = "Message_" + messageNo;
            long startTime = System.currentTimeMillis();
            if (isAsync) {
                producer.send(new ProducerRecord<Integer, String>(topic, messageNo, messageStr), new DemoCallback(startTime, messageNo, messageStr));
            } else {
                try {
                    producer.send(new ProducerRecord<Integer, String>(topic, messageNo, messageStr)).get();
                    System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")");
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }
            ++messageNo;
        }
    }
    public static void main(String[] args) {
        boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync");

        MyProducer producerThread = new MyProducer("test", isAsync);
        producerThread.start();

    }
}

class DemoCallback implements Callback {
    private final long startTime;
    private final int key;
    private final String message;

    public DemoCallback(long startTime, int key, String message) {
        this.startTime = startTime;
        this.key = key;
        this.message = message;
    }

    public void onCompletion(RecordMetadata metadata, Exception exception) {
        long elapsedTime = System.currentTimeMillis() - startTime;
        if (metadata != null) {
            System.out.println(
                    "message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
                            "), " +
                            "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
        } else {
            exception.printStackTrace();
        }
    }
}

 

好了，完成！