Streamがあります。眺めていてだいぶ時間が経っているのですが、そろそろ試してみようかなということで。Streamとは?マイクロサービス間のメッセージングを実現するプロダクトです。#ccc_ab3ドキュメントは、こちら。Guideあるアプリケーション(Source)からメッセージをBinder(キュー)を介して、別のアプリケーション(Sink)側で受け取る仕組みをSpringを使ったアプリケーションとして組み上げます。Streamの用語でメッセージの送り元をSource、受け取り側をSinkと呼ぶようですが、間に変換処理などを行うようなProcessor(Source+Sink)も導入可能なようです。KafkaとRabbitMQを選択できます。Spring側でいくつかすでに実装済みのSource、Sinkもあります。自作(標準出力)という形で組んでみます。twitterstreamを使っていることから、SourceでTwitterからStreamingで受け取ったツイートを、Sink側で標準出力で出力するという内容でいってみます。今回は、Mavenのマルチプロジェクトとして作ってみることにします。親pomの依存関係とプラグインの設定。Startersのtwitterstreamを追加。Bootはこちらに合わせました。Kafkaについては、単に起動しておけばOKですが、今回はリモート接続を行うようにしているので、listenersだけリモート接続可能にしておきました。ZooKeeperも、同じホストで動かすことにします。Kafkaが動作しているホストのIPアドレスは、172.17.0.2とします。Topicの作成は不要です。ZooKeeper)は起動済みとします。Source(twitterstream)を作るそれでは、Sourceを作成してみましょう。親pomを継承している前提で、アプリケーションの作成に必要な依存関係とプラグインの設定は以下のとおり。Kafkaを使うので以下が必要になります。Bootについては、明示的に依存関係に含める必要がありました。アプリケーションの実装はこんな感じに。tweetstreamを使うには、TwitterstreamSourceConfigurationを@ImportしておけばOKな模様。@Import(TwitterstreamSourceConfiguration.class)あとは、設定です。今回は、このような感じに。Kafkaに登録する時のデータフォーマットです。今回は、JSONで保存するようにしました。指定しなかった場合は、今回の場合StringがSinkに飛んできました...。ZooKeeperへの接続設定です。spring.cloud.stream.kafka.binder.brokers=172.17.0.2spring.cloud.stream.kafka.binder.defaultBrokerPort=9092spring.cloud.stream.kafka.binder.zkNodes=172.17.0.2spring.cloud.stream.kafka.binder.defaultZkPort=2181今回はIPアドレスとデフォルトポートという形で記述しましたが、brokersやzkNodesに「host:port」の形式で書くのもOKのようです。ドキュメント上の該当個所は、こちら。APIを使用するためのOAuth系の情報が必要です。ドキュメントは、こちらを参考に。Sourceあとはパッケージングして起動。Bootアプリケーションが起動します。ZooKeeperへ接続できれば、起動に成功します。起動してこういうログが出ればOKでしょう。Kafkaに投入し続けます。不要な時は停止しておきましょう。Sink(標準出力)を作る続いて、Sinkを作成します。Maven依存関係と、プラグインの設定はこんな感じで。</build>アプリケーションは、こんな感じで作成。}}クラスの宣言には、@SpringBootApplication以外に@EnableBinding(Sink.class)を付与。Kafka側にJSONで入っているメッセージは、@StreamListener(Sink.INPUT)を付与したメソッドでMapとして受け取れます。JSONで受けているデータのマッピング先をMapにはしょっただけですが...。{出力内容としては、ツイートのidや日時、名前、ツイート内容を出力しておしまいです。このアプリケーションに対する設定は、こちら。src/main/resources/application.propertiesspring.cloud.stream.bindings.input.destination=tweetspring.cloud.stream.bindings.input.group=group1spring.cloud.stream.kafka.binder.brokers=172.17.0.2spring.cloud.stream.kafka.binder.defaultBrokerPort=9092spring.cloud.stream.kafka.binder.zkNodes=172.17.0.2spring.cloud.stream.kafka.binder.defaultZkPort=2181server.port=9080「spring.cloud.stream.bindings.〜」の部分が「input」になります。Sourceの時は、「output」でした。KafkaのTopicを指すようになるので、Sourceと同じ値を指定します。今回の場合は「tweet」になります。「spring.cloud.stream.bindings.input.group」については、複数のSinkを使用する場合を考えると指定しておいた方が無難なようです。ZooKeeperに関する設定は、Sourceの時と同じです。あと、本筋とは直接関係はありませんが、今回SourceとSinkを同じホスト上で動かしているので、Tomcatのポートはずらしています。server.port=9080今回は、Sourceが8080、Sinkが9080を使用しています。パッケージングして起動。今回のアプリケーションの名前は、「console-sink」にしています。console-sink/target/console-sink-0.0.1-SNAPSHOT.jar起動に成功すれば、Sourceが取り込んだツイートがコンソールに出力されます。APIの結果をそのまま受け取るので、特に絞ったりすることができません。Developers実装としてはfirehoseかsampleを使うようになっています。https://github.com/spring-cloud/spring-cloud-stream-app-starters/blob/v1.0.4.RELEASE/twitter/spring-cloud-starter-stream-source-twitterstream/src/main/java/org/springframework/cloud/stream/app/twitterstream/source/TwitterStreamMessageProducer.java全パラメーターを指定できるわけでもないようですし(TODOになっています)。いうわけで、ちょっと自分で書いてみます。}}元ネタは、こちらですよ。https://github.com/spring-cloud/spring-cloud-stream-app-starters/blob/v1.0.4.RELEASE/twitter/spring-cloud-starter-stream-source-twitterstream/src/main/java/org/springframework/cloud/stream/app/twitterstream/source/TwitterStreamMessageProducer.java今回は、「#tweettest」をフィルタの対象にしています。}パッケージングして、確認。tweet-source/target/tweet-source-0.0.1-SNAPSHOT.jar結果(Sink側)。Source起動後に、「#tweettest」タグをつけて自分でツイートしました。それが拾われています。#tweettest出力されました、と。Streamを試してみました。取っ掛かりで最初はオロオロしましたが、ある程度書いてみたりドキュメント読んだり、GitHubリポジトリ見てるとなんとなく慣れそうな感じがしてきました。今後、もうちょっといろいろ試していきましょう。読者になる検索リンクはてなブログブログをはじめる週刊はてなブログはてなブログPro月別アーカイブはてなブログをはじめよう!Kazuhiraさんは、はてなブログを使っています。ブログを報告するスターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。