使用AnyEvent::Handle和tcp_connect重新连接

Question

我有一个使用AnyEvent::Handle、利用tcp_connect和tcp_server编写的简单的TCP服务器和客户端。客户端连接到服务器，并每5秒发送一次字符串Test Message。

但是，如果服务器在客户端启动时不可用，或变得不可用，则客户端脚本永远不会尝试重新连接。

如果连接句柄不可用，我希望它尝试重新连接(销毁？)。如果不可用，做一些事情(可能打印状态信息)，但是尝试每5秒重新连接将是理想的结果。

我不知道该怎么做。我已经将我的客户端和服务器代码缩减到以下内容。

客户端

#!/usr/bin/perl

use strict;
use warnings;

use AnyEvent;
use AnyEvent::Handle;
use AnyEvent::Socket;
use Compress::Zlib;

my @bulk;

# Start Timer
my $timer = AnyEvent->timer(
    after    => 5,
    interval => 5,
    cb       => sub {
        push( @bulk, "Test message" );
        flush( \@bulk );
        undef @bulk;
    } );

my $host = '127.0.0.1';
my $port = 9999;

my $conn_cv = AnyEvent->condvar;
my $conn_hdl;

$conn_hdl = AnyEvent::Handle->new(
    connect          => [$host, $port],
    keepalive        => 1,
    on_connect_error => sub {
        print "Could not connect: $_[1]\n";
        $conn_hdl->destroy;

        #$conn_cv->send;
    },
    on_error => sub {
        my ( $out_hdl, $fatal, $msg ) = @_;
        AE::log error => $msg;
        $conn_hdl->destroy;

        #$conn_cv->send;
    },
    on_read => sub {
        my ( $self ) = @_;
        $self->unshift_read(
            line => sub {
                my ( $hdl, $data ) = @_;
                print $data. "\n";
            } );
    } );

$conn_cv->recv;

# Flush array of events
sub flush {
    my ( $bulk ) = @_;
    return 0 if scalar @{$bulk} == 0;

    my $output = join( ",", @{$bulk} );
    $output = compress( $output );
    my $l = pack( "N", length( $output ) );
    $output = $l . $output;
    $conn_hdl->push_write( $output );
}

服务器

#!/usr/bin/perl

use strict;
use warnings;

use AnyEvent;
use AnyEvent::Handle;
use AnyEvent::Socket;
use Compress::Zlib;

my %holding;

my $host = '127.0.0.1';
my $port = 9999;

my %connections;

# Start Timer
my $timer = AnyEvent->timer(
    after    => 5,
    interval => 5,
    cb       => sub {
        print "Number of connected hosts: ";
        print scalar keys %connections;
        print "\n";
        foreach my $k ( keys %connections ) {
            delete $connections{$k} if $connections{$k}->destroyed;
        }
    } );

my $server_cv = AnyEvent->condvar;
my $server    = tcp_server(
    $host, $port,
    sub {
        my ( $fh, $h, $p ) = @_;
        my $handle;

        $handle = AnyEvent::Handle->new(
            fh        => $fh,
            poll      => 'r',
            keepalive => 1,
            on_read   => sub {
                my ( $self ) = @_;

                # Get Length Header
                $self->unshift_read(
                    chunk => 4,
                    sub {
                        my $len = unpack( "N", $_[1] );

                        # Get Data
                        $self->unshift_read(
                            chunk => $len,
                            sub {
                                my $data = $_[1];
                                $data = uncompress( $data );
                                print $data. "\n";
                            } );
                    } );

            },
            on_eof => sub {
                my ( $hdl ) = @_;
                $hdl->destroy();
            },
            on_error => sub {
                my ( $hdl ) = @_;
                $hdl->destroy();
            },
        );

        $connections{ $h . ':' . $p } = $handle;    # keep it alive.
    } );

$server_cv->recv;

Answer 1

您可以使用以下内容：

package MyConnector;

use strict;
use warnings;

use AE               qw( );
use AnyEvent::Handle qw( );
use Scalar::Util     qw( );

