使用 postgresql gem 异步

Question

我正在使用 Goliath （由 eventmachine 提供支持）和 postgres gem pg，目前我正在以阻塞方式使用 pg gem：conn.exec('SELECT * FROM products') （例如），我想知道是否有更好的连接到 postgres 的方法 数据库？

Answer 1

pg 库为 PostgreSQL 的异步 API 提供全面支持。我添加了如何使用它的示例到 samples/ 目录：

#!/usr/bin/env ruby

require 'pg'

# This is a example of how to use the asynchronous API to query the
# server without blocking other threads. It's intentionally low-level;
# if you hooked up the PGconn#socket to some kind of reactor, you
# could make this much nicer.

TIMEOUT = 5.0 # seconds to wait for an async operation to complete
CONN_OPTS = {
    :host     => 'localhost',
    :dbname   => 'test',
    :user     => 'jrandom',
    :password => 'banks!stealUR
我建议您阅读 PQconnectStart 函数和 
我以前没有使用过 EventMachine，但如果它允许您注册一个套接字并在其变得可读/可写时进行回调，我认为将数据库调用集成到其中会相当容易。
我一直想使用  中的想法Ilya Grigorik 的文章介绍了使用 Fibers 清理事件代码，使异步 API 更易于使用，但这还有很长的路要走。我确实有未结票< /a> 如果您有兴趣/有动力自己做的话，请跟踪它。
,
}

# Print 'x' continuously to demonstrate that other threads aren't
# blocked while waiting for the connection, for the query to be sent,
# for results, etc. You might want to sleep inside the loop or 
# comment this out entirely for cleaner output.
progress_thread = Thread.new { loop { print 'x' } }

# Output progress messages
def output_progress( msg )
    puts "\n>>> #{msg}\n"
end

# Start the connection
output_progress "Starting connection..."
conn = PGconn.connect_start( CONN_OPTS ) or 
    abort "Unable to create a new connection!"
abort "Connection failed: %s" % [ conn.error_message ] if
    conn.status == PGconn::CONNECTION_BAD

# Now grab a reference to the underlying socket so we know when the
# connection is established
socket = IO.for_fd( conn.socket )

# Track the progress of the connection, waiting for the socket to 
# become readable/writable before polling it
poll_status = PGconn::PGRES_POLLING_WRITING
until poll_status == PGconn::PGRES_POLLING_OK ||
      poll_status == PGconn::PGRES_POLLING_FAILED

    # If the socket needs to read, wait 'til it becomes readable to
    # poll again
    case poll_status
    when PGconn::PGRES_POLLING_READING
        output_progress "  waiting for socket to become readable"
        select( [socket], nil, nil, TIMEOUT ) or
            raise "Asynchronous connection timed out!"

    # ...and the same for when the socket needs to write
    when PGconn::PGRES_POLLING_WRITING
        output_progress "  waiting for socket to become writable"
        select( nil, [socket], nil, TIMEOUT ) or
            raise "Asynchronous connection timed out!"
    end

    # Output a status message about the progress
    case conn.status
    when PGconn::CONNECTION_STARTED
        output_progress "  waiting for connection to be made."
    when PGconn::CONNECTION_MADE
        output_progress "  connection OK; waiting to send."
    when PGconn::CONNECTION_AWAITING_RESPONSE
        output_progress "  waiting for a response from the server."
    when PGconn::CONNECTION_AUTH_OK
        output_progress "  received authentication; waiting for " +
                        "backend start-up to finish."
    when PGconn::CONNECTION_SSL_STARTUP
        output_progress "  negotiating SSL encryption."
    when PGconn::CONNECTION_SETENV
        output_progress "  negotiating environment-driven " +
                        "parameter settings."
    end

    # Check to see if it's finished or failed yet
    poll_status = conn.connect_poll
end

abort "Connect failed: %s" % [ conn.error_message ] unless 
    conn.status == PGconn::CONNECTION_OK

output_progress "Sending query"
conn.send_query( "SELECT * FROM pg_stat_activity" )

# Fetch results until there aren't any more
loop do
    output_progress "  waiting for a response"

    # Buffer any incoming data on the socket until a full result 
    # is ready. 
    conn.consume_input
    while conn.is_busy
        select( [socket], nil, nil, TIMEOUT ) or
            raise "Timeout waiting for query response."
        conn.consume_input
    end

    # Fetch the next result. If there isn't one, the query is 
    # finished
    result = conn.get_result or break

    puts "\n\nQuery result:\n%p\n" % [ result.values ]
end

output_progress "Done."
conn.finish

if defined?( progress_thread )
    progress_thread.kill
    progress_thread.join
end

我建议您阅读 PQconnectStart 函数和

我以前没有使用过 EventMachine，但如果它允许您注册一个套接字并在其变得可读/可写时进行回调，我认为将数据库调用集成到其中会相当容易。

我一直想使用 中的想法Ilya Grigorik 的文章介绍了使用 Fibers 清理事件代码，使异步 API 更易于使用，但这还有很长的路要走。我确实有未结票< /a> 如果您有兴趣/有动力自己做的话，请跟踪它。

Answer 2

是的，您可以从 goliath 以非阻塞方式访问 postgres。我有同样的需求，并将这个概念证明放在一起： https://github.com/levicook /goliath-postgres-spike

Answer 3

我（不再）非常熟悉 Pg，但我还没有听说任何流行的数据库可以异步连接。因此，您仍然需要在查询期间保持与数据库的连接。因此，您仍然需要阻止堆栈中的某些位置。

根据您的应用程序，您可能已经以最佳方式进行了。

但是，当您正在处理某种轮询应用程序（同一客户端在短时间内发送大量请求）并且更重要的是获取响应，即使它是空的，那么您可以编写一个 ruby​​ Fiber 或完整的线程或进程，其寿命较长，代理对数据库的查询并缓存结果。

例如：请求来自客户端 A。Goliath 应用程序使用一些唯一 ID 处理对数据库进程的查询，并使用“尚无数据”响应查询。 DB进程完成查询并将结果与​​ID一起保存到缓存中。当下一个请求来自同一客户端时，Goliath 发现它已经有等待的查询结果，从缓存中删除结果并响应客户端。同时，它会与数据库进程安排下一个查询，以便更快准备好。如果下一个请求在上一个请求完成之前到达，则不会安排新的查询（不会增加查询）。

这样，您的响应就可以快速且无阻塞，同时仍然可以尽快提供来自数据库的新数据。当然，它们可能与实际数据有点不同步，但同样，根据应用程序，这可能不是问题。

Answer 4

这个想法是结合使用数据库（Postgresql）的异步适配器和事件Web服务器（Goliath）来获得性能。 Mike Perham 去年为 Rails 2.3 编写了一个 PG activerecord 适配器。也许你可以用它。

另一个例子是，Ilya Grigorik 发布了异步 Rails 堆栈的此演示。在本例中，事件服务器是 Thin，数据库是 Mysql。安装演示并在使用或不使用 EM 感知驱动程序的情况下尝试基准测试。差异是巨大的。