Consumer.java 文件源码

/**
 * Starts the consumer thread.
 */

@Override
public void run() {
    log.debug("Starting consumer for topic {}", topic);
    ConsumerIterator<byte[], byte[]> it = stream.iterator();

    // For each message present on the partition...
    while (it.hasNext()) {
        Map<String, Object> event = null;

        // Parse it with the parser associated with the topic
        try {
            event = parser.parse(new String(it.next().message(), "UTF-8"));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }

        // Send it to the source
        if (event != null) {
            source.send(topic.getName(), event);
        }
    }

    log.debug("Finished consumer for topic {}", topic);
}