SQLException 未传播到主线程 - Clojure答案

【问题标题】：SQLException is not propagated to the main thread - ClojureSQLException 未传播到主线程 - Clojure
【发布时间】：2020-01-28 17:00:19
【问题描述】：

我在对 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 的新手，希望有任何关于此的 cmets

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

(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。

【问题讨论】：

标签： postgresql concurrency clojure jvm future

【解决方案1】：

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

【讨论】：

我已将连接池的配置方式添加到我的原始问题中。您能否详细说明我如何将您提到的内容整合到我的方法中？

【解决方案2】：

在做了更多研究之后，我意识到当在 Clojure 中使用 future 块时，所有异常都会被吞没，除非 future 得到 取消引用。因此，我可以通过如下方式更新我的 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])))

【讨论】：