Как правильно отключиться от RabbitMQ с помощью Perl AnyEvent::RabbitMQ?

Я хочу правильно отключиться от RabbitMQ. просмотр исходного кода Perl's AnyEvent::RabbitMQ (который я использую), я нашел метод close, который, кажется, закрывается все каналы открыты для RabbitMQ.

So I

подключен к RabbitMQ
открыл канал
объявил обмен
привязан к этому обмену
объявлена очередь
привязан к этой очереди
выполнить метод close в экземпляре AnyEvent::RabbitMQ (не в экземпляре ::Channel)

Соединение кажется закрытым, но журналы RabbitMQ показывают, что «соединение AMQP» было «connection_closed_abruptly».

Вот полный журнал RabbitMQ для этого соединения:

=INFO REPORT==== 14-Jan-2016::10:02:15 ===
accepting AMQP connection <0.10868.0> (127.0.0.1:57764 -> 127.0.0.1:5672)

=WARNING REPORT==== 14-Jan-2016::10:02:16 ===
closing AMQP connection <0.10868.0> (127.0.0.1:57764 -> 127.0.0.1:5672):
connection_closed_abruptly

Вот пример кода:

#!/usr/bin/perl
use strictures 1;

use AnyEvent::RabbitMQ;
use Data::Printer;

my ( $rabbitmq, $rabbitmq_channel );

my $condvar = AnyEvent->condvar;

$rabbitmq = AnyEvent::RabbitMQ->new->load_xml_spec()->connect(
  host       => '127.0.0.1',
  port       => 5672,
  user       => 'guest',
  pass       => 'guest',
  vhost      => '/',
  timeout    => 1,
  tls        => 0,
  tune       => { heartbeat => 1 },
  on_success => sub {
    ($rabbitmq) = @_;
    $rabbitmq->open_channel(
      on_success => sub {
        ($rabbitmq_channel) = @_;
        $rabbitmq_channel->confirm;
        $rabbitmq_channel->declare_exchange(
          exchange   => 'test_exchange',
          type       => 'fanout',
          on_success => sub {
            $rabbitmq_channel->bind_exchange(
              source      => 'test_exchange',
              destination => 'test_exchange',
              routing_key => '',
              on_success  => sub {
                $rabbitmq_channel->declare_queue(
                  queue      => 'test_queue',
                  on_success => sub {
                    $rabbitmq_channel->bind_queue(
                      queue       => 'test_queue',
                      exchange    => 'test_exchange',
                      routing_key => '',
                      on_success  => sub {
                        $rabbitmq->close;
                        undef $rabbitmq;
                      },
                      on_failure => sub { $condvar->send( __LINE__, @_ ) },
                    );
                  },
                  on_failure => sub { $condvar->send( __LINE__, @_ ) },
                );
              },
              on_failure => sub { $condvar->send( __LINE__, @_ ) },
            );
          },
          on_failure => sub { $condvar->send( __LINE__, @_ ) },
        );
      },
      on_failure => sub { $condvar->send( __LINE__, @_ ) },
      on_return  => sub { $condvar->send( __LINE__, @_ ) },
      on_close   => sub { $condvar->send( __LINE__, @_ ) },
    );
  },
  on_failure      => sub { $condvar->send( __LINE__, @_ ) },
  on_read_failure => sub { $condvar->send( __LINE__, @_ ) },
  on_return       => sub { $condvar->send( __LINE__, @_ ) },
  on_close        => sub { $condvar->send( __LINE__, @_ ) },
);

my $reason = [ $condvar->recv ];
p $reason;

Как правильно отключиться от RabbitMQ, используя Perl AnyEvent::RabbitMQ?

burnersk 14.01.2016 источник

Ответы (2)

arrow_upward
1
arrow_downward

Имеются индикаторы опорных циклов. Это может помешать правильному разрушению конструкции.

