??????感謝各位朋友接下來的閱讀???????

分治算法概念：?

分治算法，即分而治之：

?在計算機科學中，分治法是一種很重要的算法。分治算法，字面上的解釋是“分而治之”，分治算法主要是三點： 1.將一個復雜的問題分成兩個或更多的相同或相似的子問題，再把子問題分成更小的子問題----“分”
2.將最后子問題可以簡單的直接求解----“治”
3.將所有子問題的解合并起來就是原問題打得解----“合”

分治法解析圖：

?分治算法的基本思想：

????????????當我們求解某些問題時，由于這些問題要處理的數據相當多，或求解過程相當復雜，使得直接求解法在時間上相當長，或者根本無法直接求出。對于這類問題，我們往往先把它分解成幾個子問題，找到求出這幾個子問題的解法后，再找到合適的方法，把它們組合成求整個問題的解法。如果這些子問題還較大，難以解決，可以再把它們分成幾個更小的子問題，以此類推，直至可以直接求出解為止。這就是分治策略的基本思想。

分治算法使用：

我做過的一個超大文件上傳，為了提高效率，采用的是分治算法，現在的視頻文件的體積越來愈大，傳輸速度慢，耗時時間太長，給用戶帶來不好的體驗效果，我做的是使用前端Vue.js3.0然后配合ui庫，阿里云的Ant-design，后端采用并發異步框架tornado，來實現大文件的分片無阻塞傳輸與異步io寫入服務。

文件分片：

文件分片是在前端進行的，通過選擇文件，來獲取文件名，文件大小，自定義分片大小的閾值，
通過math函數ceil（總文件體積除以定義的單片閾值大小）進行向上取整來計算總片數 ，

然后通過slice方法來對文件進行切片，從0的地方開始切片，結束位置使用Math.min方法（開始位置加上定義分片大小），

最后將分片的下標，文件名，以及分片實體異步發送到后臺。

?我前端使用的是Ant-design：組件總覽 - Ant Design Vue (antdv.com)

