擅长:python、mysql、java
<p>我不熟悉dagster,但我在其他上下文中成功使用的一种机制是利用这样一个事实,即在类似Unix的系统中,rename或mv是一种原子操作。对于运行后清理的第一个要求:</p>
<ol>
<li><p>新文件被放入输入目录。一组输入文件可以在它们自己的目录中隔离</p>
</li>
<li><p>当管道进程启动时,它的第一个操作是从输入目录中选择一个文件(或目录),并将其移动到管道实例拥有的工作目录中。如果输入目录中没有可用的文件,进程会自动关闭</p>
</li>
<li><p>如果mv成功,进程将继续在它刚刚移动到其工作目录的文件(目录)上执行它的操作。当它完成时,它会自行清理,可能是通过对其工作目录执行递归删除</p>
</li>
<li><p>如果mv失败,则意味着另一个进程从该进程下取出了新文件。失败的过程会优雅地自行关闭</p>
</li>
</ol>
<p>对于一次只运行一个管道进程的第二个要求,您可以使用独占方式创建sentinel文件,如果该进程没有成功创建sentinel文件,则该进程将失败并退出。在Python3中,代码可能类似于</p>
<pre><code>try:
open('sentinel', 'x').close()
except FileExistsError:
exit("someone else already has sentinel")
do_stuff()
os.remove('sentinel')
</code></pre>
<p>当然,如果您的进程在do_stuff()中的某个地方崩溃,您必须手动清理sentinel文件,或者您可以使用atexit处理程序确保即使在do_stuff()中崩溃的情况下也删除sentinel</p>