Kafkaのインストール

Kafkaの最新版をダウンロードします。
Kafkaの本体をダウンロードする時には、wgetコマンドを使ったことがあります。wgetの使い方については、僕の　Macでwgetコマンドを使ってみるメモをご参照ください。

$ cd /usr/local/opt 
$ wget http://ftp.riken.jp/net/apache/kafka/2.1.0/kafka_2.12-2.1.0.tgz
$ tar -zxf kafka_2.12-2.1.0.tgz
ln -s kafka_2.12-2.1.0 kafka

kafka_2.12-2.1.0.tgz中の「1.12」は、Scalaのバージョンに指しています。Scalaをセットアップしたい場合は、合わせるバージョンを選べた方が良いと思います。
「2.1.0」は、Kafka自分自身のバージョンです。

ZooKeeperのインスタンス取得

Kafkaは、ZooKeeperを使います。Kafkaにパッケージ化されているスクリプトを使ってZooKeeperのインスタンスを取得します。

$ cd kafka
$ bin/zookeeper-server-start.sh config/zookeeper.properties
[2019-02-11 10:33:36,916] INFO Server environment:user.dir=/usr/local/opt/kafka_2.12-2.1.0 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-02-11 10:33:36,944] INFO tickTime set to 3000 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-02-11 10:33:36,944] INFO minSessionTimeout set to -1 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-02-11 10:33:36,944] INFO maxSessionTimeout set to -1 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-02-11 10:33:36,981] INFO Using org.apache.zookeeper.server.NIOServerCnxnFactory as server connection factory (org.apache.zookeeper.server.ServerCnxnFactory)
[2019-02-11 10:33:37,021] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)

Kafkaサーバーの起動

Kafkaサーバーを起動します。

$ bin/kafka-server-start.sh config/server.properties
[2019-02-11 10:38:54,256] INFO [SocketServer brokerId=0] Started processors for 1 acceptors (kafka.network.SocketServer)
[2019-02-11 10:38:54,287] INFO Kafka version : 2.1.0 (org.apache.kafka.common.utils.AppInfoParser)
[2019-02-11 10:38:54,288] INFO Kafka commitId : 809be928f1ae004e (org.apache.kafka.common.utils.AppInfoParser)
[2019-02-11 10:38:54,290] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

トピックの作成、確認

1つのパーティション(partitions)と1つのレプリカ(replication)だけを持つ”test”という名前のトピックを作成しましょう。

$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test01
Created topic "test01".

list トピックコマンドを実行して、作成されていたトピックを確認してみましょう。

$ bin/kafka-topics.sh --list --zookeeper localhost:2181
test01

プロデューサーでメッセージを送信

Kafkaは、物理のファイル或いは標準入力から受取ったメッセージを、Kafkaクラスタに送信するコマンドラインクライアントが付属しています。デフォルトで各行は別個のメッセージとして送信されています。
プロデューサを実行して、次に、コマンドラインにメッセージをコンソールに入力します。

$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test01
>hello kafka!

コンシューマーでメッセージを受信

Kafkaはメッセージを標準出力に出力するコマンドラインコンシューマも持ちます。

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test01 --from-beginning
hello kafka!