Kafka と MongoDB を使用した非同期処理

[[240575]]

前回のブログ投稿「初めての Go マイクロサービス: MongoDB と Docker を使用したマルチステージビルド」では、RESTful http エンドポイントを公開し、HTTP POST から受信したデータを MongoDB データベースに保存するサンプルの Go マイクロサービスを作成しました。

この例では、データストレージを MongoDB から分離し、それを処理するための別のマイクロサービスを作成しました。また、マイクロサービスが独自の懸念事項を非同期的に処理できるように、メッセージングレイヤーとして Kafka を追加しました。

このブログ投稿の全プロセスをこのビデオに録画しましたので、時間があればご覧ください :)

以下は、2 つのマイクロサービスを使用したこの単純な非同期処理の例の高レベルアーキテクチャ図です。

REST-Kafka-Mongo-マイクロサービス-Draw-IO

マイクロサービス 1 - /POST http 呼び出しからデータを受信する RESTful マイクロサービスです。リクエストを受信すると、http リクエストからデータを取得して Kafka に保存します。保存後、同じデータを /POST 経由で呼び出し元に送り返します。

マイクロサービス 2 - マイクロサービス 1 のデータが保存されている Kafka のトピックをサブスクライブするマイクロサービスです。メッセージがマイクロサービスによって消費されると、そのデータは MongoDB に保存されます。

先に進む前に、これらのマイクロサービスを実行するためにいくつかのものが必要です。

Kafkaをダウンロード - 私はバージョンkafka_2.11-1.1.0を使用しました
librdkafkaをインストールします - 残念ながら、このライブラリはターゲットシステム上に存在している必要があります
Kafka Goクライアントをインストールする
MongoDB を実行します。これについては、MongoDB Docker イメージを使用した以前の記事をお読みください。

さあ始めましょう！

まず、Kafka を起動します。 Kafka サーバーを実行する前に、Zookeeper を実行する必要があります。次に例を示します。

 $ cd /< download path >/ kafka_2 . 11 - 1.1 . 0
 $ bin / zookeeper - server - start . sh config / zookeeper . properties

次に、Kafka を実行します。Kafka への接続にはポート 9092 を使用します。ポートを変更する必要がある場合は、 config/server.propertiesで設定するだけです。私のような初心者の場合は、今のところデフォルトのポートを使用することをお勧めします。

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

Kafka を実行したら、MongoDB が必要になります。とても簡単です。このdocker-compose.ymlを使用するだけです。

 version : '3'
 services :
 mongodb :
 image : mongo
 ports :
      - "27017:27017"
 volumes :
      - "mongodata:/data/db"
 networks :
      - network1
 volumes :
 mongodata :
 networks :
 network1 :

Docker Compose を使用して MongoDB Docker コンテナを実行します。

 docker - compose up

以下はマイクロサービス 1 に関連するコードです。以前の例を変更して、MongoDB ではなく Kafka に保存するようにしました。

レストからカフカへ/レストカフカサンプル.go

 func jobsPostHandler ( w http . ResponseWriter , r * http . Request ) {
    //Retrieve body from http request
 b , err := ioutil . ReadAll ( r . Body )
 defer r . Body . Close ()
    if err != nil {
 panic ( err )
    }
    //Save data into Job struct
    var _job Job
 err = json . Unmarshal ( b , & _job )
    if err != nil {
 http . Error ( w , err . Error (), 500 )
        return
    }
 saveJobToKafka ( _job )
    //Convert job struct into json
 jsonString , err := json . Marshal ( _job )
    if err != nil {
 http . Error ( w , err . Error (), 500 )
        return
    }
    //Set content-type http header
    w . Header (). Set ( "content-type" , "application/json" )
    //Send back data as response
    w . Write ( jsonString )
 }
 func saveJobToKafka ( job Job ) {
 fmt . Println ( "save to kafka" )
 jsonString , err := json . Marshal ( job )
 jobString := string ( jsonString )
 fmt . Print ( jobString )
 p , err := kafka . NewProducer (& kafka . ConfigMap { "bootstrap.servers" : "localhost:9092" })
    if err != nil {
 panic ( err )
    }
    // Produce messages to topic (asynchronously)
 topic := "jobs-topic1"
    for _ , word := range [] string { string ( jobString )} {
 p . Produce (& kafka . Message {
            TopicPartition : kafka . TopicPartition { Topic : & topic , Partition : kafka . PartitionAny },
            Value :          [] byte ( word ),
        }, nil )
    }
 }

以下はマイクロサービス 2 のコードです。このコードで最も重要なことは、Kafka からデータを消費することです。保存部分については、前回のブログ投稿ですでに説明しました。ここでのコードの重要な部分は、Kafka からのデータの使用です。

kafka-to-mongo/kafka-mongo-sample.go

 func main () {
    //Create MongoDB session
 session := initialiseMongo ()
 mongoStore . session = session
 receiveFromKafka ()
 }
 func receiveFromKafka () {
 fmt . Println ( "Start receiving from Kafka" )
 c , err := kafka . NewConsumer (& kafka . ConfigMap {
        "bootstrap.servers" : "localhost:9092" ,
        "group.id" :          "group-id-1" ,
        "auto.offset.reset" : "earliest" ,
    })
    if err != nil {
 panic ( err )
    }
 c . SubscribeTopics ([] string { "jobs-topic1" }, nil )
    for {
 msg , err := c . ReadMessage (- 1 )
        if err == nil {
 fmt . Printf ( "Received from Kafka %s: %s\n" , msg . TopicPartition , string ( msg . Value ))
 job := string ( msg . Value )
 saveJobToMongo ( job )
        } else {
 fmt . Printf ( "Consumer error: %v (%v)\n" , err , msg )
            break
        }
    }
 c . Close ()
 }
 func saveJobToMongo ( jobString string ) {
 fmt . Println ( "Save to MongoDB" )
 col := mongoStore . session . DB ( database ). C ( collection )
    //Save data into Job struct
    var _job Job
 b := [] byte ( jobString )
 err := json . Unmarshal ( b , & _job )
    if err != nil {
 panic ( err )
    }
    //Insert job into MongoDB
 errMongo := col . Insert ( _job )
    if errMongo != nil {
 panic ( errMongo )
    }
 fmt . Printf ( "Saved to MongoDB : %s" , jobString )
 }