RunnableConsumer.java 文件源码

public void run() {

        try {

            ConsumerIterator<String, String> it = stream.iterator();
            while (it.hasNext()) {
                MessageAndMetadata<String, String> messageAndMetadata = it.next();
                String message = messageAndMetadata.message();
                messageHandler.handle(message);
                meter.mark();
            }
            messageHandler.flush();

        } catch (ConsumerTimeoutException e) {
            messageHandler.flush();
        }
    }