KafkaConsumer.java 文件源码

public void run() {
    ConsumerIterator<String, String> it = stream.iterator();
    while (it.hasNext()) {
        MessageAndMetadata<String, String> consumerIterator = it.next();
        String uploadMessage = consumerIterator.message();
        System.out.println(Thread.currentThread().getName()
                + " from partiton[" + consumerIterator.partition() + "]: "
                + uploadMessage);
        try {
            sendDataToIotdb.writeData(uploadMessage); // upload data to the IoTDB database

        } catch (Exception ex) {
            System.out.println("SQLException: " + ex.getMessage());
        }
    }
}