KafkaConsumerThread.java 文件源码

@Override
public void run() {
    ConsumerIterator<byte[], byte[]> it = stream.iterator();
    while (it.hasNext()) {
        MessageAndMetadata<byte[], byte[]> mam = it.next();
        String jsonStr = "";
        try {
            jsonStr = new String(mam.message());
            JSONObject jsonObject = JSONObject.parseObject(jsonStr);
            LogcenterConfig config = LogConfigCache.getLogConfigCache(jsonObject);
            IStorageApi iStorageApi = ServiceRegister.getInstance().getProvider(config.getStorageType());
            iStorageApi.save(jsonObject);
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("partition[" + mam.partition() + "]," + "offset[" + mam.offset() + "], " + jsonStr, e);
            continue;
        }
    }
}