使用Kafka和MongoDB進(jìn)行Go異步處理

作者：Melvin Vivas 2018-08-19 09:15:25

在我前面的博客文章 “我的第一個(gè) Go 微服務(wù)：使用 MongoDB 和 Docker 多階段構(gòu)建” 中，我創(chuàng)建了一個(gè) Go 微服務(wù)示例，它發(fā)布一個(gè) REST 式的 http 端點(diǎn)，并將從 HTTP POST 中接收到的數(shù)據(jù)保存到 MongoDB 數(shù)據(jù)庫。

[[240575]]

在我前面的博客文章 “我的***個(gè) Go 微服務(wù)：使用 MongoDB 和 Docker 多階段構(gòu)建” 中，我創(chuàng)建了一個(gè) Go 微服務(wù)示例，它發(fā)布一個(gè) REST 式的 http 端點(diǎn)，并將從 HTTP POST 中接收到的數(shù)據(jù)保存到 MongoDB 數(shù)據(jù)庫。

在這個(gè)示例中，我將數(shù)據(jù)的保存和 MongoDB 分離，并創(chuàng)建另一個(gè)微服務(wù)去處理它。我還添加了 Kafka 為消息層服務(wù)，這樣微服務(wù)就可以異步處理它自己關(guān)心的東西了。

如果你有時(shí)間去看，我將這個(gè)博客文章的整個(gè)過程錄制到這個(gè)視頻中了 :)

下面是這個(gè)使用了兩個(gè)微服務(wù)的簡單的異步處理示例的上層架構(gòu)圖。

rest-kafka-mongo-microservice-draw-io

rest-kafka-mongo-microservice-draw-io

微服務(wù) 1 —— 是一個(gè) REST 式微服務(wù)，它從一個(gè) /POST http 調(diào)用中接收數(shù)據(jù)。接收到請(qǐng)求之后，它從 http 請(qǐng)求中檢索數(shù)據(jù)，并將它保存到 Kafka。保存之后，它通過 /POST 發(fā)送相同的數(shù)據(jù)去響應(yīng)調(diào)用者。

微服務(wù) 2 —— 是一個(gè)訂閱了 Kafka 中的一個(gè)主題的微服務(wù)，微服務(wù) 1 的數(shù)據(jù)保存在該主題。一旦消息被微服務(wù)消費(fèi)之后，它接著保存數(shù)據(jù)到 MongoDB 中。

在你繼續(xù)之前，我們需要能夠去運(yùn)行這些微服務(wù)的幾件東西：

下載 Kafka —— 我使用的版本是 kafka_2.11-1.1.0
安裝 librdkafka —— 不幸的是，這個(gè)庫應(yīng)該在目標(biāo)系統(tǒng)中
安裝 Kafka Go 客戶端
運(yùn)行 MongoDB。你可以去看我的以前的文章中關(guān)于這一塊的內(nèi)容，那篇文章中我使用了一個(gè) MongoDB docker 鏡像。

我們開始吧！

首先，啟動(dòng) Kafka，在你運(yùn)行 Kafka 服務(wù)器之前，你需要運(yùn)行 Zookeeper。下面是示例：

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

接著運(yùn)行 Kafka —— 我使用 9092 端口連接到 Kafka。如果你需要改變端口，只需要在 config/server.properties 中配置即可。如果你像我一樣是個(gè)新手，我建議你現(xiàn)在還是使用默認(rèn)端口。

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

Kafka 跑起來之后，我們需要 MongoDB。它很簡單，只需要使用這個(gè) docker-compose.yml 即可。

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

使用 Docker Compose 去運(yùn)行 MongoDB docker 容器。

docker-compose up

這里是微服務(wù) 1 的相關(guān)代碼。我只是修改了我前面的示例去保存到 Kafka 而不是 MongoDB：

rest-to-kafka/rest-kafka-sample.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)
    }
}

這里是微服務(wù) 2 的代碼。在這個(gè)代碼中最重要的東西是從 Kafka 中消費(fèi)數(shù)據(jù)，保存部分我已經(jīng)在前面的博客文章中討論過了。這里代碼的重點(diǎn)部分是從 Kafka 中消費(fèi)數(shù)據(jù)：

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)
 
}