数据传输格式

虽然使用 AMQP 协议（高级消息队列协议）的异步通信和 gRPC 通信使用二进制协议进行数据传输，但 REST-API 通常以文本格式传输数据。与基于文本的协议相比，二进制协议的效率要高得多 [1,2]。因此，使用 gRPC 和 AMQP 进行通信会导致较低的网络负载，而使用 REST API 时可以预期更高的网络负载。

连接处理

REST-API 通常建立在 HTTP/1.1 协议之上，而 gRPC 依赖于 HTTP/2 协议的使用。 HTTP/1.1、HTTP/2 以及 AMQP 都在传输层使用 TCP 来确保稳定的连接。要建立这样的连接，需要在客户端和服务器之间进行详细的通信。这些性能影响同样适用于所有沟通方式。但是，对于 AMQP 或 HTTP/2 连接，通信连接的初始建立只需要执行一次，因为这两种协议的请求都可以多路复用。这意味着可以将现有连接重用于使用异步或 gRPC 通信的后续请求。另一方面，使用 HTTP/1.1 的 REST-API 为与远程服务器的每个请求建立新连接。

Necessary communication to establish a TCP-Connection

消息序列化

通常，在通过网络传输消息之前，使用 JSON 执行 REST 和异步通信以进行消息序列化。另一方面，gRPC 默认以协议缓冲区格式传输数据。协议缓冲区通过允许使用更高级的序列化和反序列化方法来编码和使用消息内容 [1] 来提高通信速度。然而，选择正确的消息序列化格式取决于工程师。关于性能，protocol buffers 有很多优势，但是当必须调试微服务之间的通信时，依赖人类可读的 JSON 格式可能是更好的选择。

缓存

有效的缓存策略可以显着减少服务器的负载和必要的计算资源。由于其架构，REST-API 是唯一允许有效缓存的通信方式。 REST-API 响应可以被其他服务器和缓存代理（如 Varnish）缓存和复制。这减少了 REST 服务的负载并允许处理大量的 HTTP 流量 [1]。但是，这只有在基础架构上部署更多服务（缓存代理）或使用第三方集成后才有可能。 gRPC 官方文档和 RabbitMQ 文档都没有介绍任何形式的缓存。

负载均衡

除了临时存储响应之外，还有其他技术可以提高服务速度。负载均衡器（例如 mod_proxy）可以高效透明的方式在服务之间分配 HTTP 流量 [1]。这可以实现使用 REST API 的服务的水平扩展。 Kubernetes 作为容器编排解决方案，无需任何调整即可对 HTTP/1.1 流量进行负载均衡。另一方面，对于 gRPC，需要在网络上提供另一个服务（linkerd）[3]。异步通信无需进一步的帮助即可支持负载平衡。消息代理本身扮演负载均衡器的角色，因为它能够将请求分发到同一服务的多个实例。消息代理为此目的进行了优化，并且它们的设计已经考虑到它们必须具有特别可扩展性的事实[1]。

实验

为了能够评估各个通信方法对软件质量特性的影响，开发了四个微服务来模拟电子商务平台的订单场景。

微服务部署在由三个不同服务器组成的自托管 Kubernetes 集群上。服务器通过千兆 (1000 Mbit/s) 网络连接，位于同一数据中心，服务器之间的平均延迟为 0.15 毫秒。每次实验运行时，各个服务都部署在相同的服务器上。这种行为是通过 pod 亲和性来实现的。

所有微服务都是用 GO 编程语言实现的。个别服务的实际业务逻辑，例如与数据库的通信，为了不被选择的通信方法之外的其他影响，故意不实现。因此，收集的结果不能代表这种类型的微服务架构，但可以使实验中的通信方法具有可比性。相反，业务逻辑的实现是通过将程序流程延迟 100 毫秒来模拟的。因此，在通信中，总延迟为 400 毫秒。

开源软件k6用于实现负载测试。

实现

Golang 标准库中包含的 net/http 模块用于提供 REST 接口。使用标准库中也包含的 encoding/json 模块对请求进行序列化和反序列化。所有请求都使用 HTTP POST 方法。

“谈话很便宜。给我看看密码。”

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io/ioutil"
    "log"
    "net/http"

    "github.com/google/uuid"
    "gitlab.com/timbastin/bachelorarbeit/common"
    "gitlab.com/timbastin/bachelorarbeit/config"
)

