Datalevin多线程事务异常问题排查与安全实现咨询

问题背景 #

使用Datalevin 0.7.8版本编写Clojure程序，多线程执行时出现并发异常，尝试不同连接创建方式后仍有问题，甚至损坏数据库文件。

原代码 #

(ns datalevintest.core
  (:require [datalevin.core :as dc]))

(def store (System/getenv "DBSTORE"))

(def conn (datalevin.core/get-conn store {} {:auto-entity-time? true :validate-data? true}))

(defn -main [& _]
  (dotimes [i 5]
    (future
      (locking ::println (println "Starting thread"))
      (try
        (dotimes [j 100]
          (dc/transact! conn [{:i+j (+ i j)}])
          (dc/with-transaction [tx-conn conn]
            (dc/transact! tx-conn [{:i*j (* i j)}]))
          (dc/q '[:find (pull ?e [*]) :in $ ?id :where [?e :db/id ?id]]
                (dc/db conn) 2345))
        (catch Throwable t (.printStackTrace t))
        (finally (println "Thread" i "done"))))
  (println "END"))

出现的异常情况 #

多线程共用连接时的异常

1. 常见异常：

clojure.lang.ExceptionInfo: Fail to transact to LMDB: "Transaction is not in ready state" {}
at datalevin.binding.java.LMDB.transact_kv(java.clj:484)
...（省略部分栈信息）

2. 少见异常：

clojure.lang.ExceptionInfo: Fail to get-first: nil {:dbi "datalevin/eav", :k-range [:all-back], :k-type :eav, :v-type :id}
at datalevin.scan$get_first.invokeStatic(scan.cljc:233)
...（省略部分栈信息）

每个线程单独创建连接的异常

将连接创建逻辑移入future中使用create-conn时，出现以下异常，且会损坏数据库文件：

java.lang.NullPointerException: Cannot read field "e"
at datalevin.storage.Store.init_max_eid(storage.cljc:302)
...（省略部分栈信息）

代码错误与解决方案 #

核心错误 #

1. Datalevin连接并非线程安全：尽管get-conn会复用同一目录的连接，但底层LMDB事务不支持多线程并发操作。跨线程共用同一个连接时，多个线程同时操作会导致事务状态冲突，触发"Transaction is not in ready state"这类异常。
2. 多线程并发初始化数据库会损坏文件：多个线程同时调用create-conn操作同一个数据库文件时，会出现写入冲突，破坏LMDB的文件结构，导致后续无法启动。

安全的实现方式 #

方案1：使用连接池（推荐）

借助Clojure连接池库管理连接，每个线程从池获取独立连接，使用后归还：

(ns datalevintest.core
  (:require [datalevin.core :as dc]
            [hikari-cp.core :as hikari]))

(def store (System/getenv "DBSTORE"))

(def pool-config
  {:adapter "datalevin"
   :jdbc-url (str "datalevin://" store)
   :maximum-pool-size 5})

(def pool (hikari/make-datasource pool-config))

(defn get-conn-from-pool []
  (dc/get-conn store {} {:auto-entity-time? true :validate-data? true}))

(defn -main [& _]
  (dotimes [i 5]
    (future
      (locking ::println (println "Starting thread"))
      (let [conn (get-conn-from-pool)]
        (try
          (dotimes [j 100]
            (dc/transact! conn [{:i+j (+ i j)}])
            (dc/with-transaction [tx-conn conn]
              (dc/transact! tx-conn [{:i*j (* i j)}]))
            (dc/q '[:find (pull ?e [*]) :in $ ?id :where [?e :db/id ?id]]
                  (dc/db conn) 2345))
          (catch Throwable t (.printStackTrace t))
          (finally
            (dc/close conn)
            (println "Thread" i "done")))))
  (println "END"))

方案2：单连接+全局锁

对连接的所有操作加全局锁，确保同一时间只有一个线程操作连接：

(ns datalevintest.core
  (:require [datalevin.core :as dc]))

(def store (System/getenv "DBSTORE"))

(def conn (dc/get-conn store {} {:auto-entity-time? true :validate-data? true}))
(def conn-lock (Object.))

(defn -main [& _]
  (dotimes [i 5]
    (future
      (locking ::println (println "Starting thread"))
      (try
        (dotimes [j 100]
          (locking conn-lock
            (dc/transact! conn [{:i+j (+ i j)}])
            (dc/with-transaction [tx-conn conn]
              (dc/transact! tx-conn [{:i*j (* i j)}]))
            (dc/q '[:find (pull ?e [*]) :in $ ?id :where [?e :db/id ?id]]
                  (dc/db conn) 2345)))
        (catch Throwable t (.printStackTrace t))
        (finally (println "Thread" i "done"))))
  (println "END"))

关键注意事项 #

• 永远不要跨线程共享同一个Datalevin连接，每个线程应使用独立连接。
• 确保数据库初始化操作是单线程完成的，避免多线程并发调用create-conn。
• 使用连接池时，线程结束后必须关闭连接，避免资源泄漏。

内容的提问来源于stack exchange，提问作者erdos