189 8069 5689

使用Kafka和MongoDB进行Go异步处理

使用Kafka和MongoDB进行GO异步处理

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

开发

后端

其他数据库

Kafka

MongoDB 在我前面的博客文章 “我的第一个 Go 微服务：使用 MongoDB 和 Docker 多阶段构建” 中，我创建了一个 Go 微服务示例，它发布一个 REST 式的 http 端点，并将从 HTTP POST 中接收到的数据保存到 MongoDB 数据库。

在我前面的博客文章 “我的***个 Go 微服务：使用 MongoDB 和 Docker 多阶段构建” 中，我创建了一个 Go 微服务示例，它发布一个 REST 式的 http 端点，并将从 HTTP POST 中接收到的数据保存到 MongoDB 数据库。

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

如果你有时间去看，我将这个博客文章的整个过程录制到 这个视频中了 ????

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

rest-kafka-mongo-microservice-draw-io

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

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

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

  1. 下载 Kafka —— 我使用的版本是 kafka_2.11-1.1.0
  2. 安装 librdkafka —— 不幸的是，这个库应该在目标系统中
  3. 安装 Kafka Go 客户端
  4. 运行 MongoDB。你可以去看我的 以前的文章 中关于这一块的内容，那篇文章中我使用了一个 MongoDB docker 镜像。

我们开始吧！

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

  
 
 
 
       
  
  
  
  1. $ cd //kafka_2.11-1.1.0
    2.    
  
  
  
  2. $ bin/zookeeper-server-start.sh config/zookeeper.properties
    3.

接着运行 Kafka —— 我使用 9092 端口连接到 Kafka。如果你需要改变端口，只需要在 config/server.properties 中配置即可。如果你像我一样是个新手，我建议你现在还是使用默认端口。

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

Kafka 跑起来之后，我们需要 MongoDB。它很简单，只需要使用这个 docker-compose.yml 即可。

  
 
 
 
       
  
  
  
  1. version: '3'
    2.    
  
  
  
  2. services:
    3.    
  
  
  
  3.  mongodb:
    4.    
  
  
  
  4.  image: mongo
    5.    
  
  
  
  5.  ports:
    6.    
  
  
  
  6.  - "27017:27017"
    7.    
  
  
  
  7.  volumes:
    8.    
  
  
  
  8.  - "mongodata:/data/db"
    9.    
  
  
  
  9.  networks:
    10.    
  
  
  
  10.  - network1
    11.    
  
  
  
  11.  
    12.    
  
  
  
  12. volumes:
    13.    
  
  
  
  13.  mongodata:
    14.    
  
  
  
  14.  
    15.    
  
  
  
  15. networks:
    16.    
  
  
  
  16.  network1:
    17.

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

  
 
 
 
       
  
  
  
  1. docker-compose up
    2.

这里是微服务 1 的相关代码。我只是修改了我前面的示例去保存到 Kafka 而不是 MongoDB：

rest-to-kafka/rest-kafka-sample.go

  
 
 
 
       
  
  
  
  1. func jobsPostHandler(w http.ResponseWriter, r *http.Request) {
    2.    
  
  
  
  2.  
    3.    
  
  
  
  3.  //Retrieve body from http request
    4.    
  
  
  
  4.  b, err := ioutil.ReadAll(r.Body)
    5.    
  
  
  
  5.  defer r.Body.Close()
    6.    
  
  
  
  6.  if err != nil {
    7.    
  
  
  
  7.  panic(err)
    8.    
  
  
  
  8.  }
    9.    
  
  
  
  9.  
    10.    
  
  
  
  10.  //Save data into Job struct
    11.    
  
  
  
  11.  var _job Job
    12.    
  
  
  
  12.  err = json.Unmarshal(b, &_job)
    13.    
  
  
  
  13.  if err != nil {
    14.    
  
  
  
  14.  http.Error(w, err.Error(), 500)
    15.    
  
  
  
  15.  return
    16.    
  
  
  
  16.  }
    17.    
  
  
  
  17.  
    18.    
  
  
  
  18.  saveJobToKafka(_job)
    19.    
  
  
  
  19.  
    20.    
  
  
  
  20.  //Convert job struct into json
    21.    
  
  
  
  21.  jsonString, err := json.Marshal(_job)
    22.    
  
  
  
  22.  if err != nil {
    23.    
  
  
  
  23.  http.Error(w, err.Error(), 500)
    24.    
  
  
  
  24.  return
    25.    
  
  
  
  25.  }
    26.    
  
  
  
  26.  
    27.    
  
  
  
  27.  //Set content-type http header
    28.    
  
  
  
  28.  w.Header().Set("content-type", "application/json")
    29.    
  
  
  
  29.  
    30.    
  
  
  
  30.  //Send back data as response
    31.    
  
  
  
  31.  w.Write(jsonString)
    32.    
  
  
  
  32.  
    33.    
  
  
  
  33. }
    34.    
  
  
  
  34.  
    35.    
  
  
  
  35. func saveJobToKafka(job Job) {
    36.    
  
  
  
  36.  
    37.    
  
  
  
  37.  fmt.Println("save to kafka")
    38.    
  
  
  
  38.  
    39.    
  
  
  
  39.  jsonString, err := json.Marshal(job)
    40.    
  
  
  
  40.  
    41.    
  
  
  
  41.  jobString := string(jsonString)
    42.    
  
  
  
  42.  fmt.Print(jobString)
    43.    
  
  
  
  43.  
    44.    
  
  
  
  44.  p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost:9092"})
    45.    
  
  
  
  45.  if err != nil {
    46.    
  
  
  
  46.  panic(err)
    47.    
  
  
  
  47.  }
    48.    
  
  
  
  48.  
    49.    
  
  
  
  49.  // Produce messages to topic (asynchronously)
    50.    
  
  
  
  50.  topic := "jobs-topic1"
    51.    
  
  
  
  51.  for _, word := range []string{string(jobString)} {
    52.    
  
  
  
  52.  p.Produce(&kafka.Message{
    53.    
  
  
  
  53.  TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
    54.    
  
  
  
  54.  Value: []byte(word),
    55.    
  
  
  
  55.  }, nil)
    56.    
  
  
  
  56.  }
    57.    
  
  
  
  57. }
    58.

