AnyEvent :: STOMP :: Client + AnyEvent :: ForkManger =间歇性错误

我想写一个进程，侦听ActiveMQ并基于消息，出去并从Web服务抓取数据，做一些处理，然后将流程数据放到另一个Web服务中。（REST/JSON）AnyEvent :: STOMP :: Client + AnyEvent :: ForkManger =间歇性错误

下面的模块正常工作，直到我跟其中一个不知名的webservices返回错误。我已经尝试了很多东西来捕捉错误，但无济于事。一旦Web服务发生错误，虽然我得到以下信息：事件

未处理的异常回调（MESSAGE， AnyEvent :: STOMP ::客户端= HASH（0x3ad5e48），HASH（0x3a6bbb0） { “行动”： “创造”， “数据”：{ “ID”：40578737， “类型”： “警报”， “谁”：空}， “GUID”： “ADCCEE0C-73A7-11E6-8084-74B346D1CA67”， “主机名”： “MYSERVER”， “PID”：48632}）： $ fork_manager-> start（）方法应该在管理器进程中调用

OK，我在概念理解，子进程正在尝试启动另一个进程和叉子经理说这是一个不行。但是，考虑到下面的模块，启动新进程来处理长时间运行的处理的正确方法是什么。或者为什么一个子进程死亡造成这种例外，我怎样才能防止这种

这里的模块（精简）

package consumer; 

use AnyEvent::ForkManager; 
use AnyEvent::STOMP::Client; 
use JSON; 
use Data::Dumper; 
use v5.18; 
use Moose; 

sub run { 
    my $self = shift; 
    my $pm  = AnyEvent::ForkManager->new(max_workers => 20); 
    my $stomp = AnyEvent::STOMP::Client->new(); 

    $stomp->connect(); 
    $stomp->on_connected(sub { 
     my $stomp = shift; 
     $stomp->subscribe('/topic/test'); 
     say "Connected to STOMP"; 
    }); 

    $pm->on_start(sub { 
     my ($pm,$pid,@params) = @_; 
     say "Starting $pid worker"; 
    }); 

    $pm->on_finish(sub { 
     my ($pm, $pid,@params) = @_; 
     say "Finished $pid worker"; 
    }); 

    $pm->on_error(sub { 
     say Dumper(\@_); 
    }); 

    $stomp->on_message(sub { 
     my ($stomp, $header, $body) = @_; 
     my $href = decode_json $body; 
     $pm->start(cb => sub { 
      my ($pm, @params) = @_; 
      $self->process(@params); 
     }, 
     args => [ $href->{id}, $href->{data}->{type}, $href->{data}->{who} ], 
     ); 
    }); 

    my $cv = AnyEvent->condvar; 
    $cv->recv; 
} 

sub process { 
    say "Processing ".Dumper(\@_); 
    sleep 5; 
    if (int(rand(10)) < 5) { 
     die "OOPS"; # this triggers the error message above 
    } 
    say "Done Processing $_[1]"; 
} 

1;

继承人以上模块的驱动程序：

#!/usr/bin/env perl 

use v5.18; 
use lib '.'; 
use consumer; 

my $c = consumer->new(); 
$c->run;

最后一个流量发生器，你可以用它来看到这个行动：

#!/usr/bin/env perl 

use lib '../lib'; 
use lib '../../lib'; 
use v5.18; 

use Data::Dumper; 
use JSON; 
use Net::STOMP::Client; 

$ENV{'scot_mode'} = "testing"; 

my $stomp = Net::STOMP::Client->new(
    host => "127.0.0.1", 
    port => 61613 
); 
$stomp->connect(); 

for (my $i = 1; $i < 1000000; $i++) { 
    my $href = { 
     id  => $i, 
     type => "event", 
     what => "foo", 
    }; 
    my $json = encode_json $href; 
    say "Sending ".Dumper($href); 
    $stomp->send(
     destination => "/topic/test", 
     body  => $json, 
    ); 
} 

$stomp->disconnect();

来源

2016-09-05 Todd

我能够s通过使用Try :: Catch并将这个调用包装为self>过程，并尝试使用类似下面的方法：

$stomp->on_message(sub { 
     my ($stomp, $header, $body) = @_; 
     my $href = decode_json $body; 
     $pm->start(cb => sub { 
      my ($pm, @params) = @_; 
      try { 
       $self->process(@params); 
      } 
      catch { 
       # error handling stuff 
      }; 
     }, 
     args => [ $href->{id}, $href->{data}->{type}, $href->{data}->{who} ], 
     ); 
    } 
);

来源

2018-03-01 23:29:20 Todd

AnyEvent :: STOMP :: Client + AnyEvent :: ForkManger =间歇性错误

回答

相关问题