SQLException 未传播到主线程 - Clojure

Question

我在运行针对 Postgres 的查询时遇到问题。如果查询时间超过 10 秒左右，将出现如下所示的 SQLException

com.mchange.v2.c3p0.impl.NewPooledConnection@33d2d0dc handling a throwable.: org.postgresql.util.PSQLException: An I/O error occurred while sending to the backend.
at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:326)
at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:428)
at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:354)
at org.postgresql.jdbc.PgPreparedStatement.executeWithFlags(PgPreparedStatement.java:169)
at org.postgresql.jdbc.PgPreparedStatement.executeQuery(PgPreparedStatement.java:117)
at com.mchange.v2.c3p0.impl.NewProxyPreparedStatement.executeQuery(NewProxyPreparedStatement.java:1418)
at clojure.java.jdbc$execute_query_with_params.invokeStatic(jdbc.clj:993)
at clojure.java.jdbc$execute_query_with_params.invoke(jdbc.clj:987)
at clojure.java.jdbc$db_query_with_resultset_STAR_.invokeStatic(jdbc.clj:1016)
at clojure.java.jdbc$db_query_with_resultset_STAR_.invoke(jdbc.clj:996)
at clojure.java.jdbc$reducible_query$reify__15393.reduce(jdbc.clj:1272)
at clojure.core$reduce.invokeStatic(core.clj:6544)
at clojure.core$reduce.invoke(core.clj:6527)
at cenx.constellation.common$reducible__GT_chan$fn__978.invoke(common.clj:103)
at clojure.core$binding_conveyor_fn$fn__4676.invoke(core.clj:1938)
at clojure.lang.AFn.call(AFn.java:18)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.EOFException
at org.postgresql.core.PGStream.receiveChar(PGStream.java:290)
at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:1963)
at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:300)
... 19 more

现在我只想能够通知我的主线程发生了某些事情。我的代码块如下

(defn func (
[coll name]
    (let [ch (async/chan 100)]
        (future (if (not (Thread/interrupted))
            (try
                (let [  _ (log/info "Start Reading from" name)
                        i (reduce (fn [i v]
                                    (when (zero? (mod i 10000))
                                        (log/debug "row[" i "]: " name))
                                    (if (nil? v)
                                        (log/warn "skipping")
                                    (async/>!! ch [i v]))
                                    (inc i))
                            0 coll)
                        _ (log/info "End Reading, read" i "rows from" name)]
                (async/close! ch))
            (catch SQLException se
                (log/info "SQLException " (.getMessage se))
                (throw (SQLException. "SQLException in my catch block")))
            (finally
                (log/info "Thread is not interrupted")))
        (throw (InterruptedException. "Thread got interrupted"))))
    ch)))

虽然在工作线程中捕获了异常，但主线程没有捕获我从 catch 块内部抛出的异常。我是 clojure 的新手，非常感谢对此的任何评论

编辑：我正在添加我用来创建和配置连接池的方法

(defn- pool [{:keys [subprotocol host port db user pool] :as _spec}] {:datasource (doto (ComboPooledDataSource.) (.setJdbcUrl (str "jdbc:" subprotocol "://" host ":" port "/" db)) (.setUser user) (.setPassword (get-postgres-password)) (.setInitialPoolSize (:min-size pool)) (.setMinPoolSize (:min-size pool)) (.setMaxPoolSize (:max-size pool)) (.setMaxConnectionAge (* 1 60 60)) (.setMaxIdleTime (* 1/2 60 60)) (.setMaxIdleTimeExcessConnections (* 5 60)) (.setIdleConnectionTestPeriod (* 10 60)))})

有关现有配置，请参阅 here。

Answer 1

要检索任何内部抛出的异常，请添加 ConnectionEventListener / StatementEventListener to your NewPooledConnection。

Answer 2

在做了更多研究后，我意识到在 Clojure 中使用 future 块时，所有异常都被吞没，除非 future 得到 derefed。所以我可以通过如下更新我的 defn 并稍后在另一个 defn 中取消引用 future 来解决我的问题在。

(defn func
([coll name]
(let [ch (async/chan 100)
future-f (future
               (let [i (reduce (fn [i v]
                                 (when (zero? (mod i 10000))
                                   (log/debug " row[" i "]: " name))

                                 (if (nil? v)
                                   (log/warn "skipping")
                                   (async/>!! ch [i v]))
                                 (inc i))
                               0 coll)
                     _ (log/info "End Reading, read" i "rows from" name)]
                 (async/close! ch)))]
 [ch future-f])))

SQLException 未传播到主线程 - Clojure

SQLException is not propagated to the main thread - Clojure

postgresql

concurrency

jvm

future

clojure