type restServer struct {
    httpClient http.Client
}

func (server *restServer) handler(res http.ResponseWriter, req *http.Request) {
    // only allow post request.
    if req.Method != http.MethodPost {
        bytes, _ := json.Marshal(map[string]string{
            "error": "invalid request method",
        })
        http.Error(res, string(bytes), http.StatusBadRequest)
        return
    }

    reqId := uuid.NewString()

    // STEP 1 / 4
    log.Println("(REST) received new order", reqId)

    var submitOrderDTO common.SubmitOrderRequestDTO

    b, _ := ioutil.ReadAll(req.Body)

    err := json.Unmarshal(b, &submitOrderDTO)
    if err != nil {
        log.Fatalf(err.Error())
    }

    checkIfInStock(1)

    invoiceRequest, _ := http.NewRequest(http.MethodPost, 
    fmt.Sprintf("%s/invoices", config.MustGet("customerservice.rest.address").
     (string)), bytes.NewReader(b))
    // STEP 2
    r, err := server.httpClient.Do(invoiceRequest)
    // just close the response body
    r.Body.Close()
    if err != nil {
        panic(err)
    }

    shippingRequest, _ := http.NewRequest(http.MethodPost, 
    fmt.Sprintf("%s/shipping-jobs", config.MustGet("shippingservice.rest.address").
     (string)), bytes.NewReader(b))

    // STEP 3
    r, err = server.httpClient.Do(shippingRequest)
    // just close the response body
    r.Body.Close()
    if err != nil {
        panic(err)
    }

    handleProductDecrement(1)
    // STEP 5
    res.WriteHeader(201)
    res.Write(common.NewJsonResponse(map[string]string{
        "state": "success",
    }))
}

func startRestServer() {
    server := restServer{
        httpClient: http.Client{},
    }
    http.HandleFunc("/orders", server.handler)
    done := make(chan int)
    go http.ListenAndServe(config.MustGet("orderservice.rest.port").(string), nil)
    log.Println("started rest server")
    <-done
}

RabbitMQ 消息代理用于异步通信，部署在同一个 Kubernetes 集群上。消息代理和各个微服务之间的通信使用 github.com/spreadway/amqp 库进行。该库是 GO 编程语言官方文档推荐的。

package main

import (
    "encoding/json"
    "log"

    "github.com/streadway/amqp"
    "gitlab.com/timbastin/bachelorarbeit/common"
    "gitlab.com/timbastin/bachelorarbeit/config"
    "gitlab.com/timbastin/bachelorarbeit/utils"
)

func handleMsg(message amqp.Delivery, ch *amqp.Channel) {
    log.Println("(AMQP) received new order")
    var submitOrderRequest common.SubmitOrderRequestDTO
    err := json.Unmarshal(message.Body, &submitOrderRequest)
    utils.FailOnError(err, "could not unmarshal message")

    checkIfInStock(1)

    handleProductDecrement(1)
    ch.Publish(config.MustGet("amqp.billingRequestExchangeName").(string), "", 
     false, false, amqp.Publishing{
        ContentType: "application/json",
        Body:        message.Body,
    })

}

func getNewOrderChannel(conn *amqp.Connection) (*amqp.Channel, string) {
    ch, err := conn.Channel()
    utils.FailOnError(err, "could not create channel")

    ch.ExchangeDeclare(config.MustGet("amqp.newOrderExchangeName").
    (string), "fanout", false, false, false, false, nil)

    queue, err := ch.QueueDeclare(config.MustGet("orderservice.amqp.consumerName").
    (string), false, false, false, false, nil)

    utils.FailOnError(err, "could not create queue")

    ch.QueueBind(queue.Name, "", config.MustGet("amqp.newOrderExchangeName").
    (string), false, nil)
    return ch, queue.Name
}

func startAmqpServer() {
    conn := common.NewAmqpConnection(config.MustGet("amqp.host").(string))
    defer conn.Close()

    orderChannel, queueName := getNewOrderChannel(conn)

    msgs, err := orderChannel.Consume(
        queueName,
        config.MustGet("orderservice.amqp.consumerName").(string),
        true,
        false,
        false,
        false,
        nil,
    )

    utils.FailOnError(err, "could not consume")

    forever := make(chan bool)
    log.Println("started amqp server:", queueName)
    go func() {
        for d := range msgs {
            go handleMsg(d, orderChannel)
        }
    }()
    <-forever
}

