【发布时间】：2016-04-23 12:31:17
【问题描述】：
我正在实现一个分布式 cronjob 执行系统（所谓的 cron 计算集群）。当操作时间到了时，Cronjobs 应该排队到消息队列 (RabbitMQ) 中。另一方面（集群的节点/工作人员）是一个 Perl 守护进程，它利用 AnyEvent::RabbitMQ 从消息队列中接收一个 cronjob/task/消息，处理该任务并从消息队列等。
我使用通过AnyEvent::RabbitMQ 实现的 RabbitMQ 的心跳功能来帮助 RabbitMQ 识别断开的连接。
不要介意心跳间隔的实际值！我也有很长的工作需要几天的时间。因此，将间隔设置为最长的 cronjob 将不是一种选择。
请参阅以下 sn-p 以在 Perl 守护进程工作程序中执行实际的 cronjob。它是在“AnyEvent->timer”中实现的，不会对消息执行 DoSing RabbitMQ。使用此方法是因为 RabbitMQ 的consume 被禁止（被管理）。
sub _timer_tick {
$rabbitmq_channel->get(
queue => 'job_queue',
on_success => sub {
my ($amqp_method) = @_;
if ( not $amqp_method->{empty} ) {
pause_timer();
progress_job($amqp_method);
resume_timer();
}
},
on_failure => sub { $quit_programm->send( 'RABBITMQ_ERROR', @_ ) },
);
return;
}
progress_job() 是解析消息并执行作业的位置。 pause_timer() 和resume_timer() 控制触发_timer_tick() 的AnyEvent->timer。
use Capture::Tiny 'capture';
sub progress_job {
my ($amqp_method) = @_;
my $job = decode_json( $amqp_method->{body}->to_raw_payload() );
my ( $stdout, $stderr, $exit ) = capture {
system $job->{execute};
};
return;
}
第一个长时间运行的作业进入，系统“崩溃”并显示各种错误消息。有时它会抛出“未知频道 ID：1”，有时它会抛出“频道已关闭”。所以我做了“愚蠢的调试”（试图弄乱配置），发现当heartbeat间隔比progress_job()内所用的时间短时，这些错误将被抛出。经过一番思考，这是有道理的。 progress_job() 是一个阻塞子程序，AnyEvent 无法继续向 RabbitMQ 发送心跳包。
我对解决阻塞-heatbeat 问题的第一个快速想法是在子进程中分叉并执行progress_job()。 AnyEvents documentation on FORK 指出，当孩子内部无法访问事件系统（例如通过 AnyEvent）时，使用 fork 是省事的。
下一个想法：好的，没有访问事件系统的权限，所以我可以做 fork。
但是：计时器应该在progress_job() 返回后恢复（resume_timer()）。理论上，resume_timer() 会在fork() 之后被调用，而不是在progress_job() 返回之后。所以我停止了我的实现。
我的问题：如何解决最后一点？在progress_job()（或者换句话说，分叉的孩子）返回之后如何resume_timer()？
由于分叉和事件系统不是线程安全的，我无法将resume_timer() 放在孩子中。
【问题讨论】：
标签： multithreading perl rabbitmq fork anyevent