使用 Kafka 和 MongoDB 进行 Go 异步处理
Kafka 异步
![](/data/attachment/album/201808/17/003015f11bgzwcfcbmb1i9.png)


在我前面的博客文章 “[我的第一个 Go 微服务：使用 MongoDB 和 Docker 多阶段构建](https://www.melvinvivas.com/my-first-go-microservice/)” 中，我创建了一个 Go 微服务示例，它发布一个 REST 式的 http 端点，并将从 HTTP POST 中接收到的数据保存到 MongoDB 数据库。


在这个示例中，我将数据的保存和 MongoDB 分离，并创建另一个微服务去处理它。我还添加了 Kafka 为消息层服务，这样微服务就可以异步处理它自己关心的东西了。



> 
> 如果你有时间去看，我将这个博客文章的整个过程录制到 [这个视频中了](https://youtu.be/xa0Yia1jdu8) :)
> 
> 
> 


下面是这个使用了两个微服务的简单的异步处理示例的上层架构图。


![rest-kafka-mongo-microservice-draw-io](/data/attachment/album/201808/17/003027x0suw8drsu8mefoa.jpg)


微服务 1 —— 是一个 REST 式微服务，它从一个 /POST http 调用中接收数据。接收到请求之后，它从 http 请求中检索数据，并将它保存到 Kafka。保存之后，它通过 /POST 发送相同的数据去响应调用者。


微服务 2 —— 是一个订阅了 Kafka 中的一个主题的微服务，微服务 1 的数据保存在该主题。一旦消息被微服务消费之后，它接着保存数据到 MongoDB 中。


在你继续之前，我们需要能够去运行这些微服务的几件东西：


1. [下载 Kafka](https://kafka.apache.org/downloads) —— 我使用的版本是 kafka\_2.11-1.1.0
2. 安装 [librdkafka](https://github.com/confluentinc/confluent-kafka-go) —— 不幸的是，这个库应该在目标系统中
3. 安装 [Kafka Go 客户端](https://github.com/confluentinc/confluent-kafka-go)
4. 运行 MongoDB。你可以去看我的 [以前的文章](https://www.melvinvivas.com/my-first-go-microservice/) 中关于这一块的内容，那篇文章中我使用了一个 MongoDB docker 镜像。


我们开始吧！


首先，启动 Kafka，在你运行 Kafka 服务器之前，你需要运行 Zookeeper。下面是示例：



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

```

接着运行 Kafka —— 我使用 9092 端口连接到 Kafka。如果你需要改变端口，只需要在 `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 的相关代码。我只是修改了我前面的示例去保存到 Kafka 而不是 MongoDB：


[rest-to-kafka/rest-kafka-sample.go](https://github.com/donvito/learngo/tree/master/rest-kafka-mongo-microservice/rest-to-kafka)



```
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](https://github.com/donvito/learngo/tree/master/rest-kafka-mongo-microservice/kafka-to-mongo)



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

}

```

我们来演示一下，运行微服务 1。确保 Kafka 已经运行了。



```
$ go run rest-kafka-sample.go

```

我使用 Postman 向微服务 1 发送数据。


![Screenshot-2018-04-29-22.20.33](/data/attachment/album/201808/17/003028mn1rtexr0rwer1z1.png)


这里是日志，你可以在微服务 1 中看到。当你看到这些的时候，说明已经接收到了来自 Postman 发送的数据，并且已经保存到了 Kafka。


![Screenshot-2018-04-29-22.22.00](/data/attachment/album/201808/17/003029yl56iivizi5v5toy.png)


因为我们尚未运行微服务 2，数据被微服务 1 只保存在了 Kafka。我们来消费它并通过运行的微服务 2 来将它保存到 MongoDB。



```
$ go run kafka-mongo-sample.go

```

现在，你将在微服务 2 上看到消费的数据，并将它保存到了 MongoDB。


![Screenshot-2018-04-29-22.24.15](/data/attachment/album/201808/17/003030irn5ddzkk2rd05d1.png)


检查一下数据是否保存到了 MongoDB。如果有数据，我们成功了！


![Screenshot-2018-04-29-22.26.39](/data/attachment/album/201808/17/003030xuajm6iijt9jtuja.png)


完整的源代码可以在这里找到：


<https://github.com/donvito/learngo/tree/master/rest-kafka-mongo-microservice>


现在是广告时间：如果你喜欢这篇文章，请在 Twitter [@donvito](https://twitter.com/donvito) 上关注我。我的 Twitter 上有关于 Docker、Kubernetes、GoLang、Cloud、DevOps、Agile 和 Startups 的内容。欢迎你们在 [GitHub](https://github.com/donvito) 和 [LinkedIn](https://www.linkedin.com/in/melvinvivas/) 关注我。


开心地玩吧！




---


via: <https://www.melvinvivas.com/developing-microservices-using-kafka-and-mongodb/>


作者：[Melvin Vivas](https://www.melvinvivas.com/author/melvin/)  译者：[qhwdw](https://github.com/qhwdw) 校对：[wxy](https://github.com/wxy)


本文由 [LCTT](https://github.com/LCTT/TranslateProject) 原创编译，[Linux中国](https://linux.cn/) 荣誉推出


OK
In my previous blog post [“My First Go Microservice using MongoDB and Docker Multi-Stage Builds”](https://www.melvinvivas.com/my-first-go-microservice/), I created a Go microservice sample which exposes a REST http endpoint and saves the data received from an HTTP POST to a MongoDB database.

In this example, I decoupled the saving of data to MongoDB and created another microservice to handle this. I also added Kafka to serve as the messaging layer so the microservices can work on its own concerns asynchronously.

In case you have time to watch, I recorded a walkthrough of this blog post in the

[video below]:)

Here is the high-level architecture of this simple asynchronous processing example wtih 2 microservices.
[
](/static/bd535ae9e3842814f6c0ccaafb88e9b0/a8e08/rest-kafka-mongo-microservice-draw-io.jpg)

![rest-kafka-mongo-microservice-draw-io rest-kafka-mongo-microservice-draw-io](/static/bd535ae9e3842814f6c0ccaafb88e9b0/a8e08/rest-kafka-mongo-microservice-draw-io.jpg)

**Microservice 1** - is a REST microservice which receives data from a /POST http call to it. After receiving the request, it retrieves the data from the http request and saves it to Kafka. After saving, it responds to the caller with the same data sent via /POST

**Microservice 2** - is a microservice which subscribes to a topic in Kafka where Microservice 1 saves the data. Once a message is consumed by the microservice, it then saves the data to MongoDB.

Before you proceed, we need a few things to be able to run these microservices:

[Download Kafka](https://kafka.apache.org/downloads)- I used version kafka_2.11-1.1.0- Install
[librdkafka](https://github.com/confluentinc/confluent-kafka-go)- Unfortunately, this library should be present in the target system - Install the
[Kafka Go Client by Confluent](https://github.com/confluentinc/confluent-kafka-go) - Run MongoDB. You can check my
[previous blog post](https://www.melvinvivas.com/my-first-go-microservice/)about this where I used a MongoDB docker image.

Let’s get rolling!

Start Kafka first, you need Zookeeper running before you run the Kafka server. Here’s how

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


Then run Kafka - I am using port 9092 to connect to Kafka. If you need to change the port, just configure it in config/server.properties. If you are just a beginner like me, I suggest to just use default ports for now.

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


After running Kafka, we need MongoDB. To make it simple, just use this docker-compose.yml.

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


Run the MongoDB docker container using Docker Compose

`docker-compose up`


Here is the relevant code of **Microservice 1**. I just modified my previous example to save to Kafka rather than MongoDB.

[rest-to-kafka/rest-kafka-sample.go](https://github.com/donvito/learngo/tree/master/rest-kafka-mongo-microservice/rest-to-kafka)

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


Here is the code of **Microservice 2**. What is important in this code is the consumption from Kafka, the saving part I already discussed in my previous blog post. Here are the important parts of the code which consumes the data from Kafka.

[kafka-to-mongo/kafka-mongo-sample.go](https://github.com/donvito/learngo/tree/master/rest-kafka-mongo-microservice/kafka-to-mongo)

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


Let’s get down to the demo, run Microservice 1. Make sure Kafka is running.

`$ go run rest-kafka-sample.go`


I used Postman to send data to Microservice 1
[
](/static/cbb6202a5abeeb670a91673453ab471b/0e149/Screenshot-2018-04-29-22.20.33.png)

![Screenshot-2018-04-29-22.20.33 Screenshot-2018-04-29-22.20.33](/static/cbb6202a5abeeb670a91673453ab471b/d9199/Screenshot-2018-04-29-22.20.33.png)

Here is the log you will see in Microservice 1. Once you see this, it means data has been received from Postman and saved to Kafka
[
](/static/8d909cadc3de3cff336987f583dfdf4d/c27e7/Screenshot-2018-04-29-22.22.00.png)

![Screenshot-2018-04-29-22.22.00 Screenshot-2018-04-29-22.22.00](/static/8d909cadc3de3cff336987f583dfdf4d/d9199/Screenshot-2018-04-29-22.22.00.png)

Since we are not running Microservice 2 yet, the data saved by Microservice 1 will just be in Kafka. Let’s consume it and save to MongoDB by running Microservice 2.

`$ go run kafka-mongo-sample.go`


Now you’ll see that Microservice 2 consumes the data and saves it to MongoDB
[
](/static/69d094e6602bc3c2b10be0226d2b4e68/ba168/Screenshot-2018-04-29-22.24.15.png)

![Screenshot-2018-04-29-22.24.15 Screenshot-2018-04-29-22.24.15](/static/69d094e6602bc3c2b10be0226d2b4e68/d9199/Screenshot-2018-04-29-22.24.15.png)

Check if data is saved in MongoDB. If it is there, we’re good!
[
](/static/032f2d2015adefcbdbf1bb5598b90d96/b77b1/Screenshot-2018-04-29-22.26.39.png)

![Screenshot-2018-04-29-22.26.39 Screenshot-2018-04-29-22.26.39](/static/032f2d2015adefcbdbf1bb5598b90d96/d9199/Screenshot-2018-04-29-22.26.39.png)

Complete source code can be found here
[https://github.com/donvito/learngo/tree/master/rest-kafka-mongo-microservice](https://github.com/donvito/learngo/tree/master/rest-kafka-mongo-microservice)

Shameless plug! If you like this blog post, please follow me in Twitter ** @donvito**. I tweet about Docker, Kubernetes, GoLang, Cloud, DevOps, Agile and Startups. Would love to connect in

**and**

[GitHub](https://github.com/donvito)[LinkedIn](https://www.linkedin.com/in/melvinvivas/)

Enjoy!