gRPC 客户端和服务器使用 gRPC 文档推荐的 google.golang.org/grpc 库。数据的序列化是使用协议缓冲区完成的。

package main

import (
    "log"
    "net"

    "context"

    "gitlab.com/timbastin/bachelorarbeit/common"
    "gitlab.com/timbastin/bachelorarbeit/config"
    "gitlab.com/timbastin/bachelorarbeit/pb"
    "gitlab.com/timbastin/bachelorarbeit/utils"
    "google.golang.org/grpc"
)

type OrderServiceServer struct {
    CustomerService pb.CustomerServiceClient
    ShippingService pb.ShippingServiceClient
    pb.UnimplementedOrderServiceServer
}

func (s *OrderServiceServer) SubmitOrder(ctx context.Context, 
    request *pb.SubmitOrderRequest) (*pb.SuccessReply, error) {
    log.Println("(GRPC) received new order")
    if s.CustomerService == nil {
        s.CustomerService, _ = common.NewCustomerServiceClient()
    }
    if s.ShippingService == nil {
        s.ShippingService, _ = common.NewShippingServiceClient()
    }

    checkIfInStock(1)

    // call the product service on each iteration to decrement the product.
    _, err := s.CustomerService.CreateAndProcessBilling(ctx, &pb.BillingRequest{
        BillingInformation: request.BillingInformation,
        Products:           request.Products,
    })

    utils.FailOnError(err, "could not process billing")

    // trigger the shipping job.
    _, err = s.ShippingService.CreateShippingJob(ctx, &pb.ShippingJob{
        BillingInformation: request.BillingInformation,
        Products:           request.Products,
    })

    utils.FailOnError(err, "could not create shipping job")

    handleProductDecrement(1)

    return &pb.SuccessReply{Success: true}, nil
}

func startGrpcServer() {
    listen, err := net.Listen("tcp", config.MustGet("orderservice.grpc.port").(string))
    if err != nil {
        log.Fatalf("could not listen: %v", err)
    }

    grpcServer := grpc.NewServer()

    orderService := OrderServiceServer{}
    // inject the clients into the server
    pb.RegisterOrderServiceServer(grpcServer, &orderService)

    // start the server
    log.Println("started grpc server")
    if err := grpcServer.Serve(listen); err != nil {
        log.Fatalf("could not start grpc server: %v", err)
    }
}

收集数据

检查成功和失败的订单处理的数量，以确认它们所经过的时间。如果直到确认的持续时间超过 900 毫秒，则订单流程被解释为失败。选择此持续时间是因为在实验中可能会出现无限长的等待时间，尤其是在使用异步通信时。每次试验都会报告失败和成功订单的数量。

每种架构总共进行了 12 次不同的测量，每种情况下同时请求的数量不同，传输的数据量也不同。首先，在低负载下测试每种通信方式，然后在中等负载下，最后在高负载下测试。低负载模拟 10 个，中等负载模拟 100 个，高负载模拟 300 个同时向系统发出的请求。在这六次测试运行之后，要传输的数据量会增加，以了解各个接口的序列化方法的效率。数据量的增加是通过订购多个产品来实现的。

结果

gRPC API 架构是实验中研究的性能最佳的通信方法。在低负载下，它可以接受的订单数量是使用 REST 接口的系统的 3.41 倍。此外，平均响应时间比 REST-API 低 9.71 毫秒，比 AMQP-API 低 9.37 毫秒。

1 simultaneous request, low bandwidth usage. x: Communication Method (Number of Requests), y: Duration in milliseconds

1 simultaneous request, high bandwidth usage. x: Communication Method (Number of Requests), y: Duration in milliseconds

总体而言，这种趋势在更多并发请求的实验中继续存在。对于 100 个并发请求，使用 gRPC-API 架构可以处理的订单数量是 REST-API 的 3.7 倍。与 AMQP 的差异要小得多。 GRPC 的处理能力比 AMQP 多 8.06%（1170 个订单）。虽然 gRPC 可以在 418.99 毫秒内处理 95% 的请求，但 AMQP 只能在 557.39 毫秒内完成，而 REST 在 1425.13 毫秒内完成。

100 simultaneous requests, low bandwidth usage. x: Communication Method (Number of Requests), y: Duration in milliseconds