はじめに
前回試した、Spring Cloud Streamでredisにつなげるやり方でkafkaでやってみる。
環境
-
Mac
-
IntelliJ 15
-
kafka
手順
IntelliJからプロジェクトを作成する
プロジェクトを作成メニューを開く
左のペインから「Spring Initializr」を選択して、「Next」ボタンをクリック
「Name」に適当に名前をつけて「Next」ボタンをクリック
「Dependencies」からStream Kafkaを選択して、「Next」ボタンをクリック
「Project Name」に適当に名前をいれて「Finish」ボタンをクリック
※しばらくしてGradleの設定画面がでてきたら、「OK」ボタンをクリックする。
プロジェクトができあがる
デフォルトの状態で起動してみる
自動的に生成されたsrc/main/java/xxx.xxx/XXXXXXApplicationを右クリックして、「Run 'XXXX'」を選択
立ち上がる
この時点ではまだkafkaに接続しないようだ。
kafkaに接続できるように設定する
kafkaバインド用のクラスを作成する
XXXXXApplicationクラスと同じパッケージ内にkafkaバインド用のクラスを作成する。
作成するクラスは、spring-cloud-streamのgithubのサンプルを参考にして適当に作成する。
クラス名は、SampleSourceにした。
fixedDelayは何秒おきにメッセージを送るかという設定
maxMessagesPerPollは１回に最大何件メッセージを送るかという設定
package com.example;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.support.GenericMessage;
import java.time.LocalDateTime;
@EnableBinding(Source.class)
public class SampleSource {
@Bean
@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "1000", maxMessagesPerPoll = "1"))
public MessageSource<?> source() {
return () -> new GenericMessage<>(LocalDateTime.now());
}
}
ラムダ式でコンパイルエラーがでる場合は、Projectの言語レベルを変更する。
プロジェクトを右クリックして、「Open Module Settings」を選択する。
「Language level」を「8 - Lambdas, type annotations etc.」に変更して「OK」ボタンをクリック
application.propertiesをymlに変更する
application.propertiesのファイルを右クリックして、「Refactor」-「Rename...」
拡張子をymlに変更して「Refactor」ボタンをクリック
application.ymlにkafkaのtopicを指定する
topicは存在しないtopicを指定しもOK。存在しない場合は、勝手に作成される。
spring:
cloud:
stream:
bindings:
output: spring.cloud.stream.test.topic
Sourceクラスを作成した状態でアプリケーションを立ち上げてみる
zooKeeperを立ち上げる
~ ❯❯❯ zkServer start
JMX enabled by default
Using config: /usr/local/etc/zookeeper/zoo.cfg
Starting zookeeper ... STARTED
kafkaを立ち上げる
~ ❯❯❯ kafka-server-start.sh /usr/local/etc/kafka/server.properties
起動ログは後半省略
[2015-12-05 12:13:28,666] INFO Verifying properties (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,701] INFO Property broker.id is overridden to 0 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,701] INFO Property log.cleaner.enable is overridden to false (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,701] INFO Property log.dirs is overridden to /usr/local/var/lib/kafka-logs (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,702] INFO Property log.retention.check.interval.ms is overridden to 300000 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,702] INFO Property log.retention.hours is overridden to 168 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,702] INFO Property log.segment.bytes is overridden to 1073741824 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,702] INFO Property num.io.threads is overridden to 8 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,702] INFO Property num.network.threads is overridden to 3 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,702] INFO Property num.partitions is overridden to 1 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,703] INFO Property num.recovery.threads.per.data.dir is overridden to 1 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,703] INFO Property port is overridden to 9092 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,703] INFO Property socket.receive.buffer.bytes is overridden to 102400 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,703] INFO Property socket.request.max.bytes is overridden to 104857600 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,703] INFO Property socket.send.buffer.bytes is overridden to 102400 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,703] INFO Property zookeeper.connect is overridden to localhost:2181 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,703] INFO Property zookeeper.connection.timeout.ms is overridden to 6000 (kafka.utils.VerifiableProperties)
[2015-12-05 12:13:28,740] INFO [Kafka Server 0], starting (kafka.server.KafkaServer)
[2015-12-05 12:13:28,743] INFO [Kafka Server 0], Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer)
[2015-12-05 12:13:28,753] INFO Starting ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2015-12-05 12:13:28,761] INFO Client environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT (org.apache.zookeeper.ZooKeeper)
[2015-12-05 12:13:28,761] INFO Client environment:host.name=softbank126097236203.bbtec.net (org.apache.zookeeper.ZooKeeper)
[2015-12-05 12:13:28,761] INFO Client environment:java.version=1.8.0_66 (org.apache.zookeeper.ZooKeeper)
..........
SpringBootを立ち上げる
手順は前と同じで、SpringBootのApplicationクラスを右クリックして「Run XXXX」を選択する。
立ち上がる。
が、このままだと本当にメッセージが書き込まれているかわからない。
kafkaのコマンドでtopicの存在を確認する。
できている。
~ ❯❯❯ /usr/local/bin/kafka-topics.sh --list --zookeeper localhost:2181
spring.cloud.stream.test.topic
~ ❯❯❯ /usr/local/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic spring.cloud.stream.test.topic
Topic:spring.cloud.stream.test.topic PartitionCount:1 ReplicationFactor:1 Configs:
Topic: spring.cloud.stream.test.topic Partition: 0 Leader: 0 Replicas: 0 Isr: 0
kafkaのメッセージを受け取ってログに出す
kafkaのメッセージを読み込むクラスを作成する
Sourceと同じパッケージに、SampleSinkという名前で以下を作成
package com.example;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
@EnableBinding(Sink.class)
public class SampleSink {
private static Logger logger = LoggerFactory.getLogger(SampleSink.class);
@ServiceActivator(inputChannel = Sink.INPUT)
public void sink(Object payload) {
logger.info("Received: " + payload);
}
}
application.ymlにinputの接続設定を追加
spring:
cloud:
stream:
bindings:
output: spring.cloud.stream.test.topic
input: spring.cloud.stream.test.topic
SpringBootのアプリケーションを再度立ち上げる
約1秒ごとにメッセージが書き込まれて、SampleSinkクラスによってログに出力されているのがわかる。
