利用京东云Serverless服务快速构建5G时代的IoT应用

Posted 2021-03-16 jdclouddeveloper

10月31日，在2019年中国国际信息通信展览会上，工信部宣布：5G商用正式启动。5G商用时代来了！

5G的商用，使得数据传输速度、响应速度、连接数据、数据传输量、传输可靠性等方面都有了显著的提升，这一技术的突破才使得很多领域的应用场景得以真正的落地实施，走进普通人的生活，而这其中就包括物联网。

虽然物联网的概念很简明，就是把物品与互联网相连接并进行信息交换和通讯，从而实现万物智能化，但是由于各种原因，商业化的应用并没有大规模的落地，其中一方面是由于网络传输等原因的限制。如今，5G的到来使得物联网应用在网络层面不再受限，更好地促进了物联网的发展。对于企业来说，快速构建一个商用的物联网服务，抢占先机就显得尤为重要。

京东云Serverless服务正适应了如今企业的这种需求。Serverless一系列产品生态可以让用户以业务为导向，无需关心底层服务器部署以及承载能力，实施周期短，无预付价格按使用量付费。这些特性是适应互联网5G快速时代、快速构建的不二选择，为企业节省成本的同时解决了技术上的难题。

以下面一个车联网数据为例子，我们来看一下如何用队列服务来构建一个高可用高可靠的无服务应用。在车联网应用中客户端负载可能会在24小时内扩展/缩减3、4个数量级，这些特性天然适合Serverless服务动态扩缩，按量付费。

案例场景：车联网中搭载数据收集传感器的汽车向云端传输行驶数据，云端利用队列服务削峰填谷、动态扩缩的能力接收数据，然后转存到Elasticsearch进行更好的数据检索和应用，为用户提供更好的服务或者为公司提供更多的商业价值。
需求：队列服务SDK，队列服务接入点地址，Elasticsearch接入点地址（已经创建好ES实例），京东云用户AK/SK
代码语言: Go

Step1

创建队列服务客户端以及资源创建

 func CreateClient() *sqs.SQS {
     ses, _ := session.NewSession(&aws.Config{
         Region: aws.String("cn-north-1"),
         Credentials: credentials.NewStaticCredentials("your AccessKey",
             "your SecretKey", ""),
         Endpoint:   aws.String("http://jqs.cn-north-1.jdcloud.com"),
         DisableSSL: aws.Bool(true),
     })
     _, err := ses.Config.Credentials.Get()
    if err != nil {
        log.Fatal("凭据创建失败", err)
    }
    client := sqs.New(ses)
    return client
}

func CreateQueueTask(name string) string {
    resp, err := sqsClient.CreateQueue(&sqs.CreateQueueInput{
        QueueName: aws.String(name),
    })
    if err != nil {
        log.Println("Create Queue Failed:", err)
        return ""
    }
    return *resp.QueueUrl
}

Step2

每个车辆设备发送消息

 func SendTask(url string, message interface{}) {
    body, _ := json.Marshal(message)
     _, err := sqsClient.SendMessage(&sqs.SendMessageInput{
         MessageBody: aws.String(string(body)),
         QueueUrl:    aws.String(url),
     })
     if err != nil {
         log.Println("Send Message Failed:", err)
         return
    }
    return
}

测试数据：

测试机器的配置：CPU64、内存256G、带宽100Mbps（京东云云主机）

场景：公网；send-单条（JQS）

Step3

从队列服务中拉取消息转存到Elasticsearch中

func ReceiveMessageTask(url string) interface{} {
     result, err := sqsClient.ReceiveMessage(&sqs.ReceiveMessageInput{
         AttributeNames:        aws.StringSlice([]string{"All"}),
         MaxNumberOfMessages:   aws.Int64(1),
         MessageAttributeNames: aws.StringSlice([]string{"All"}),
         QueueUrl:              aws.String(url),
         VisibilityTimeout:     aws.Int64(30),
         WaitTimeSeconds:       aws.Int64(0),
     })
    log.Println(*result.Messages[0].Body)
    if err != nil {
        log.Println("Receive Message Failed:", err)
        return ""
    }

    if len(result.Messages) > 0 {
        _, delErr := sqsClient.DeleteMessage(&sqs.DeleteMessageInput{
            QueueUrl:      aws.String(url),
            ReceiptHandle: result.Messages[0].ReceiptHandle,
        })
        if err != nil {
            log.Println("Delete Message Failed:", delErr)
        }

        message := new(gpsMessage)
        Err := json.Unmarshal([]byte(*result.Messages[0].Body), message)
        if Err != nil {
            log.Println("Receive Message Unmarshal Failed", Err)
            return ""
        }
        return message
    }
    return ""
}

func Build(url string, method string, body interface{}) []byte {
    client := &http.Client{}
    //向服务端发送get请求
    bodyData, _ := json.Marshal(body)
    request, _ := http.NewRequest(method, url, bytes.NewReader(bodyData))

    request.Header.Set("Accept", "application/json")
    request.Header.Set("Content-Type", "application/json")
    //接收服务端返回给客户端的信息
    response, _ := client.Do(request)
    r, _ := ioutil.ReadAll(response.Body)
    return r
}

func PostMessageToES(p string, body interface{}) string {
    postReturn := new(postRes)
    postResponse := Build(p, "POST", body)
    err := json.Unmarshal(postResponse, postReturn)
    if err != nil {
        log.Println("Unmarshal Failed", err)
    }
    jsonS, _ := json.Marshal(postReturn)
    log.Println("postResult:", string(jsonS))
    return postReturn.Id
}

Step4

从Elasticsearch按照需求索引搜索信息

func GetMessageFromES(p string) {
     message := new(esMessage)
     getResponse := Build(p, "GET", nil)
     log.Println("getPath:", p)
     Err := json.Unmarshal(getResponse, message)
     if Err != nil {
         log.Println("Unmarshal Failed", Err)
     }
     jsonM, _ := json.Marshal(message)
    log.Println("getResult:", string(jsonM))
}

结果示例：

可以按照车辆信息进行数据的检索，绘制出汽车动态地图，利用这些数据来帮助自动驾驶、动态导航、车辆健康度分析等多种场景进行实现。

京东云的队列服务作为Serverless开发中的BaaS服务，实现了中间件服务的无运维和毫秒级扩缩能力，支持京东云的合作伙伴零成本启动业务和按使用量付费的模式，帮助用户解决资源扩缩和阈值监控等复杂问题。结合函数服务FaaS使用，可以满足更丰富的场景，并且调用整个京东云Serverless生态，打造基于云原生21世纪的开放式的全新应用。点击“了解”，快来进行体验吧！

欢迎点击“京东云”了解更多精彩内容