假设您需要执行一些依赖于某些临时文件的操作.由于我们在这里谈论Node,这些操作显然是异步的.为了知道何时可以删除临时文件,等待所有操作完成的惯用方法是什么？
这是一些显示我想要做的代码:
do_something(tmp_file_name, function(err) {});
do_something_other(tmp_file_name, function(err) {});
fs.unlink(tmp_file_name);
但是如果我这样写,第三次调用可以在前两次有机会使用该文件之前执行.我需要一些方法来保证前两个调用已经完成(调用它们的回调),然后继续运行而不嵌套调用(并使它们在实践中同步).
我想过在回调上使用事件发射器并将计数器注册为接收器.计数器将接收已完成的事件并计算仍有待处理的操作数.当最后一个完成时,它将删除该文件.但是存在竞争条件的风险,我不确定这通常是怎么做的.
Node人如何解决这类问题？
Alf*_*red 94
更新:
现在我建议看看:
Promise对象用于延迟和异步计算.Promise表示尚未完成的操作,但预计将来会发生.
你应该使用promises来解决这个问题:
Run Code Online (Sandbox Code Playgroud)fs.readFile("file.json", function (err, val) { if (err) { console.error("unable to read file"); } else { try { val = JSON.parse(val); console.log(val.success); } catch (e) { console.error("invalid json in file"); } } });进入:
Run Code Online (Sandbox Code Playgroud)fs.readFileAsync("file.json").then(JSON.parse).then(function (val) { console.log(val.success); }) .catch(SyntaxError, function (e) { console.error("invalid json in file"); }) .catch(function (e) { console.error("unable to read file"); });
发电机:例如通过公司.
使用promises为nodejs和浏览器提供基于生成器的控制流优度,让你以一种很好的方式编写非阻塞代码.
Run Code Online (Sandbox Code Playgroud)var co = require('co'); co(function *(){ // yield any promise var result = yield Promise.resolve(true); }).catch(onerror); co(function *(){ // resolve multiple promises in parallel var a = Promise.resolve(1); var b = Promise.resolve(2); var c = Promise.resolve(3); var res = yield [a, b, c]; console.log(res); // => [1, 2, 3] }).catch(onerror); // errors can be try/catched co(function *(){ try { yield Promise.reject(new Error('boom')); } catch (err) { console.error(err.message); // "boom" } }).catch(onerror); function onerror(err) { // log any uncaught errors // co will not throw any errors you do not handle!!! // HANDLE ALL YOUR ERRORS!!! console.error(err.stack); }
如果我理解正确,我认为你应该看看非常好的异步库.你应该特别看看这个系列.只是来自github页面的片段的副本:
async.series([
function(callback){
// do some stuff ...
callback(null, 'one');
},
function(callback){
// do some more stuff ...
callback(null, 'two');
},
],
// optional callback
function(err, results){
// results is now equal to ['one', 'two']
});
// an example using an object instead of an array
async.series({
one: function(callback){
setTimeout(function(){
callback(null, 1);
}, 200);
},
two: function(callback){
setTimeout(function(){
callback(null, 2);
}, 100);
},
},
function(err, results) {
// results is now equals to: {one: 1, two: 2}
});
作为一个加号,这个库也可以在浏览器中运行.
Mic*_*lon 22
最简单的方法是在启动异步操作时递增整数计数器,然后在回调中递减计数器.根据复杂性,回调可以将计数器检查为零,然后删除该文件.
稍微复杂的是维护一个对象列表,每个对象都有你需要识别操作的任何属性(它甚至可以是函数调用)以及状态代码.回调会将状态代码设置为已完成.
然后你会有一个循环等待(使用process.nextTick)并检查是否所有任务都已完成.这种方法优于计数器的优点是,如果所有未完成的任务都可以完成,则在发出所有任务之前,计数器技术会导致您过早地删除文件.
goo*_*gic 11
// simple countdown latch
function CDL(countdown, completion) {
this.signal = function() {
if(--countdown < 1) completion();
};
}
// usage
var latch = new CDL(10, function() {
console.log("latch.signal() was called 10 times.");
});
没有"本机"解决方案,但节点有一百万个流量控制库.你可能会喜欢Step:
Step(
function(){
do_something(tmp_file_name, this.parallel());
do_something_else(tmp_file_name, this.parallel());
},
function(err) {
if (err) throw err;
fs.unlink(tmp_file_name);
}
)
或者,正如迈克尔所说,计数器可能是一个更简单的解决方案.看看这个信号量模型.你会这样使用它:
do_something1(file, queue('myqueue'));
do_something2(file, queue('myqueue'));
queue.done('myqueue', function(){
fs.unlink(file);
});
我想提供另一种解决方案,利用节点:事件核心的编程范例的速度和效率.
您可以使用Promises或用于管理流控制的模块async来完成所有工作,例如,可以使用事件和简单的状态机来完成,我认为这种方法可能比其他选项更容易理解.
例如,假设您希望并行处理多个文件的长度:
const EventEmitter = require('events').EventEmitter;
// simple event-driven state machine
const sm = new EventEmitter();
// running state
let context={
tasks: 0, // number of total tasks
active: 0, // number of active tasks
results: [] // task results
};
const next = (result) => { // must be called when each task chain completes
if(result) { // preserve result of task chain
context.results.push(result);
}
// decrement the number of running tasks
context.active -= 1;
// when all tasks complete, trigger done state
if(!context.active) {
sm.emit('done');
}
};
// operational states
// start state - initializes context
sm.on('start', (paths) => {
const len=paths.length;
console.log(`start: beginning processing of ${len} paths`);
context.tasks = len; // total number of tasks
context.active = len; // number of active tasks
sm.emit('forEachPath', paths); // go to next state
});
// start processing of each path
sm.on('forEachPath', (paths)=>{
console.log(`forEachPath: starting ${paths.length} process chains`);
paths.forEach((path) => sm.emit('readPath', path));
});
// read contents from path
sm.on('readPath', (path) => {
console.log(` readPath: ${path}`);
fs.readFile(path,(err,buf) => {
if(err) {
sm.emit('error',err);
return;
}
sm.emit('processContent', buf.toString(), path);
});
});
// compute length of path contents
sm.on('processContent', (str, path) => {
console.log(` processContent: ${path}`);
next(str.length);
});
// when processing is complete
sm.on('done', () => {
const total = context.results.reduce((sum,n) => sum + n);
console.log(`The total of ${context.tasks} files is ${total}`);
});
// error state
sm.on('error', (err) => { throw err; });
// ======================================================
// start processing - ok, let's go
// ======================================================
sm.emit('start', ['file1','file2','file3','file4']);
哪个会输出:
start: beginning processing of 4 paths forEachPath: starting 4 process chains readPath: file1 readPath: file2 processContent: file1 readPath: file3 processContent: file2 processContent: file3 readPath: file4 processContent: file4 The total of 4 files is 4021
请注意,流程链任务的顺序取决于系统负载.
您可以将程序流设想为:
start -> forEachPath -+-> readPath1 -> processContent1 -+-> done
+-> readFile2 -> processContent2 -+
+-> readFile3 -> processContent3 -+
+-> readFile4 -> processContent4 -+
为了重复使用,创建一个模块以支持各种流控制模式(即串行,并行,批量,同时,直到等)将是微不足道的.
| 归档时间： |
|
| 查看次数： |
49779 次 |
| 最近记录： |