Блог

Я использую Goliath (который работает на eventmachine) и postgres gem pg , в настоящее время я использую pg gem блокирующим способом: conn.exec('SELECT * FROM products') (например) и мне интересно, есть ли лучший способ подключиться к базе данных postgres?

pg Библиотека обеспечивает полную поддержку асинхронного API PostgreSQL. Я добавил пример того, как его использовать, в 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$',
}

# 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 "nnQuery result:n%pn" % [ result.values ]
end

output_progress "Done."
conn.finish

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

Я бы рекомендовал вам прочитать документацию по функции PQconnectStart и раздел «Асинхронная обработка команд» руководства PostgreSQL, а затем сравнить это с примером выше.

Я раньше не использовал EventMachine, но если он позволяет вам регистрировать сокет и обратные вызовы, когда он становится доступным для чтения / записи, я бы подумал, что было бы довольно легко интегрировать в него вызовы базы данных.

Я собирался использовать идеи из статьи Ильи Григорика об использовании волокон для очистки событийного кода, чтобы упростить использование async API, но это далеко не так. У меня есть открытый тикет, чтобы отслеживать это, если вы заинтересованы / мотивированы сделать это самостоятельно.

Да, вы можете получить доступ к postgres неблокирующим способом из goliath. У меня была такая же потребность, и я собрал это доказательство концепции: https://github.com/levicook/goliath-postgres-spike

Я не (больше) хорошо знаком с Pg, но я не слышал, чтобы какая-либо популярная база данных могла синхронизировать соединения. Таким образом, вам все равно нужно поддерживать соединение с базой данных на время выполнения запроса. Поэтому вам все равно нужно заблокировать некоторые места в стеке.

В зависимости от вашего приложения вы, возможно, уже делаете это наилучшим возможным способом.

Но когда вы имеете дело с каким-то приложением для опроса (где один и тот же клиент отправляет множество запросов за короткое время), и более важно получить ответ, даже если он пустой, тогда вы могли бы написать ruby Fiber или завершенный поток или процесс, который долговечен и отправляет запросы прокси в БД и кэширует результаты.

Например: запрос поступает от клиента A. Приложение Goliath обрабатывает запрос к процессу базы данных с некоторым уникальным идентификатором и отвечает на запрос с «пока нет данных». Процесс базы данных завершает запрос и сохраняет результаты в кэше с идентификатором. Когда поступает следующий запрос от того же клиента, Goliath видит, что у него уже есть ожидающие результаты запроса, удаляет результаты из кэша и отвечает клиенту. В то же время он планирует следующий запрос с помощью процесса DB, чтобы он был готов раньше. Если следующий запрос поступает до завершения предыдущего, новый запрос не планируется (без умножения запросов).

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

Идея состоит в том, чтобы использовать асинхронный адаптер для базы данных (Postgresql) в сочетании с веб-сервером evented (Goliath) для повышения производительности. Майк Перхам написал адаптер PG activerecord для Rails 2.3 в прошлом году. Может быть, вы сможете это использовать.

В качестве другого примера Илья Григорик выпустил эту демонстрационную версию стека async Rails. В этом случае сервер событий является тонким, а базой данных является Mysql. Установите демонстрационную версию и попробуйте тест с драйвером, поддерживающим EM, и без него. Разница разительная.