<a-upload :before-upload="beforeUpload" @change="handleChange"><a-button>自動上傳視頻</a-button></a-upload><img :src="src" v-show="src" /> export default {data() {return {//分片個數shardCount: 0,finished: 0,src: "",websrc: "",size:0,};},

methods:

methods: {//前端進行判斷，如果errcode==0，則調用輪詢接口，會將filename,分片數，以及尺寸大小，jobid//傳送后端輪詢中，進行分片判斷，若不等就是分片傳輸未成功，判斷文件大小刪除分片，刪除定時任務//發起定時任務task:function(filename){this.myaxios(this.taskurl + "scheduler/", "get", { "filename": filename,"count":this.shardCount,"size":this.size,"job_id":'1' }).then((data) => {console.log(data);});},//發送分片文件pushshard: function (file, filename, count) {const axiosupload = this.axios.create({ withCredentials: false });let data = new FormData();data.append("file", file);data.append("filename", filename);data.append("count", count);//發起請求axiosupload({method: "POST",url: this.baseurl + "upload/",data: data,}).then((data) => {console.log("111", data);if (data.data.errcode == 0) {this.finished += 1;if (this.finished == this.shardCount) {this.myaxios(this.baseurl + "upload/", "put", {filename: filename,size:this.size}).then((data) => {if (data.errcode == 0) {this.src = this.upload_dir + "/" + filename;this.websrc = filename;this.finished = 0;this.task(filename)}});}}});},handleChange(file) {//獲取文件大小 1024字節=1kbvar size = file.file.sizethis.size = sizeconsole.log('總大',size)//定義分片大小1G=1024MB，1MB=1024KB，1KB=1024字節var shardSize = 2 *1000 * 1024//總片數//向上取整this.shardCount = Math.ceil(size / shardSize)console.log(this.shardCount)//切片操作for (var i = 0; i < this.shardCount; i++) {// 開始位置var start = i * shardSize;//結束位置var end = Math.min(size, start + shardSize)console.log("切片", start, end);//切片var shardfile = file.file.slice(start, end)this.pushshard(shardfile, file.file.name, i)}},// 上傳圖片beforeUpload: function () {return false;},//提交submit: function () {var temp = [];//取值for (var i = this.forms.length - 1; i >= 0; i--) {temp.push({label: this.forms[i]["label"],value: this.forms[i]["value"],});}temp.push({ "video": this.src });temp = JSON.stringify(temp);this.myaxios(this.baseurl + "merchantform/", "put", { form: temp }).then((data) => {console.log(data);});}, }

后端分片邏輯：?

后端接口使用post請求，獲取到分片實體，下標和文件名，以及獲取文件內容
使用await異步寫入，分片成功，

然后再通過put請求，獲取到filename和前端的文件大小，使用get.path.getsize方法獲取文件大小，
如果前端傳來的文件和讀取的文件大小不相等，則異步打開文件句柄，使用open方法循環讀取分片文件，將讀取到的
分片文件異步寫入，手動關閉句柄，停止循環，將文件數量+1，合并完成

后端異步io寫入：

為了避免同步寫入引起的阻塞，安裝aiofiles庫：

pip3 install aiofiles

aiofiles用于處理asyncio應用程序中的本地磁盤文件，配合Tornado的異步非阻塞機制，可以有效的提升文件寫入效率：

# 上傳文件 import aiofiles import os from base import BaseHandler # 導入路由 from tornado.web import url# 分片上傳 post請求體傳參，不會被url截斷 class SliceUploadHandler(BaseHandler):async def post(self):# 獲取分片實體file = self.request.files["file"][0]# 獲取下標count = self.get_argument("count",None)# 文件名filename = self.get_argument("filename",None)# 獲取文件內容content = file["body"]# 異步寫入async with aiofiles.open("./static/uploads/{}_{}".format(filename,count),"wb") as file:# 異步await file.write(content)self.finish({"errcode":0,'msg':'分片上傳成功'})async def put(self):filename = self.get_argument("filename",None)print('filename',filename)count = 0size = self.get_argument("size", None)print('size',size)try:filesize = os.path.getsize("./static/uploads/{}".format(filename))except Exception as e:print(str(e))filesize = 0if int(size) != filesize:#異步打開文件句柄async with aiofiles.open("./static/uploads/{}".format(filename),"ab") as file:while True:try:# 循環讀取分片文件shard_file = open("./static/uploads/{}_{}".format(filename,count),'rb')# 異步寫入await file.write(shard_file.read())# 手動關閉句柄shard_file.close()except Exception as e:print(str(e))breakcount = count + 1self.finish({"errcode":0,"msg":'合并完畢'})# 聲明路由 urlpatterns = [url('/upload/',SliceUploadHandler),]

效果展示：

我這里采用圖片代用視頻啦！

重要問題：

當然了，用戶上傳文件時候，也會出現各種問題，就比如因為網絡或者其他的某些因素導致
分片的任務中斷，一定會影響到后面的文件合成，所以這個時候需要添加一個輪詢服務，來確保
出現異常能提示并且解決異常。這里我們使用基于tornado的Apscheduler庫來調度分片任務。

import redis import os import jsonfrom datetime import datetime from tornado.ioloop import IOLoop, PeriodicCallbackfrom tornado.web import Application, RequestHandler, urlfrom apscheduler.schedulers.tornado import TornadoSchedulerfrom test_async import mainfrom base import BaseHandlerfrom apscheduler.jobstores.redis import RedisJobStorefrom thread_lock import start_pool_threadscheduler = Nonejob_ids = []# 聲明任務：檢查分片def task(job_id,filename,size,count):count = int(count)filelist = os.listdir("./static/uploads/")temp = []for i in range(count):for x in filelist:if x == "{}_{}".format(filename,i):temp.append(x)if len(temp) != count:print("分片未傳輸成功")return Falsetry:filesize = os.path.getsize("./static/uploads/{}".format(filename))except Exception as e:print(str(e))filesize = 0if int(size) != filesize:print("分片合并未成功")return False# 刪除分片for x in range(count):os.remove("./static/uploads/{}_{}".format(filename,x))# 刪除定時任務scheduler.remove_job(job_id)jobs = Jobs()job_info = {"job_id": job_id, "filename": filename, "count": str(count), "size": size}job_info = json.dumps(job_info)print(job_info)jobs.remove(job_info)# job_ids.remove(job_id)print("分片執行完畢")# 初始化 def init_scheduler():global schedulerjobstores = {"default": RedisJobStore(jobs_key="jobs", run_times_key="jobtimes", host="localhost", port=6379,password=123)}scheduler = TornadoScheduler(jobstores=jobstores)# scheduler = TornadoScheduler()scheduler.start()# jobs = Jobs()# job_list = jobs.get_job_list()# for x in job_list:# x = json.loads(x)## scheduler.add_job(task, "interval", seconds=3, id=x["job_id"],# args=(x["job_id"], x["filename"], x["size"], x["count"]))print("定時任務啟動")# 聲明服務控制器 class SchedulerHandler(BaseHandler):async def get(self):job_id = self.get_argument("job_id",None)filename = self.get_argument("filename",None)size = self.get_argument("size",None)count = self.get_argument("count", None)jobs = Jobs()job_list = jobs.get_job_list()job_info = {"job_id": job_id, "filename": filename, "count": count, "size": size}job_info = json.dumps(job_info)if job_info not in job_list:jobs.insert(job_id,filename,count,size)scheduler.add_job(task,"interval",seconds=3,id=job_id,args=(job_id,filename,size,count))print("定時任務入隊")res = {"errcode": 0, "msg": "ok任務完成"}else:print("該任務已經存在")res = {"errcode": 1, "msg": 'fail失敗'}self.finish(res)# global job_ids# if job_id not in job_ids:## job_ids.append(job_id)## scheduler.add_job(task,"interval",seconds=3,id=job_id,args=(job_id,filename,size,count))## print("定時任務入隊")# res = {"errcode":0,"msg":"ok任務完成"}## else:# print("該任務已經存在")# res = {"errcode":1,"msg":'fail失敗'}## self.finish(res)if __name__ == '__main__':routes = [url(r"/scheduler/",SchedulerHandler)]# init_scheduler()# 聲明tornado對象print('56789')application = Application(routes,debug=True)application.listen(8888)IOLoop.current().start()

總結 ：分治法對超大文件進行分片切割，同時并發異步發送，可以提高傳輸效率，降低傳輸時間。

遵循三步：

（1）分解，將要解決的問題劃分成若干規模較小的同類問題；

（2）求解，當子問題劃分得足夠小時，用較簡單的方法解決；

（3）合并，按原問題的要求，將子問題的解逐層合并構成原問題的解。

總結

以上是生活随笔為你收集整理的分治算法 --- 详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。