kafka producer consumer 구현하기

kafka producer과 consumer를 java로 구현해보자. 일단 카프카가 설치 되어 있어야 한다. 카프카 설치는 여기에서 보면 된다. 아주 간단한 샘플용이니 참고만 하길 바란다. kafka version은 0.9.0.1 버전으로 하였다. 바로 소스를 보자.

<dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.9.0.1</version>
    </dependency>

    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.1.7</version>
    </dependency>
</dependencies>

일단 디펜더시는 카프카 클라이언트와 필자는 logback을 자주 써서 logback으로 디펜더시를 설정 하였다. 다음 코드는 producer 코드이다. 간단하게 만들었다.

public static void main(String[] args) throws IOException {

  Properties configs = new Properties();
  configs.put("bootstrap.servers", "localhost:9092");
  configs.put("acks", "all");
  configs.put("block.on.buffer.full", "true");
  configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

  KafkaProducer<String, String> producer = new KafkaProducer<>(configs);

  producer.send(new ProducerRecord<>("test", "hello world"),
    (metadata, exception) -> {
      if (metadata != null) {
        System.out.println(
          "partition(" + metadata.partition() + "), offset(" + metadata.offset() + ")");
      } else {
        exception.printStackTrace();
      }
    });
  producer.flush();
  producer.close();
}

test라는 topic을 예전에 생성 해놔서 topic을 test로 지정해놨다. http://kafka.apache.org/documentation.html#producerconfigs 여기 가면 producer config 파라미터들이 정리되어 있다. 소스는 딱히 설명 하지 않아도 보면 알 듯 싶다.

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback)

중요한건 send API가 실제 보내는 API이다. ProducerRecord 클래스의 첫번째 인자는 topic이고 두번째 인자는 value로 지정 되어 있다. 만약 콘솔로 consumer를 띄었다면 콘솔에 찍혀 있는 걸 확인 할 수 있다. 다음으론 consumer예제 클래스이다.

public static void main(String[] args) {
  Properties configs = new Properties();
  configs.put("bootstrap.servers", "localhost:9092");
  configs.put("session.timeout.ms", "10000");
  configs.put("group.id", "test");
  configs.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);
  consumer.subscribe(Arrays.asList("test"));
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(500);
    for (ConsumerRecord<String, String> record : records) {
      switch (record.topic()) {
        case "test":
          System.out.println(record.value());
          break;
        default:
          throw new IllegalStateException("get message on topic " + record.topic());
      }
    }
  }
}

http://kafka.apache.org/documentation.html#consumerconfigs 마찬가지로 컨슈머의 설정은 해당 URL에 잘 정리 되어 있다. 예전 config와 새로운 config 설정이 있으니 참고하기 바란다.

consumer.subscribe(Arrays.asList("test"))

test라는 topic을 구독하겠다는 것이다 List로 들어가니 여러 topic을 구독 할 수 있다. 그리고 해당 topic(여기서는 test)으로 오면 실제 값을 출력 해준다. 이렇게 아주 간단하게 producer와 consumer를 구현해봤다. 물론 간단한 코드 이므로 실 운영에서는 사용하지 못할 듯하다. 좀더 리펙토링이 필요 할 것이다. 참고 정도만 하면 될듯 싶다. 소스는 github에 올려놔야 겠다.