使用 R 池包连接到 PostgreSQL 数据库答案

【问题标题】：Reconnect to PostgreSQL database with R's pool package使用 R 池包连接到 PostgreSQL 数据库
【发布时间】：2019-03-04 18:00:12
【问题描述】：

我有一个使用 R plumber 构建的 API，它使用 RPostgreSQL 和 pool 连接到 PostgreSQL 数据库（尽管如果我使用的是 Shiny 应用程序，这也适用）：

# create the connection pool
pool <- dbPool(
  drv = PostgreSQL(),
  host = Sys.getenv("DB_HOST"),
  port = 5432,
  dbname = "db",
  user = Sys.getenv("DB_USER"),
  password = Sys.getenv("DB_PASSWORD")
)

# start the API
pr <- plumb("plumber.R")

# on stop, close the pool
pr$registerHooks(
  list("exit" = function() { poolClose(pool) })
)

我想每天导入新数据。最简单的方法是创建一个新数据库并将其推广到生产环境：

CREATE DATABASE db_new;
-- create the tables
-- bulk-insert the data
SELECT pg_terminate_backend (pid) FROM pg_stat_activity WHERE datname = 'db';
DROP DATABASE db;
ALTER DATABASE db_new RENAME TO db;

这速度很快，并且可以最大限度地减少停机时间。问题是 pool 然后失去与数据库的连接并且不会自动尝试重新连接：

> tbl(pool, "users")
Error in postgresqlExecStatement(conn, statement, ...) : 
  RS-DBI driver: (could not Retrieve the result : FATAL:  terminating connection due to administrator command
server closed the connection unexpectedly
    This probably means the server terminated abnormally
    before or while processing the request.
)

即使我不是每天都更换数据库，数据库服务器偶尔也会重新启动，这也会导致我的应用程序崩溃。重新连接似乎不是池、RPostgreSQL 或 DBI 的功能。有谁知道解决这个问题的方法吗？

【问题讨论】：

略相关：github.com/rstudio/pool/issues/72（虽然是去年发起的，但遗憾的是还没有受到关注）。
我们在谈论一个闪亮的应用程序吗？
@JonnyCrunch，它被列为管道工 API 端点，可能与闪亮的应用程序一起提供（例如，在 RStudio Connect 中），但有点不同。
我的建议是“observerEvent”shiny 样式，应该链接到reactive 检查连接是否丢失，但我不知道这是否适用于此类问题
为什么不在你的连接字符串中连接到不同的数据库或公共数据库，当你发出终止命令时它不会退出你自己的会话？

标签： r postgresql r-dbi rpostgresql plumber

【解决方案1】：

我最近遇到了类似的问题，因为当实例的wait_timeout 超出时，MySQL 连接被关闭。我遇到了your post on RStudio Community，并受到您的解决方案的启发。如果您仍在使用它，并且正在寻找一种在包装您使用的实际功能时避免额外查询的解决方案，这里是一个演示我想出的东西的代表，以及一个证明它有效的示例：

library(dplyr, warn.conflicts = FALSE)
library(pool)
library(RMariaDB)

generate_safe_query <- function(pool) {
  function(db_function, ...) {
    tryCatch({
      db_function(pool, ...)
    }, error = function(e) {
      if (grepl("Lost connection to MySQL server during query", e$message)) {
        # Preserve `validationInterval` so that it can be restored
        validation_interval <- pool$validationInterval
        # Trigger destruction of dead connection
        pool$validationInterval <- 0
        refreshed_connection <- poolCheckout(pool)
        poolReturn(refreshed_connection)
        # Restore original `validationInterval`
        pool$validationInterval <- validation_interval
        # Execute the query with the new connection
        db_function(pool, ...)
      } else {
        # Unexpected error
        stop(e)
      }
    })
  }
}

mysql_pool <- dbPool(MariaDB(),
                     host = "127.0.0.1",
                     username = "root",
                     password = "",
                     dbname = "test")

safe_query <- generate_safe_query(mysql_pool)

# Works
safe_query(tbl, "notes")
#> # Source:   table<notes> [?? x 2]
#> # Database: mysql 8.0.15 [root@127.0.0.1:/test]
#>      id note 
#>   <int> <chr>
#> 1     1 NOTE1

# Set the `wait_timeout` to 5 seconds for this session
invisible(safe_query(dbExecute, "SET SESSION wait_timeout = 5"))

# Wait longer than `wait_timeout` to trigger a disconnect
Sys.sleep(6)

# Still works; warning will appear notifying that connection was
# destroyed and replaced with a new one
safe_query(tbl, "notes")
#> Warning: It wasn't possible to activate and/or validate the object. Trying
#> again with a new object.
#> # Source:   table<notes> [?? x 2]
#> # Database: mysql 8.0.15 [root@127.0.0.1:/test]
#>      id note 
#>   <int> <chr>
#> 1     1 NOTE1

safe_query(poolClose)
# Or, equivalently: 
# poolClose(mysql_pool)

^{由reprex package (v0.3.0) 于 2019 年 5 月 30 日创建}

generate_safe_query 返回的函数将适用于任何数据库查询函数（例如dbExecute、dbGetQuery 等）。显然，您需要更新它匹配的错误消息以满足您的需要。

我还打开了我自己的 Community topic 选项，我认为应该包含在 dbPool 中，这将减少对此类解决方法的需求。

【讨论】：

谢谢，太好了。实际上，我最终用 SQLAlchemy 重写了我在 Python 中处理的 API，但我还有其他人可以从这个解决方案中受益。
很高兴听到您认识它可以帮助的其他人！我还在我链接到的社区主题上发布了解决方案，并使其更加通用，因为它可以匹配多个错误消息。

【解决方案2】：

我正在使用具有以下功能的普通 DBI（无池）以始终为 DBI 调用提供活动连接（例如 DBI::dbExistsTable(rdsConnect(), "mytable")）。

#' Connect returns a database connection.
#' Retrieves the connection parameters from configuration.
#' 
#' FIXME: dbIsValid is not implemented
#' https://github.com/tomoakin/RPostgreSQL/issues/76
#' workaround implemented with isPostgresqlIdCurrent()
#' @return rds allocated connection
rdsConnect <- function() {
  if (!((exists("rds") && (isPostgresqlIdCurrent(rds))))) {
    source('./config.R', local = TRUE)
    print("New PostgreSQL connection")
    rds <<- DBI::dbConnect(RPostgreSQL::PostgreSQL(),
      dbname = rds_params("rds_database"),
      host = rds_params("rds_host"),
      user = rds_params("rds_user"),
      password = rds_params("rds_password")
    )
  } else print("Valid PostgreSQL connection")
  return(rds)
}

【讨论】：