如何使用pyscopg2的异步特性？

Question

我想使用不同的表执行3个不同的postgresql查询。每个查询需要2秒钟才能执行。我想知道是否可能同时运行这3个查询，以便我可以节省4秒钟的时间。我尝试使用pyscopg2的异步特性，但它只返回最后一个查询的结果。有人能指出我做错了什么吗？

import select
import psycopg2
import psycopg2.extensions

def wait(conn):
    while 1:
        state = conn.poll()
        if state == psycopg2.extensions.POLL_OK:
            break
        elif state == psycopg2.extensions.POLL_WRITE:
            select.select([], [conn.fileno()], [])
        elif state == psycopg2.extensions.POLL_READ:
            select.select([conn.fileno()], [], [])
        else:
            raise psycopg2.OperationalError("poll() returned %s" % state)


aconn = psycopg2.connect(
  dbname=pg_name,
  user=pg_username,
  host=pg_host,
  password=pg_password,
  async=1)

wait(aconn)
acurs = aconn.cursor()

acurs.execute(
              "SELECT 1;"
              "SELECT ST_Length(ST_GeomFromText"
              "('LINESTRING(743238 2967416,743238 2967450)',4326));"
              "SELECT 3;"
             )
wait(acurs.connection)
result = acurs.fetchall()
print result

这只会打印出："result": [[3]]

。

- Rahul

Answer 1

根据 Psycopg 介绍：

[Psycopg]是libpq的包装器，它是官方的PostgreSQL客户端库。

然后，查看 libpq 文档 中的 PQexec()（用于向PostgreSQL数据库发送SQL查询的函数）时，我们看到以下说明（重点在于我）：

在单个PQexec调用中发送的多个查询将在单个事务中处理，除非查询字符串中包含显式的BEGIN/COMMIT命令将其分成多个事务。 但请注意，返回的PGresult结构仅描述了从字符串执行的最后一个命令的结果。

因此，不幸的是，你试图做的事情并不受 psycopg2 和 libpq 的支持。（这并不是说其他与PostgreSQL的客户端接口不支持它，但这超出了本问题的范围。）

所以回答你的问题，你做错的是在一个execute()调用中执行多个SQL查询，并尝试在之后检索所有结果，事实上这是不可能的。你需要明确地执行每个查询并单独检索结果，否则尝试找到另一个支持一次返回多个结果集的PostgreSQL API。

---

Python数据库API 2.0规范确实允许库实现可选的nextset()方法，该方法将cursor移动到执行的查询返回的下一个结果集，但是出于明显的原因，psycopg2并未实现此方法，并且事实上如果您尝试调用它，会引发一个NotSupportedError异常（请参阅文档）。

- villapx

Answer 2

看起来从2.2版本开始支持了

def wait(conn):
    while True:
        state = conn.poll()
        if state == psycopg2.extensions.POLL_OK:
            break
        elif state == psycopg2.extensions.POLL_WRITE:
            select.select([], [conn.fileno()], [])
        elif state == psycopg2.extensions.POLL_READ:
            select.select([conn.fileno()], [], [])
        else:
            raise psycopg2.OperationalError("poll() returned %s" % state)

来源：https://www.psycopg.org/docs/advanced.html#asynchronous-support（异步支持）

- Sameer Alibhai