使用 DBD::Pg 的异步查询失败：在前一个异步查询完成之前无法执行

Question

我使用的是 CentOS 5.5 Linux（与 Redhat 5.5 相同） 使用股票 perl v5.8.8 并已安装 DBD-Pg-2.17.1 通过 CPAN shell 我正在使用 postgresql-server-8.4.5-1PGDG.rhel5 和朋友们。

我准备了一个简单的测试用例来演示 我的问题 - 它列在底部。

我的代码运行正常，当我删除 {pg_async => PG_ASYNC}

我的背景是我有一个小的 Facebook 游戏 作为带有 IO::Poll 的非分叉 Unix 守护进程运行。 我想为球员添加一些统计数据， 但我不想限制我的轮询循环，所以我会 喜欢发送主要是 INSERT/UPDATE 命令 asynchronously 我不需要任何返回值 来自数据库 - 因为用于读取/显示 我会有单独的网络脚本。

令人惊讶的是，我收到了错误消息 DBD::Pg::st 执行失败：在前一个异步查询完成之前无法执行即使我没有使用 PG_OLDQUERY_WAIT

这是我的代码（我的守护进程应该重新连接 每当连接丢失时到 PostgreSQL， 这就是为什么我使用 *_cached 方法并且不退出 eval { .... } 异常）：

#!/usr/bin/perl -w

use strict;
use DBI;
use DBD::Pg qw(:async);

use constant DBNAME => 'snake';
use constant DBUSER => 'snake';
use constant DBPASS => 'snake';

use constant SQL_CREATE_TABLES => q{
        /*
        create table pref_users (
                id varchar(32) primary key,
                first_name varchar(32),
                last_name varchar(32),
                female boolean,
                avatar varchar(128),
                city varchar(32),
                lat real check (-90 <= lat and lat <= 90),
                lng real check (-90 <= lng and lng <= 90),
                last_login timestamp default current_timestamp,
                last_ip inet,
                medals smallint check (medals > 0)
        );

        create table pref_rate (
                obj varchar(32) references pref_users(id),
                subj varchar(32) references pref_users(id),
                good boolean,
                fair boolean,
                nice boolean,
                about varchar(256),
                last_rated timestamp default current_timestamp
        );

        create table pref_money (
                id varchar(32) references pref_users,
                yw char(7) default to_char(current_timestamp, 'YYYY-WW'),
                money real
        );
        create index pref_money_yw_index on pref_money(yw);

        create table pref_pass (
                id varchar(32) references pref_users
        );

        create table pref_misere (
                id varchar(32) references pref_users
        );
        */

        create or replace function pref_update_users(_id varchar,
            _first_name varchar, _last_name varchar, _female boolean,
            _avatar varchar, _city varchar, _last_ip inet) returns void as $BODY$
                begin

                update pref_users set
                    first_name = _first_name,
                    last_name = _last_name,
                    female = _female,
                    avatar = _avatar,
                    city = _city,
                    last_ip = _last_ip
                where id = _id;

                if not found then
                        insert into pref_users(id, first_name,
                            last_name, female, avatar, city, last_ip)
                        values (_id, _first_name, _last_name,
                            _female, _avatar, _city, _last_ip);
                end if;
                end;
        $BODY$ language plpgsql;
};

eval {
        my $dbh = DBI->connect_cached('dbi:Pg:dbname=' .
            DBNAME, DBUSER, DBPASS, {
            AutoCommit => 1,
            PrintWarn => 1,
            PrintError => 1,
            RaiseError => 1,
            FetchHashKeyName => 'NAME_lc',
            pg_enable_utf8 => 1
        }, {pg_async => PG_ASYNC});

        $dbh->do(SQL_CREATE_TABLES, {pg_async => PG_ASYNC});
};
warn $@ if $@;

for my $i (1..10) {
        eval {
                my $dbh = DBI->connect_cached('dbi:Pg:dbname=' .
                    DBNAME, DBUSER, DBPASS, {
                    AutoCommit => 1,
                    PrintWarn => 1,
                    PrintError => 1,
                    RaiseError => 1,
                    FetchHashKeyName => 'NAME_lc',
                    pg_enable_utf8 => 1
                }, {pg_async => PG_ASYNC});

                #$dbh->pg_result;

                my $sth = $dbh->prepare_cached(
                    q{select pref_update_users(?, ?, ?, ?, ?, ?, NULL)}, {pg_async => PG_ASYNC});

                $sth->execute('ID123', 'Alexander', 'Farber', undef, undef, undef);
        };
        warn $@ if $@;
}

谢谢你， 亚历克斯

Answer 1

DBD::Pg 的异步支持就是这样工作的，一次只有一个活动的异步查询。 PG_OLDQUERY_CANCEL 和 PG_OLDQUERY_WAIT 常量需要设置为 on 如果你想取消或等待当前活动的查询然后执行新的查询，而不是抛出旧查询的错误。

您可以将查询添加到 AoH（哈希数组）或 Thread::Queue（忽略名称，它可用作通用队列对象并在计时器上执行它们（一旦前一个完成）（或使用 IO::Poll 将 $dbh->{pg_socket} 套接字添加到您的轮询套接字并检查查询的准备情况，并在该套接字有数据要读取时执行队列中的下一个查询，指示与 pg_ready 大致相同。