協(xié)程，也被稱為“用戶態(tài)線程”抖剿，是可以由用戶去實(shí)現(xiàn)并發(fā)調(diào)度的一種語(yǔ)言設(shè)施醇滥。
設(shè)想，用戶并不知道協(xié)程礼预，只提供一般的阻塞api眠砾，用戶如下使用:
import time
time.sleep(1.0)
print('after 1s')
可以翻譯成協(xié)程實(shí)現(xiàn):
import time
def __thread():
(yield time.sleep(1.0))
print('after 1s')
scheduler.spawn(__thread() )
本方案示例了javascript里協(xié)程調(diào)度器，鑒于python generator和javascript generator模型的一致性托酸，應(yīng)很容易實(shí)現(xiàn)對(duì)應(yīng)python版本褒颈。
1.耗時(shí)操作提供異步版本API，如 settimeout 和 node.js API設(shè)計(jì)
//異步等待
settimeout(callback, ms)
//異步讀文件
var fs = require('fs');
fs.readFile(this.path, function(err, data) {
//...
});
2.包裝成 異步標(biāo)記對(duì)象
class coTime extends iYield {
constructor(time) {
super();
this.time = time;
}
start(task) {
task.yield.done = 'waiting';
setTimeout(() => {
task.yield = task.cofunc.next(this.time);
}, this.time);
}
}
//最終提供的API
function wait(ms) {
return new coTime(ms);
}
3.如多線程一樣使用
// 創(chuàng)建10個(gè)(用戶)線程
for (let i = 0; i < 10; ++i) {
coManager.thread(
(function*() {
while (true) {
let time = yield wait(Math.random() * 2000);
console.log(`thread ${i} wait:` + time);
}
})()
);
}
這里 coManager是協(xié)程調(diào)度器获高，原理是，實(shí)現(xiàn)一個(gè)異步任務(wù)隊(duì)列吻育，每次異步等待時(shí)念秧，將協(xié)程中斷(suspend)，每異步返回時(shí)布疼，將中斷的協(xié)程繼續(xù)摊趾，完整可執(zhí)行的代碼如下
// 并發(fā)任務(wù)對(duì)象
class coTask {
constructor(cofunc) {
this.cofunc = cofunc;
this.yield = cofunc.next();
}
}
class iYield {
// 開始異步
start(task) {}
}
// 讀文件異步調(diào)用
class coReadFile extends iYield {
constructor(path) {
super();
this.path = path;
}
start(task) {
item.yield.done = 'waiting';
var fs = require('fs');
fs.readFile(this.path, function(err, data) {
if (err) {
task.yield = task.cofunc.throw(err);
} else {
task.yield = task.cofunc.next(data); //done and next
}
});
}
}
// helper
function readFile(path) {
return new coReadFile(path);
}
// 定時(shí)器異步對(duì)象
class coTime extends iYield {
constructor(time) {
super();
this.time = time;
}
start(task) {
task.yield.done = 'waiting';
setTimeout(() => {
task.yield = task.cofunc.next(this.time);
}, this.time);
}
}
function wait(ms) {
return new coTime(ms);
}
// 協(xié)程管理器，負(fù)責(zé)調(diào)度
class CoroutineManager {
constructor() {
this.taskLs = [];
this.update.bind(this);
setInterval(() => {
let taskLs = this.taskLs;
this.taskLs = [];
// 任務(wù)隊(duì)列輪循
for (let item of taskLs) {
if (item.yield.done != true) {
this.taskLs.push(item);
}
}
for (let item of taskLs) {
if (item.yield.done === 'waiting') {
continue;
} else if (item.yield.value instanceof iYield) {
item.yield.value.start(item);
} else {
item.yield = item.cofunc.next();
}
}
}, 1);
}
// 開(用戶)線程
thread(cofunc) {
let yValue = cofunc.next();
if (yValue.done) {
return;
}
this.taskLs.push({ yield: yValue, cofunc });
}
}
let coManager = new CoroutineManager();
// 創(chuàng)建10個(gè)（用戶）線程
for (let i = 0; i < 10; ++i) {
coManager.thread(
(function*() {
while (true) {
let time = yield wait(Math.random() * 2000);
console.log(`thread ${i} wait:` + time);
}
})()
);
}
python協(xié)程:
from browser import timer as systimer;
import random
import time
class Co:
def __init__(self):
pass
def getVal(self):
return None
class timer(Co):
def __init__(self, last):
self.last = last
self.start = time.clock()
def isDone(self):
last = time.clock()-self.start
if last>=self.last:
return True
else:
return False
def getVal(self):
return int(random.random()*self.last)
class Scheduler:
def __init__(self):
self.working = []
self.starting = []
self.timer = None
def run(self):
self.timer = systimer.set_interval(lambda: self.update(), 20);
def update(self):
starting = self.starting;
self.starting = []
for fun in starting:
try:
print(1)
r = next(fun)
print(2)
while not isinstance(r, Co):
r = next(fun)
self.working.append((fun, r))
except StopIteration:
pass
except Exception as ex:
print(ex)
working = self.working;
self.working = []
for (fun, r) in working:
try:
if r.isDone():
val = r.getVal()
r = fun.send(val)
while not isinstance(r, Co):
r = next(fun)
self.working.append((fun, r))
else:
self.working.append((fun, r))
except StopIteration:
pass
except Exception as ex:
print(ex)
def startCorotine(self, fun):
self.starting.append(fun)
scheduler = Scheduler()
