通過開源Kafka用戶端寫入Lindorm流引擎資料 -

Lindorm流引擎完全相容開源Kafka API，您可以通過Kafka API編寫程式寫入Lindorm流引擎資料，也可以通過開源的三方工具採集並寫入Lindorm流引擎資料，例如FluentD、Debezium等。本文介紹通過開源Kafka用戶端串連Lindorm流引擎並寫入Lindorm流引擎資料的程式碼範例。

前提條件

已安裝Java環境，要求安裝JDK 1.7及以上版本。
已將用戶端IP地址添加至Lindorm執行個體的白名單中，具體操作請參見設定白名單。
已擷取Lindorm流引擎的Lindorm Stream Kafka地址，具體操作請參見查看串連地址。
說明
Lindorm流引擎的Lindorm Stream Kafka地址為專用網路地址，需確保應用程式部署的環境和Lindorm執行個體使用相同的專用網路ID。

操作步驟

下載開源Kafka用戶端。在pom.xml中配置Maven依賴，具體內容如下：

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

串連Lindorm流引擎並寫入Lindorm流引擎資料。完整的程式碼範例如下：

說明

寫入的資料格式支援JSON、Avro和CSV。
代碼中Lindorm Stream Kafka地址為專用網路地址，擷取方法請參見查看串連地址。

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.codehaus.jettison.json.JSONObject;

import java.util.Properties;
import java.util.concurrent.Future;

public class KafkaToLindormStreamDemo {

    public static void main(String[] args) {
        Properties props = new Properties();

        //設定Lindorm Stream Kafka地址，這個Lindorm Stream Kafka地址為專用網路地址，需確保應用程式部署的環境和Lindorm執行個體使用相同的專用網路ID。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "Lindorm Stream Kafka地址");
       //指定資料流表的物理資料存放區在某個Topic上
        String topic = "log_topic";

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        try {
            JSONObject json = new JSONObject();
            //寫入流引擎資料
            json.put("timestamp", System.currentTimeMillis());
            json.put("loglevel", "ERROR");
            json.put("thread", "[ReportFinishedTask7-thread-4]");
            json.put("class", "engine.ImporterTaskManager(318)");
            json.put("detail", "Remove tasks fail: job name=e35318e5-52ea-48ab-ad2a-0144ffc6955e , task name=prepare_e35318e5-52ea-48ab-ad2a-0144ffc6955e , runningTasks=0");
            Future<RecordMetadata> future = producer.send(
          new ProducerRecord<String, String>(topic, json.getString("thread") + json.getLong("timestamp"),
              json.toString()));
            producer.flush();
            try {
                RecordMetadata recordMetadata = future.get();
                System.out.println("Produce ok:" + recordMetadata.toString());
            } catch (Throwable t) {
                System.out.println("Produce exception " + t.getMessage());
                t.printStackTrace();
            }
        } catch (Exception e) {
            System.out.println("Produce exception " + e.getMessage());
            e.printStackTrace();
        }
    }
}