这里是微服务 2 的代码。在这个代码中最重要的东西是从 Kafka 中消费数据，保存部分我已经在前面的博客文章中讨论过了。这里代码的重点部分是从 Kafka 中消费数据：

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

  
 
 
 
       
  
  
  
  1. func main() {
    2.    
  
  
  
  2.  
    3.    
  
  
  
  3.  //Create MongoDB session
    4.    
  
  
  
  4.  session := initialiseMongo()
    5.    
  
  
  
  5.  mongoStore.session = session
    6.    
  
  
  
  6.  
    7.    
  
  
  
  7.  receiveFromKafka()
    8.    
  
  
  
  8.  
    9.    
  
  
  
  9. }
    10.    
  
  
  
  10.  
    11.    
  
  
  
  11. func receiveFromKafka() {
    12.    
  
  
  
  12.  
    13.    
  
  
  
  13.  fmt.Println("Start receiving from Kafka")
    14.    
  
  
  
  14.  c, err := kafka.NewConsumer(&kafka.ConfigMap{
    15.    
  
  
  
  15.  "bootstrap.servers": "localhost:9092",
    16.    
  
  
  
  16.  "group.id": "group-id-1",
    17.    
  
  
  
  17.  "auto.offset.reset": "earliest",
    18.    
  
  
  
  18.  })
    19.    
  
  
  
  19.  
    20.    
  
  
  
  20.  if err != nil {
    21.    
  
  
  
  21.  panic(err)
    22.    
  
  
  
  22.  }
    23.    
  
  
  
  23.  
    24.    
  
  
  
  24.  c.SubscribeTopics([]string{"jobs-topic1"}, nil)
    25.    
  
  
  
  25.  
    26.    
  
  
  
  26.  for {
    27.    
  
  
  
  27.  msg, err := c.ReadMessage(-1)
    28.    
  
  
  
  28.  
    29.    
  
  
  
  29.  if err == nil {
    30.    
  
  
  
  30.  fmt.Printf("Received from Kafka %s: %s\n", msg.TopicPartition, string(msg.Value))
    31.    
  
  
  
  31.  job := string(msg.Value)
    32.    
  
  
  
  32.  saveJobToMongo(job)
    33.    
  
  
  
  33.  } else {
    34.    
  
  
  
  34.  fmt.Printf("Consumer error: %v (%v)\n", err, msg)
    35.    
  
  
  
  35.  break
    36.    
  
  
  
  36.  }
    37.    
  
  
  
  37.  }
    38.    
  
  
  
  38.  
    39.    
  
  
  
  39.  c.Close()
    40.    
  
  
  
  40.  
    41.    
  
  
  
  41. }
    42.    
  
  
  
  42.  
    43.    
  
  
  
  43. func saveJobToMongo(jobString string) {
    44.    
  
  
  
  44.  
    45.    
  
  
  
  45.  fmt.Println("Save to MongoDB")
    46.    
  
  
  
  46.  col := mongoStore.session.DB(database).C(collection)
    47.    
  
  
  
  47.  
    48.    
  
  
  
  48.  //Save data into Job struct
    49.    
  
  
  
  49.  var _job Job
    50.    
  
  
  
  50.  b := []byte(jobString)
    51.    
  
  
  
  51.  err := json.Unmarshal(b, &_job)
    52.    
  
  
  
  52.  if err != nil {
    53.    
  
  
  
  53.  panic(err)
    54.    
  
  
  
  54.  }
    55.    
  
  
  
  55.  
    56.    
  
  
  
  56.  //Insert job into MongoDB
    57.    
  
  
  
  57.  errMongo := col.Insert(_job)
    58.    
  
  
  
  58.  if errMongo != nil {
    59.    
  
  
  
  59.  panic(errMongo)
    60.    
  
  
  
  60.  }
    61.    
  
  
  
  61.  
    62.    
  
  
  
  62.  fmt.Printf("Saved to MongoDB : %s", jobString)
    63.    
  
  
  
  63.  
    64.    
  
  
  
  64. }
    65.

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

  
 
 
 
       
  
  
  
  1. $ go run rest-kafka-sample.go
    2.

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

Screenshot-2018-04-29-22.20.33

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

Screenshot-2018-04-29-22.22.00

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

  
 
 
 
       
  
  
  
  1. $ go run kafka-mongo-sample.go
    2.

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

Screenshot-2018-04-29-22.24.15

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

Screenshot-2018-04-29-22.26.39

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

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


当前标题：使用Kafka和MongoDB进行Go异步处理
分享地址：http://cdxtjz.cn/article/cohodsc.html

其他资讯