Haskell中的并发数据库连接池

Question

我是一名学习Haskell的Java程序员.
我在一个小型网络应用程序上工作,该应用程序使用Happstack并通过HDBC与数据库进行通信.

我编写了select和exec函数,我就像这样使用它们:

module Main where

import Control.Exception (throw)

import Database.HDBC
import Database.HDBC.Sqlite3 -- just for this example, I use MySQL in production

main = do
    exec "CREATE TABLE IF NOT EXISTS users (name VARCHAR(80) NOT NULL)" []

    exec "INSERT INTO users VALUES ('John')" []
    exec "INSERT INTO users VALUES ('Rick')" []

    rows <- select "SELECT name FROM users" []

    let toS x = (fromSql x)::String
    let names = map (toS . head) rows

    print names

你看,很简单.有查询,参数和结果.
连接创建和提交/回滚内容隐藏在select和exec中.
这很好,我不想在我的"逻辑"代码中关心它.

exec :: String -> [SqlValue] -> IO Integer
exec query params = withDb $ \c -> run c query params

select :: String -> [SqlValue] -> IO [[SqlValue]]
select query params = withDb $ \c -> quickQuery' c query params

withDb :: (Connection -> IO a) -> IO a
withDb f = do
    conn <- handleSqlError $ connectSqlite3 "users.db"
    catchSql
        (do r <- f conn
            commit conn
            disconnect conn
            return r)
        (\e@(SqlError _ _ m) -> do
            rollback conn
            disconnect conn
            throw e)

坏点:

• 始终为每个呼叫创建一个新连接 - 这会在重负载时导致性能下降
• DB url"users.db"是硬编码的 - 我不能在没有编辑的情况下在其他项目中重用这些功能

问题1:如何引入具有一些已定义(最小,最大)并发连接数的连接池,以便在select/exec调用之间重用连接？

问题2:如何使"users.db"字符串可配置？(如何将其移动到客户端代码？)

它应该是一个透明的功能:用户代码不应该要求显式连接处理/释放.

Answer 1

的资源池包提供了可用于数据库连接池高性能资源池.例如:

import Data.Pool (createPool, withResource)

main = do
    pool <- createPool newConn delConn 1 10 5
    withResource pool $ \conn -> doSomething conn

创建一个包含1个子池和最多5个连接的数据库连接池.每个连接在被销毁之前都可以空闲10秒.

Answer 2

问题2:我从未使用过HDBC,但我可能会写这样的东西.

trySql :: Connection -> (Connection -> IO a) -> IO a
trySql conn f = handleSql catcher $ do
    r <- f conn
    commit conn
    return r
  where catcher e = rollback conn >> throw e

打开Connection函数外部的某个位置,不要在函数内断开它.

问题1:嗯,连接池似乎难以实现......

import Control.Concurrent
import Control.Exception

data Pool a =
    Pool { poolMin :: Int, poolMax :: Int, poolUsed :: Int, poolFree :: [a] }

newConnPool low high newConn delConn = do
    cs <- handleSqlError . sequence . replicate low newConn
    mPool <- newMVar $ Pool low high 0 cs
    return (mPool, newConn, delConn)

delConnPool (mPool, newConn, delConn) = do
    pool <- takeMVar mPool
    if length (poolFree pool) /= poolUsed pool
      then putMVar mPool pool >> fail "pool in use"
      else mapM_ delConn $ poolFree pool

takeConn (mPool, newConn, delConn) = modifyMVar mPool $ \pool ->
    case poolFree pool of
        conn:cs ->
            return (pool { poolUsed = poolUsed pool + 1, poolFree = cs }, conn)
        _ | poolUsed pool < poolMax pool -> do
            conn <- handleSqlError newConn
            return (pool { poolUsed = poolUsed pool + 1 }, conn)
        _ -> fail "pool is exhausted"

putConn (mPool, newConn, delConn) conn = modifyMVar_ mPool $ \pool ->
    let used = poolUsed pool in
    if used > poolMin conn
      then handleSqlError (delConn conn) >> return (pool { poolUsed = used - 1 })
      else return $ pool { poolUsed = used - 1, poolFree = conn : poolFree pool }

withConn connPool = bracket (takeConn connPool) (putConn conPool)

你可能不应该逐字逐句,因为我甚至没有编译测试它(并且fail有非常不友好的),但想法是做类似的事情

connPool <- newConnPool 0 50 (connectSqlite3 "user.db") disconnect

并connPool根据需要传球.