Kafka

Apache KafkaKafka - 图1open in new window is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.

Coding Example

kafka Producer Producing Data Java Code Example

  1.     Properties props = new Properties();
  2.     props.put("bootstrap.servers", "127.0.0.1:9092");
  3.     props.put("key.serializer", StringSerializer.class);
  4.     props.put("value.serializer", StringSerializer.class);
  5.     KafkaProducer<String, String> producer = new KafkaProducer<>(props);
  6.     producer.send(
  7.         new ProducerRecord<>(
  8.             "Kafka-Test", "key", "root.kafka," + System.currentTimeMillis() + ",value,INT32,100"));
  9.     producer.close();

kafka Consumer Receiving Data Java Code Example

  1.     Properties props = new Properties();
  2.     props.put("bootstrap.servers", "127.0.0.1:9092");
  3.     props.put("key.deserializer", StringDeserializer.class);
  4.     props.put("value.deserializer", StringDeserializer.class);
  5.     props.put("auto.offset.reset", "earliest");
  6.     props.put("group.id", "Kafka-Test");
  7.     KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
  8.     kafkaConsumer.subscribe(Collections.singleton("Kafka-Test"));
  9.     ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1));

Example of Java Code Stored in IoTDB Server

  1.     SessionPool pool =
  2.         new SessionPool.Builder()
  3.             .host("127.0.0.1")
  4.             .port(6667)
  5.             .user("root")
  6.             .password("root")
  7.             .maxSize(3)
  8.             .build();
  9.     List<String> datas = new ArrayList<>(records.count());
  10.     for (ConsumerRecord<String, String> record : records) {
  11.       datas.add(record.value());
  12.     }
  13.     int size = datas.size();
  14.     List<String> deviceIds = new ArrayList<>(size);
  15.     List<Long> times = new ArrayList<>(size);
  16.     List<List<String>> measurementsList = new ArrayList<>(size);
  17.     List<List<TSDataType>> typesList = new ArrayList<>(size);
  18.     List<List<Object>> valuesList = new ArrayList<>(size);
  19.     for (String data : datas) {
  20.       String[] dataArray = data.split(",");
  21.       String device = dataArray[0];
  22.       long time = Long.parseLong(dataArray[1]);
  23.       List<String> measurements = Arrays.asList(dataArray[2].split(":"));
  24.       List<TSDataType> types = new ArrayList<>();
  25.       for (String type : dataArray[3].split(":")) {
  26.         types.add(TSDataType.valueOf(type));
  27.       }
  28.       List<Object> values = new ArrayList<>();
  29.       String[] valuesStr = dataArray[4].split(":");
  30.       for (int i = 0; i < valuesStr.length; i++) {
  31.         switch (types.get(i)) {
  32.           case INT64:
  33.             values.add(Long.parseLong(valuesStr[i]));
  34.             break;
  35.           case DOUBLE:
  36.             values.add(Double.parseDouble(valuesStr[i]));
  37.             break;
  38.           case INT32:
  39.             values.add(Integer.parseInt(valuesStr[i]));
  40.             break;
  41.           case TEXT:
  42.             values.add(valuesStr[i]);
  43.             break;
  44.           case FLOAT:
  45.             values.add(Float.parseFloat(valuesStr[i]));
  46.             break;
  47.           case BOOLEAN:
  48.             values.add(Boolean.parseBoolean(valuesStr[i]));
  49.             break;
  50.         }
  51.       }
  52.       deviceIds.add(device);
  53.       times.add(time);
  54.       measurementsList.add(measurements);
  55.       typesList.add(types);
  56.       valuesList.add(values);
  57.     }
  58.     pool.insertRecords(deviceIds, times, measurementsList, typesList, valuesList);