ConsumerAuditUtil.java 文件源码

@Override
public void afterPropertiesSet() throws Exception {
  auditExecutorService.submit(() -> {
    while (!auditStopped.get() && !Thread.currentThread().isInterrupted()) {
      List<ConsumerAudit> toAudit = Lists.newArrayList();
      try {
        Queues.drain(audits, toAudit, BATCH_SIZE, BATCH_TIMEOUT, BATCH_TIMEUNIT);
        if (!toAudit.isEmpty()) {
          consumerService.createConsumerAudits(toAudit);
        }
      } catch (Throwable ex) {
        Tracer.logError(ex);
      }
    }
  });
}