f:id:kotapontan:20210612213528p:plain 分散メッセージキューとしてメジャーなApache Kafkaをプロジェクトで使う機会がありそうなので、HelloWorldな感じで動かしてみた。

目標

Apache Kafkaのインストール
公式ページのQuickStartをやってみる
Javaからキューを操作

Apache Kafkaのインストール

Apache Kafkaの公式ページを参省してインストールする。
インストール環境は、以下のとおり。

OS	Version
CentOS	7.6.1810 (Core)

kafka.apache.org STEP1からSTEP4までを実施。

解凍とディレクトリ移動

tar -xzf kafka_2.13-2.8.0.tgz
cd kafka_2.13-2.8.0

起動

bin/zookeeper-server-start.sh config/zookeeper.properties

別のターミナルを起動して、以下を実行

# bin/kafka-server-start.sh config/server.properties

イベント保管用のトピックを作成

別のターミナルを起動して、以下を実行

# bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
Created topic quickstart-events.

bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events    TopicId: 0b86Irq7Qn2_OK1lwHG0dg PartitionCount: 1   ReplicationFactor: 1    Configs: segment.bytes=1073741824
    Topic: quickstart-events    Partition: 0    Leader: 0   Replicas: 0 Isr: 0

トピックにイベントを追加する

トピックにイベントを追加する。「This is my first event」と「This is my second event」を入力し、Ctrl＋Cを押下する。

bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
> This is my first event
> This is my second event

トピックからイベントを取得する

トピックからイベントを取得する。スクリプトを終了するときはCtrl+Cを押下する。追加したイベントのメッセージが取得できた。

bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

Javaから動かしてみる

スクリプトを使用して動かしてみたが、次はJavaのプログラムからキューにメッセージを追加してみる。
ソース自体はApache Kafkaが公開しているJavadocに書かれている内容で実行することができた。

Producerのソースコード

package jp.kafkasample.producer;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class SampleProducer {
    
    public static void main(String args[]) {
        
        SampleProducer producer = new SampleProducer();
        producer.publish();
    }

    private void publish() {
        
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.56.10:9092");
        props.put("acks",  "all");
        props.put("retries",  0);
        props.put("linger.ms",  1);
        props.put("key.serializer",  "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",  "org.apache.kafka.common.serialization.StringSerializer");
        
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 100; i++) {
                // my-topicというトピックにキーバリューで登録
                producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), "value is " + Integer.toString(i)));
            }
        }
        
        System.out.println("End!!");
    }

}

Consumerのソースコード

package jp.kafkasample.consumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class SampleConsumer {
    
    public static void main(String args[]) {
        
        SampleConsumer consumer = new SampleConsumer();
        consumer.pull();
    }

    private void pull() {

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "192.168.56.10:9092");
        props.setProperty("group.id",  "test");
        props.setProperty("enable.auto.commit",  "true");
        props.setProperty("auto.commit.interval.ms", "1000");
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("my-topic", "quickstart-events"));
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record: records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }

}

感想

インストールからプログラムの動作まで、マニュアルがきちんと書かれているのでスムーズに実施することができた。
まだ、KafkaのHello Worldぐらいのレベルなので、今後、実務で使うために必要なメッセージの保存期間などを調べていけたらと思う。

参考

qiita.com

ヒトリ歩き

愚痴とかいろいろ書きます

Apache KafkaのHello World

目標