光伏电站建设的行业网站,做个外贸网站设计,建设网站装配式建筑楼房,华为网站建设目标简介#xff1a;Serverless 架构的按量付费模式#xff0c;可以在保证在线编程功能性能的前提下#xff0c;进一步降低成本。本文将会以阿里云函数计算为例#xff0c;通过 Serverless 架构实现一个 Python 语言的在线编程功能#xff0c;并对该功能进一步的优化#xff…简介Serverless 架构的按量付费模式可以在保证在线编程功能性能的前提下进一步降低成本。本文将会以阿里云函数计算为例通过 Serverless 架构实现一个 Python 语言的在线编程功能并对该功能进一步的优化使其更加贴近本地本地代码执行体验。
随着计算机科学与技术的发展越来越多的人开始接触编程也有越来越多的在线编程平台诞生。以 Python 语言的在线编程平台为例大致可以分为两类:
一类是 OJ 类型的即在线评测的编程平台这类的平台特点是阻塞类型的执行即用户需要一次性将代码和标准输入内容提交当程序执行完成会一次性将结果返回另一类则是学习、工具类的在线编程平台例如 Anycodes 在线编程等网站这一类平台的特点是非阻塞类型的执行即用户可以实时看到代码执行的结果以及可以实时内容进行内容的输入。
但是无论是那种类型的在线编程平台其背后的核心模块 “代码执行器”或“判题机”都是极具有研究价值一方面这类网站通常情况下都需要比要严格的“安全机制”例如程序会不会有恶意代码出现死循环、破坏计算机系统等程序是否需要隔离运行运行时是否会获取到其他人提交的代码等
另一方面这类平台通常情况下都会对资源消耗比较大尤其是比赛来临时更是需要突然间对相关机器进行扩容必要时需要大规模集群来进行应对。同时这类网站通常情况下也都有一个比较大的特点那就是触发式即每个代码执行前后实际上并没有非常紧密的前后文关系等。
随着 Serverless 架构的不断发展很多人发现 Serverless 架构的请求级隔离和极致弹性等特性可以解决传统在线编程平台所遇到的安全问题和资源消耗问题Serverless 架构的按量付费模式可以在保证在线编程功能性能的前提下进一步降低成本。所以通过 Serverless 架构实现在线编程功能的开发就逐渐的被更多人所关注和研究。本文将会以阿里云函数计算为例通过 Serverless 架构实现一个 Python 语言的在线编程功能并对该功能进一步的优化使其更加贴近本地本地代码执行体验。
在线编程功能开发
一个比较简单的、典型的在线编程功能在线执行模块通常情况下是需要以下几个能力
在线执行代码用户可以输入内容可以返回结果标准输出、标准错误等
除了在线编程所需要实现的功能之外在线编程在 Serverless 架构下所需要实现的业务逻辑也仅仅被收敛到关注代码执行模块即可获取客户端发送的程序信息包括代码、标准输入等将代码缓存到本地执行代码获取结果但会给客户端整个架构的流程简图为 关于执行代码部分可以通过 Python 语言的 subprocess 依赖中的 Popen() 方法实现在使用 Popen() 方法时有几个比较重要的概念需要明确
subprocess.PIPE一个可以被用于 Popen 的stdin 、stdout 和 stderr 3 个参数的特殊值表示需要创建一个新的管道subprocess.STDOUT一个可以被用于 Popen 的 stderr 参数的输出值表示子程序的标准错误汇合到标准输出
所以当我们想要实现可以:
进行标准输入stdin获取标准输出stdout以及标准错误stderr的功能
可以简化代码实现为 除代码执行部分之外在 Serverless 架构下获取到用户代码并将其存储过程中需要额外注意函数实例中目录的读写权限。通常情况下在函数计算中如果不进行硬盘挂载只有/tmp/目录是有可写入权限的。所以在该项目中我们将用户传递到服务端的代码进行临时存储时需要将其写入临时目录/tmp/在临时存储代码的时候还需要额外考虑实例复用的情况所以此时可以为临时代码提供临时的文件名例如 # -*- coding: utf-8 -*- import randomrandom Str lambda num5: .join(random.sample(abcdefghijklmnopqrstuvwxyz, num)) path /tmp/%s% randomStr(5) 完整的代码实现为 # -*- coding: utf-8 -*- import json import uuid import random import subprocess # 随机字符串 randomStr lambda num5: .join(random.sample(abcdefghijklmnopqrstuvwxyz, num)) # Response class Response: def __init__(self, start_response, response, errorCodeNone): self.start start_response responseBody { Error: {Code: errorCode, Message: response}, } if errorCode else { Response: response } # 默认增加uuid便于后期定位 responseBody[ResponseId] str(uuid.uuid1()) self.response json.dumps(responseBody) def __iter__(self): status 200 response_headers [(Content-type, application/json; charsetUTF-8)] self.start(status, response_headers) yield self.response.encode(utf-8) def WriteCode(code, fileName): try: with open(fileName, w) as f: f.write(code) return True except Exception as e: print(e) return False def RunCode(fileName, input_data): child subprocess.Popen(python %s % (fileName), stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.STDOUT, shellTrue) output child.communicate(inputinput_data.encode(utf-8)) return output[0].decode(utf-8) def handler(environ, start_response): try: request_body_size int(environ.get(CONTENT_LENGTH, 0)) except (ValueError): request_body_size 0 requestBody json.loads(environ[wsgi.input].read(request_body_size).decode(utf-8)) code requestBody.get(code, None) inputData requestBody.get(input, ) fileName /tmp/ randomStr(5) responseData RunCode(fileName, inputData) if code and WriteCode(code, fileName) else Error return Response(start_response, {result: responseData}) 完成核心的业务逻辑编写之后我们可以将代码部署到阿里云函数计算中。部署完成之后我们可以获得到接口的临时测试地址。通过 PostMan 对该接口进行测试以 Python 语言的输出语句为例 print(HELLO WORLD) 可以看到当我们通过 POST 方法携带代码等作为参数发起请求后获得到的响应为 我们通过响应结果可以看到系统是可以正常输出我们的预期结果“HELLO WORLD” 至此我们完成了标准输出功能的测试接下来我们对标准错误等功能进行测试此时我们将刚刚的输出代码进行破坏 print(HELLO WORLD) 使用同样的方法再次进行代码执行可以看到结果 结果中我们可以看到 Python 的报错信息是符合我们的预期的至此完成了在线编程功能的标准错误功能的测试接下来我们进行标准输入功能的测试由于我们使用的 subprocess.Popen() 方法是一种阻塞方法所以此时我们需要将代码和标准输入内容一同放到服务端。测试的代码为 tempInput input(please input: ) print(Output: , tempInput) 测试的标准输入内容为“serverless devs”。
当我们使用同样的方法发起请求之后我们可以看到 系统是正常输出预期的结果。至此我们完成了一个非常简单的在线编程服务的接口。该接口目前只是初级版本仅用于学习使用其具有极大的优化空间
超时时间的处理代码执行完成可以进行清理
当然通过这个接口也可以看到这样一个问题那就是代码执行过程中是阻塞的我们没办法进行持续性的输入也没有办法实时输出即使需要输入内容也是需要将代码和输入内容一并发送到服务端。这种模式和目前市面上常见的 OJ 模式很类似但是就单纯的在线编程而言还需要进一步对项目优化使其可以通过非阻塞方法实现代码的执行并且可以持续性的进行输入操作持续性的进行内容输出。
更贴近“本地”的代码执行器
我们以一段代码为例 import time print(hello world) time.sleep(10) tempInput input(please: ) print(Input data: , tempInput) 当我们在本地的执行这段 Python 代码时整体的用户侧的实际表现是
系统输出 hello world系统等待 10 秒系统提醒我们 please我们此时可以输入一个字符串系统输出 Input data 以及我们刚刚输入的字符串
但是这段代码如果应用于传统 OJ 或者刚刚我们所实现的在线编程系统中表现则大不相同
代码与我们要输入内容一同传给系统系统等待 10 秒输出 hello world、please以及最后输 Input data 和我们输入的内容
可以看到OJ 模式上的在线编程功能和本地是有非常大的差距的至少在体验层面这个差距是比较大的。为了减少这种体验不统一的问题我们可以将上上述的架构进一步升级通过函数的异步触发以及 Python 语言的 pexpect.spawn() 方法实现一款更贴近本地体验的在线编程功能 在整个项目中包括了两个函数两个存储桶
业务逻辑函数该函数的主要操作是业务逻辑包括创建代码执行的任务通过对象存储触发器进行异步函数执行以及获取函数输出结果以及对任务函数的标准输入进行相关操作等执行器函数该函数的主要作用是执行用户的函数代码这部分是通过对象存储触发通过下载代码、执行代码、获取输入、输出结果等代码获取从代码存储桶输出结果和获取输入从业务存储桶代码存储桶该存储桶的作用是存储代码当用户发起运行代码的请求 业务逻辑函数收到用户代码后会将代码存储到该存储桶再由该存储桶处罚异步任务业务存储桶该存储桶的作用是中间量的输出主要包括输出内容的缓存、输入内容的缓存该部分数据可以通过对象存储的本身特性进行生命周期的制定
为了让代码在线执行起来更加贴近本地体验该方案的代码分为两个函数分别进行业务逻辑处理和在线编程核心功能。
其中业务逻辑处理函数主要是
获取用户的代码信息生成代码执行 ID并将代码存到对象存储异步触发在线编程函数的执行返回生成代码执行 ID获取用户的输入信息和代码执行 ID并将内容存储到对应的对象存储中获取代码的输出结果根据用户指定的代码执行 ID将执行结果从对象存储中读取出来并返回给用户
整体的业务逻辑为 实现的代码为 # -*- coding: utf-8 -*- import os import oss2 import json import uuid import random # 基本配置信息 AccessKey { id: os.environ.get(AccessKeyId), secret: os.environ.get(AccessKeySecret) } OSSCodeConf { endPoint: os.environ.get(OSSConfEndPoint), bucketName: os.environ.get(OSSConfBucketCodeName), objectSignUrlTimeOut: int(os.environ.get(OSSConfObjectSignUrlTimeOut)) } OSSTargetConf { endPoint: os.environ.get(OSSConfEndPoint), bucketName: os.environ.get(OSSConfBucketTargetName), objectSignUrlTimeOut: int(os.environ.get(OSSConfObjectSignUrlTimeOut)) } # 获取获取/上传文件到OSS的临时地址 auth oss2.Auth(AccessKey[id], AccessKey[secret]) codeBucket oss2.Bucket(auth, OSSCodeConf[endPoint], OSSCodeConf[bucketName]) targetBucket oss2.Bucket(auth, OSSTargetConf[endPoint], OSSTargetConf[bucketName]) # 随机字符串 randomStr lambda num5: .join(random.sample(abcdefghijklmnopqrstuvwxyz, num)) # Response class Response: def __init__(self, start_response, response, errorCodeNone): self.start start_response responseBody { Error: {Code: errorCode, Message: response}, } if errorCode else { Response: response } # 默认增加uuid便于后期定位 responseBody[ResponseId] str(uuid.uuid1()) self.response json.dumps(responseBody) def __iter__(self): status 200 response_headers [(Content-type, application/json; charsetUTF-8)] self.start(status, response_headers) yield self.response.encode(utf-8) def handler(environ, start_response): try: request_body_size int(environ.get(CONTENT_LENGTH, 0)) except (ValueError): request_body_size 0 requestBody json.loads(environ[wsgi.input].read(request_body_size).decode(utf-8)) reqType requestBody.get(type, None) if reqType run: # 运行代码 code requestBody.get(code, None) runId randomStr(10) codeBucket.put_object(runId, code.encode(utf-8)) responseData runId elif reqType input: # 输入内容 inputData requestBody.get(input, None) runId requestBody.get(id, None) targetBucket.put_object(runId -input, inputData.encode(utf-8)) responseData ok elif reqType output: # 获取结果 runId requestBody.get(id, None) targetBucket.get_object_to_file(runId -output, /tmp/ runId) with open(/tmp/ runId) as f: responseData f.read() else: responseData Error return Response(start_response, {result: responseData}) 执行器函数主要是通过代码存储桶触发从而进行代码执行的模块这一部分主要包括
从存储桶获取代码并通过 pexpect.spawn() 进行代码执行通过 pexpect.spawn().read_nonblocking() 非阻塞的获取间断性的执行结果并写入到对象存储通过 pexpect.spawn().sendline() 进行内容输入
整体流程为 代码实现为 # -*- coding: utf-8 -*- import os import re import oss2 import json import time import pexpect # 基本配置信息 AccessKey { id: os.environ.get(AccessKeyId), secret: os.environ.get(AccessKeySecret) } OSSCodeConf { endPoint: os.environ.get(OSSConfEndPoint), bucketName: os.environ.get(OSSConfBucketCodeName), objectSignUrlTimeOut: int(os.environ.get(OSSConfObjectSignUrlTimeOut)) } OSSTargetConf { endPoint: os.environ.get(OSSConfEndPoint), bucketName: os.environ.get(OSSConfBucketTargetName), objectSignUrlTimeOut: int(os.environ.get(OSSConfObjectSignUrlTimeOut)) } # 获取获取/上传文件到OSS的临时地址 auth oss2.Auth(AccessKey[id], AccessKey[secret]) codeBucket oss2.Bucket(auth, OSSCodeConf[endPoint], OSSCodeConf[bucketName]) targetBucket oss2.Bucket(auth, OSSTargetConf[endPoint], OSSTargetConf[bucketName]) def handler(event, context): event json.loads(event.decode(utf-8)) for eveEvent in event[events]: # 获取object code eveEvent[oss][object][key] localFileName /tmp/ event[events][0][oss][object][eTag] # 下载代码 codeBucket.get_object_to_file(code, localFileName) # 执行代码 foo pexpect.spawn(python %s % localFileName) outputData startTime time.time() # timeout可以通过文件名来进行识别 try: timeout int(re.findall(timeout(.*?)s, code)[0]) except: timeout 60 while (time.time() - startTime) / 1000 timeout: try: tempOutput foo.read_nonblocking(size999999, timeout0.01) tempOutput tempOutput.decode(utf-8, ignore) if len(str(tempOutput)) 0: outputData outputData tempOutput # 输出数据存入oss targetBucket.put_object(code -output, outputData.encode(utf-8)) except Exception as e: print(Error: , e) # 有输入请求被阻塞 if str(e) Timeout exceeded.: try: # 从oss读取数据 targetBucket.get_object_to_file(code -input, localFileName -input) targetBucket.delete_object(code -input) with open(localFileName -input) as f: inputData f.read() if inputData: foo.sendline(inputData) except: pass # 程序执行完成输出 elif End Of File (EOF) in str(e): targetBucket.put_object(code -output, outputData.encode(utf-8)) return True # 程序抛出异常 else: outputData outputData \n\nException: %s % str(e) targetBucket.put_object(code -output, outputData.encode(utf-8)) return False 当我们完成核心的业务逻辑编写之后我们可以将项目部署到线上。
项目部署完成之后和上文的测试方法一样在这里也通过 PostMan 对接口进行测试。此时我们需要设定一个覆盖能较全的测试代码包括输出打印、输入、一些 sleep() 等方法 当我们通过 PostMan 发起请求执行这段代码之后我们可以看到系统为我们返回了预期的代码执行 ID 我们可以看到系统会返回给我们一个代码执行 ID该执行 ID 将会作为我们整个请求任务的 ID此时我们可以通过获取输出结果的接口来获取结果 由于代码中有 time.sleep(10) 所以迅速获得结果的时候是看不到后半部分的输出结果我们可以设置一个轮训任务不断通过该 ID 对接口进行刷新 可以看到10 秒钟后代码执行到了输入部分 tempInput input(please: ) 此时我们再通过输入接口进行输入操作 完成之后我们可以看到输入成功result: ok的结果此时我们继续刷新之前获取结果部分的请求 可以看到我们已经获得到了所有结果的输出。
相对于上文的在线编程功能这种“更贴近本地的代码执行器“变得复杂了很多但是在实际使用的过程中却可以更好的模拟出本地执行代码时的一些现象例如代码的休眠、阻塞、内容的输出等。
总结
无论是简单的在线代码执行器部分还是更贴近“本地”的代码执行器部分这篇文章在所应用的内容是相对广泛的。通过这篇文章你可以看到
HTTP 触发器的基本使用方法对象存储触发器的基本使用方函数计算组件、对象存储组件的基本使用方法组件间依赖的实现方法
同时通过这篇文章也可以从一个侧面看到这样一个常见问题的简单解答我有一个项目我是每个接口一个函数还是多个接口复用一个函数
针对这个问题其实最主要的是看业务本身的诉求如果多个接口表达的含义是一致的或者是同类的类似的并且多个接口的资源消耗是类似的那么放在一个函数中来通过不同的路径进行区分是完全可以的如果出现资源消耗差距较大或者函数类型、规模、类别区别过大的时候将多个接口放在多个函数下也是没有问题的。
本文实际上是抛砖引玉无论是 OJ 系统的“判题机”部分还是在线编程工具的“执行器部分”都可以很好的和 Serverless 架构有着比较有趣的结合点。这种结合点不仅仅可以解决传统在线编程所头疼的事情安全问题资源消耗问题并发问题流量不稳定问题更可以将 Serverless 的价值在一个新的领域发挥出来。
原文链接 本文为阿里云原创内容未经允许不得转载。
