Kafka.java 文件源码

public void run() {
    try {
        while(true){
            ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
            while (it.hasNext()) {
                String m = null;
                try {
                    m = new String(it.next().message(),
                            this.kafkaInput.encoding);
                    Map<String, Object> event = this.decoder
                            .decode(m);
                    if (event!=null&&event.size()>0){
                        this.kafkaInput.process(event);
                    } 
                } catch (Exception e) {
                    logger.error("process event:{} failed:{}",m,e.getCause());
                }
            }
        }
    } catch (Exception t) {
        logger.error("kakfa Consumer fetch is error:{}",t.getCause());
    }
}