使用 Kafka 和 MongoDB 進行 Go 非同步處理 | Linux 中國

使用 Docker Compose 去執行 MongoDB docker 容器。

docker-compose up

這裡是微服務 1 的相關程式碼。我只是修改了我前面的示例去儲存到 Kafka 而不是 MongoDB：

rest-to-kafka/rest-kafka-sample.go[5]

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

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

$ go run rest-kafka-sample.go

$ go run kafka-mongo-sample.go