sub new {
   my $class = shift;
   my %opts = @_;

   my $self = bless({}, $class);

   {
      Scalar::Util::weaken(my $self = $self);

      my $on_connect       = delete($opts{on_connect});
      my $on_connect_error = delete($opts{on_connect_error});

      my $tries    = delete($opts{tries})    ||  5;
      my $cooldown = delete($opts{cooldown}) || 15;

      $self->{_connect} = sub {
         $self->{_timer} = undef;

         $self->{_handle} = AnyEvent::Handle->new(
            %opts,

            on_connect => sub {
               my ($handle, $host, $port, $retry) = @_;

               $self->{handle} = $handle;
               delete @{$self}{qw( _connect _handle _timer )};

               $on_connect->($handle, $host, $port, $retry)
                  if $on_connect;
            },

            on_connect_error => sub {
               my ($handle, $message) = @_;

               if (!$tries--) {
                  $on_connect_error->($handle, $message)
                     if $on_connect_error;

                  delete @{$self}{qw( _connect _handle _timer )};

                  return;
               }

               # This will happen when this callback returns,
               # but that might not be for a while, so let's
               # do it now in case it saves resources.
               $handle->destroy();

               $self->{_timer} = AE::timer($cooldown, 0, $self->{_connect});
            },
         );
      };

      $self->{_connect}->();
   }

   return $self;
}

sub handle {
   my ($self) = @_;
   return $self->{handle};
}

1;

我很确定它没有内存泄漏(不像您的代码)。您使用它的方式如下：

use strict;
use warnings;

use AE          qw( );
use MyConnector qw( );

my $host = $ARGV[0] || 'www.stackoverflow.com';
my $port = $ARGV[1] || 80;

my $conn_cv = AE::cv();

my $connector = MyConnector->new(
   connect   => [ $host, $port ],
   keepalive => 1,

   on_connect => sub {
       print("Connected successfully\n");
       $conn_cv->send();
   },

   on_connect_error => sub {
       warn("Could not connect: $_[1]\n");
       $conn_cv->send();
   },

   # ...
);

$conn_cv->recv();

Answer 2

多亏了ikegami，我想我可能想要跟踪连接状态，并与另一个AnyEvent计时器观察器一起使用它来重新连接，如果连接没有建立。如果连接状态($isConnected)为零，则每秒钟都会发生连接尝试。同时，当重新建立连接时，事件将排队等待。

如果有更简单的方法来完成这个任务，我会全神贯注，但现在我认为这将解决这个问题。

my @bulk;
my $host = '127.0.0.1';
my $port = 9999;
my $isConnected = 0;

my $conn_cv = AnyEvent->condvar;
my $conn_hdl;

# Flush Timer
my $timer = AnyEvent->timer(
    after => 5,
    interval => 5,
    cb => sub {
        push(@bulk,"Test message");
        if ($isConnected == 1) {
            flush(\@bulk);
            undef @bulk;
        }   
    }
);

# Reconnect Timer
my $reconn = AnyEvent->timer(
    after => 1,
    interval => 1,
    cb => sub {

        if ($isConnected == 0) {

            $conn_hdl = AnyEvent::Handle->new(
                connect => [$host, $port],
                keepalive => 1,
                on_connect => sub {
                    $isConnected = 1;
                },
                on_connect_error => sub {
                    warn "Could not connect: $_[1]\n";
                    $conn_hdl->destroy;
                    $isConnected = 0;       
                },
                on_error => sub {
                    my ($out_hdl, $fatal, $msg) = @_;
                    AE::log error => $msg;
                    $conn_hdl->destroy;
                    $isConnected = 0;
                },
                on_eof => sub {
                    warn "EOF\n";
                    $conn_hdl->destroy;
                    $isConnected = 0;
                },
                on_read => sub {
                    my ($self) = @_;
                        $self->unshift_read(line => sub {
                            my ($hdl,$data) = @_;
                            print $data."\n";
                    });
                }
            );  
        }
    }
);

$conn_cv->recv;