[Kafka]kafka consumer 0.9ノート

へんか

コンセプト

コンフィギュレーション

管理

bin/kafka-consumer-groups --bootstrap-server host:9092 --new-consumer --list

例

Basic Poll Loop

public abstract class BasicConsumeLoop implements Runnable {
  private final KafkaConsumer consumer;
  private final List topics;
  private final AtomicBoolean shutdown;
  private final CountDownLatch shutdownLatch;
  public BasicConsumeLoop(Properties config, List topics) {
    this.consumer = new KafkaConsumer<>(config);
    this.topics = topics; this.shutdown = new AtomicBoolean(false);
  this.shutdownLatch = new CountDownLatch(1); }
  public abstract void process(ConsumerRecord record);
  public void run() {
    try { 
      consumer.subscribe(topics);
      while (!shutdown.get()) {
        //    wakeup ， 
        // ConsumerRecords records = consumer.poll(Long.MAX_VALUE);
        //  consumer.wakeup();
        ConsumerRecords records = consumer.poll(500);
        records.forEach(record -> process(record));
      }
      } finally {
        consumer.close();
        shutdownLatch.countDown();
      }
  }
  public void shutdown() throws InterruptedException {
    shutdown.set(true);
    //  consumer.close 
    shutdownLatch.await();
  }
}

送信offset

private void doCommitSync() {
  try {
    consumer.commitSync();
  } catch (WakeupException e) {
    // we're shutting down, but finish the commit first and then
    // rethrow the exception so that the main loop can exit
    doCommitSync();
    throw e;
  } catch (CommitFailedException e) {
    // the commit failed with an unrecoverable error. if there is any
    // internal state which depended on the commit, you can clean it
    // up here. otherwise it's reasonable to ignore the error and go on
    log.debug("Commit failed", e);
  }
}

public void run() {
  try {
    consumer.subscribe(topics);

    while (true) {
      ConsumerRecords records = consumer.poll(Long.MAX_VALUE);
      records.forEach(record -> process(record));
      doCommitSync();
    }
  } catch (WakeupException e) {
    // ignore, we're closing
  } catch (Exception e) {
    log.error("Unexpected error", e);
    } finally {
      consumer.close();
      shutdownLatch.countDown();
    }
}

Delivery Guarantees####

private boolean doCommitSync() {
  try {
      consumer.commitSync();
      return true;
  } catch (CommitFailedException e) {
    // the commit failed with an unrecoverable error. if there is any
    // internal state which depended on the commit, you can clean it
    // up here. otherwise it's reasonable to ignore the error and go on
    log.debug("Commit failed", e);
    return false;
  }
}

public void run() {
  try {
    consumer.subscribe(topics);

    while (true) {
      ConsumerRecords records = consumer.poll(Long.MAX_VALUE);
      if (doCommitSync())
        records.forEach(record -> process(record));
    }
  } catch (WakeupException e) {
    // ignore, we're closing
  } catch (Exception e) {
    log.error("Unexpected error", e);
    } finally {
      consumer.close();
      shutdownLatch.countDown();
    }
}

非同期offsetコミット

public void run() {
  try {
    consumer.subscribe(topics);

    while (true) {
      ConsumerRecords records = consumer.poll(Long.MAX_VALUE);
      records.forEach(record -> process(record));
      consumer.commitAsync(new OffsetCommitCallback() {
        public void onComplete(Map offsets, Exception exception) {
          if (e != null)
            log.debug("Commit failed for offsets {}", offsets, e);
        }
      });
    }
  } catch (WakeupException e) {
    // ignore, we're closing
  } catch (Exception e) {
    log.error("Unexpected error", e);
  } finally {
    consumer.close();
    shutdownLatch.countDown();
  }
}

private void doCommitSync() {
  try {
    consumer.commitSync();
  } catch (WakeupException e) {
    // we're shutting down, but finish the commit first and then
    // rethrow the exception so that the main loop can exit
    doCommitSync();
    throw e;
  } catch (CommitFailedException e) {
    // the commit failed with an unrecoverable error. if there is any
    // internal state which depended on the commit, you can clean it
    // up here. otherwise it's reasonable to ignore the error and go on
    log.debug("Commit failed", e);
  }
}

public void run() {
  try {
    consumer.subscribe(topics, new ConsumerRebalanceListener() {
      public void onPartitionsRevoked(Collection partitions) {
        doCommitSync();
      }

      public void onPartitionsAssigned(Collection partitions) {}
    });

    while (true) {
      ConsumerRecords records = consumer.poll(Long.MAX_VALUE);
      records.forEach(record -> process(record));
      consumer.commitAsync();
    }
  } catch (WakeupException e) {
    // ignore, we're closing
  } catch (Exception e) {
    log.error("Unexpected error", e);
  } finally {
    try {
      doCommitSync();
    } finally {
      consumer.close();
      shutdownLatch.countDown();
    }
  }
}

マルチスレッド

public class KafkaConsumerRunner implements Runnable {
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final KafkaConsumer consumer;

     public void run() {
         try {
             consumer.subscribe(Arrays.asList("topic"));
             while (!closed.get()) {
                 ConsumerRecords records = consumer.poll(10000);
                 // Handle new records
             }
         } catch (WakeupException e) {
             // Ignore exception if closing
             if (!closed.get()) throw e;
         } finally {
             consumer.close();
         }
     }

     // Shutdown hook which can be called from a separate thread
     public void shutdown() {
         closed.set(true);
         consumer.wakeup();
     }
 }

 closed.set(true);
 consumer.wakeup();