"my $rabbitmq; $rabbitmq = ..." выкрикивает возможность эталонного цикла.
"my $rabbitmq_channel; $rabbitmq_channel = ..." выкрикивает возможность эталонного цикла.
$rabbitmq_channel принадлежит (хранится) $rabbitmq, но также захватывается обработчиками событий $rabbitmq_channel.

Изменения, отмеченные <===, заменяют неприемлемый код.

Изменение, отмеченное <---, может быть необходимо. Если $rabbitmq_channel не определено в обратных вызовах, удалите это изменение.

use Scalar::Util qw( weaken );

my $done_cv = AnyEvent->condvar;

my $rabbitmq = AnyEvent::RabbitMQ->new->load_xml_spec()->connect(  # <===
  host       => '127.0.0.1',
  port       => 5672,
  user       => 'guest',
  pass       => 'guest',
  vhost      => '/',
  timeout    => 1,
  tls        => 0,
  tune       => { heartbeat => 1 },
  on_success => sub {
    my ($rabbitmq) = @_;  # <===
    $rabbitmq->open_channel(
      on_success => sub {
        my ($rabbitmq_channel) = @_;  # <===
        {  # <---
          my $rabbitmq_channel = weaken($rabbitmq_channel);  # <---
          $rabbitmq_channel->confirm;
          $rabbitmq_channel->declare_exchange(
            exchange   => 'test_exchange',
            type       => 'fanout',
            on_success => sub {
              $rabbitmq_channel->bind_exchange(
                source      => 'test_exchange',
                destination => 'test_exchange',
                routing_key => '',
                on_success  => sub {
                  $rabbitmq_channel->declare_queue(
                    queue      => 'test_queue',
                    on_success => sub {
                      $rabbitmq_channel->bind_queue(
                        queue       => 'test_queue',
                        exchange    => 'test_exchange',
                        routing_key => '',
                        on_success  => sub { $done_cv->send( __LINE__, @_ ) },  # <===
                        on_failure => sub { $done_cv->send( __LINE__, @_ ) },
                      );
                    },
                    on_failure => sub { $done_cv->send( __LINE__, @_ ) },
                  );
                },
                on_failure => sub { $done_cv->send( __LINE__, @_ ) },
              );
            },
            on_failure => sub { $done_cv->send( __LINE__, @_ ) },
          );
        }  # <---
      },
      on_failure => sub { $done_cv->send( __LINE__, @_ ) },
      on_return  => sub { $done_cv->send( __LINE__, @_ ) },
      on_close   => sub { $done_cv->send( __LINE__, @_ ) },
    );
  },
  on_failure      => sub { $done_cv->send( __LINE__, @_ ) },
  on_read_failure => sub { $done_cv->send( __LINE__, @_ ) },
  on_return       => sub { $done_cv->send( __LINE__, @_ ) },
  on_close        => sub { $done_cv->send( __LINE__, @_ ) },
);

my $reason = [ $done_cv->recv ];
p $reason;

Надеюсь, это поможет.

ikegami 18.01.2016

arrow_upward
0
arrow_downward

Проблема заключается в ошибке в самой библиотеке AnyEvent::RabbitMQ.pm. Я точно не знаю, как исправить сам sub close, но ключевая часть заключается в том, что он никогда не выполняет код, который отправляет методы Connection::Close и Connection::CloseOk на ваш сервер RabbitMQ во время глобальной деконструкции. Вы можете подтвердить это, выполнив следующие действия после настройки соединения AMQP.

$rabbitmq->_push_write(Net::AMQP::Protocol::Connection::Close->new());
$rabbitmq->_push_write(Net::AMQP::Protocol::Connection::CloseOk->new());

Это немного хак, поэтому я работаю над правильным способом сделать это, и, надеюсь, сопровождающий примет запрос на включение.

Per Jensen 16.04.2016

Как правильно отключиться от RabbitMQ с помощью Perl AnyEvent::RabbitMQ?

Ответы (2)

Похожие вопросы