原文转载自《刘悦的技术博客》https://v3u.cn/a_id_218分而治之算法是一种非常古老但实用的方法。初衷是将一个较大的整体分解成小的部分，让每个小部分都不足以对抗大的整体。战国时期，秦国以破坏纵横联盟为分治手段；19世纪，比利时殖民者占领卢旺达，将卢旺达种族分为胡图族和图西族，企图割据和控制。如果是这样。21世纪，人们倾向于在Leetcode平台上做分而治之的算法题，但实际上，从工业的角度来看，算法如果不结合实际的业务场景，算法永远都是虚无缥缈的，并且只会出现在开发过程中。在笔者的一次不经意采访中，真正的算法不是空洞的，它应该帮助我们解决实际问题，是的，它应该落地成为一个实体。分片上传大文件就是这样一个符合分而治之算法的场景。现在视频文件的体积越来越大，高清视频的体积在2-4g左右，但是4K视频的分辨率是标准高清的四倍。需要四倍的存储空间——只需两到三分钟的未压缩4K电影，或电影预告片的长度，就可以达到500GB。8K的视频文件大到难以想象，现在12K正在兴起。对于如此庞大的文件，如何设计合理的数据传输方案呢？这里我们以前后端分离项目为例。前端使用Vue.js3.0配合ui库Ant-desgin，后端使用并发异步框架Tornado实现大文件的分片非阻塞传输和异步IO写入服务。前端碎片化首先安装Vue3.0或更高版本：npminstall-g@vue/cli安装异步请求库axios：npminstallaxios--save然后安装Ant-desgin：npmi--saveant-design-vue@next-SAnt-desgin虽然因为圣诞“彩蛋门”事件臭名昭著，但客观来说，它仍然是业界不可多得的优秀UI框架之一。然后在项目程序入口文件中导入使用：import{createApp}from'vue'importAppfrom'./App.vue'import{router}from'./router/index'importaxiosfrom'axios'importqsfrom'qs'从'ant-design-vue'导入Antd；导入'ant-design-vue/dist/antd.css';constapp=createApp(App)app.config.globalProperties.axios=axios;app.config.globalProperties。upload_dir="https://localhost/static/";app.config.globalProperties.weburl="http://localhost:8000";应用程序使用（路由器）；app.use(Antd);app.mount('#app')然后参考Ant-desgin官方文档：https://antdv.com/components/...构建上传控件：上传文件注意绑定的before-upload需要强制返回false并设置为手动上传：beforeUpload:function(file){returnfalse;}然后声明分片方法：fileupload:function(file){varsize=file.file.size;//总大小varshardSize=200*1024;//分片大小this.shardCount=Math.ceil(size/shardSize);//切片总数console.日志（this.shardCount）；for(vari=0;i{this.finished+=1;console.log(this.finished);如果(this.finished==this.shardCount){this.mergeupload(file.file.name);}}).catch(function(err){//上传失败});}}具体的分片逻辑是，大文件的总体积是根据单个分片的体积除以大小并向上取整得到文件的分片数。这里为了测试方便，单片体积设置为200kb，可以随时修改。然后，使用Math.min方法计算分片过程中每个切片的最小值。开始和结束位置，然后通过slice方法进行切片操作，最后将下标、文件名、切片体异步发送到后台。当所有分片请求发送完毕，封装分片合并方法，请求后端发起合并分片操作：mergeupload:function(filename){this.myaxios(this.weburl+"/upload/","put",{"filename":filename}).then(data=>{console.log(data);});}至此，前端分片逻辑就完成了。后端异步IO写入为了避免同步写入造成的阻塞，安装aiofiles库：pip3installaiofilesaiofiles用于asyncio应用中处理本地磁盘文件。配合Tornado的异步非阻塞机制，有效提高文件写入效率:importaiofiles#切片上传类SliceUploadHandler(BaseHandler):asyncdefpost(self):file=self.request.files["file"][0]filename=self.get_argument("filename")count=self.get_argument("count")filename='%s_%s'%(filename,count)#构成片段的唯一标识符contents=file['body']#异步读取文件asyncwithaiofiles.open('./static/uploads/%s'%filename,"wb")asf:awaitf.write(contents)return{"filename":file.filename,"errcode":0}这里，后端获取分片实体和文件名，以及片段标识符，将片段文件以文件名\_片段标识符的格式异步写入系统目录，以一张378kb的png图片为例，片段文件应该依次为200kb和178kb，如图：所有切片文件写入成功后，出现切片合并界面被触发：importaiofiles#sliceuploadclassSliceUploadHandler(BaseHandler):asyncdefpost(self):file=self.request.files["file"][0]filename=self.get_argument("filename")count=self.get_argument("count")filename='%s_%s'%(filename,count)#构成该片段唯一一标识符contents=file['body']#异步读取文件asyncwithaiofiles.open('./static/uploads/%s'%filename,"wb")asf:awaitf.write(contents)return{"filename":file.filename,"errcode":0}asyncdefput(self):filename=self.get_argument("filename")chunk=0asyncwithaiofiles.open('./static/uploads/%s'%filename,'ab')astarget_file:whileTrue:try:source_file=open('./static/uploads/%s_%s'%(filename,chunk),'rb')awaittarget_file.write(source_file.read())source_file.close()除了Exceptionase:print(str(e))breakchunk=chunk+1self.finish({"msg":"ok","errcode":0})这里用文件名寻址，然后遍历合并。注意句柄写入方式是增量的，否则会逐层覆盖碎片文件。同时，它还具有断点续写的功能。有些逻辑会把分片的个数传给后端，让后台判断合并分片的个数。其实大可不必，因为如果寻址失败，会自动抛出异常，跳出循环，从而节省了一个参数的带宽占用。在真实的超大文件传输场景中，轮询服务可能会因为网络或其他因素中断分片任务。这时候就需要通过降级快速响应，返回后台数据，避免用户长时间等待。这里我们使用基于Tornado的Apscheduler库的Schedule分片任务：pipinstallapscheduler然后编写job.py轮询服务文件：fromdatetimeimportdatetimefromtornado.ioloopimportIOLoop,PeriodicCallbackfromtornado.webimportRequestHandler,Applicationfromapscheduler.schedulers。tornadoimportTornadoSchedulerscheduler=Nonejob_ids=[]#初始化definit_scheduler():globalschedulerscheduler=TornadoScheduler()scheduler.start()print('[SchedulerInit]APSchedulerhasbeenstarted')#要执行的定时任务是这里deftask1(options):print('{}[APScheduler][Task]-{}'.format(datetime.now().strftime('%Y-%m-%d%H:%M:%S.%f'),options))classMainHandler(RequestHandler):defget(self):self.write('添加作业
删除作业')classSchedulerHandler(RequestHandler):defget(self):globaljob_idsjob_id=self.get_query_argument('job_id',None)action=self.get_query_argument('action',None)ifjob_id:#添加if'add'==action:如果job_id不在job_ids中：job_ids.append(job_id)scheduler.add_job(task1,'interval',seconds=3,id=job_id,args=(job_id,))self.write('[TASKADDED]-{}'.format(job_id))else:self.write('[TASKEXISTS]-{}'.format(job_id))#removeelif'remove'==action:ifjob_idinjob_ids:scheduler.remove_job(job_id)job_ids.remove(job_id)self.write('[TASKREMOVED]-{}'.format(job_id))else:self.write('[TASKNOTFOUND]-{}'.format(job_id))else:self.write('[INVALIDPARAMS]INVALIDjob_idoraction')if__name__=="__main__":routes=[(r"/",MainHandler),(r"/scheduler/?",SchedulerHandler),]init_scheduler()app=Application(routes,debug=True)app.listen(8888)IOLoop.current().start()每次调用分片接口时，都会为分片文件创建一个定时任务监控，如果分片成功，则删除分片文件，同时删除任务，否则，启用降级计划。一篇：聚是群火散星，前端Vue.js+elementUI结合后端FastAPI分片上传大文件，逻辑类似，只是方法略有不同，就是相当有相互引用，最终代码开源在Github：https://github.com/zcxey2911/...\_Vuejs3\_Edu，分享给各位亲们。原文转载自《刘越的技术博客》https://v3u.cn/a_id_218
