为什么使用通道 sub/unsub 的以下代码会出现内存泄漏？答案

【问题标题】：Why do I have memory leak for the following code with channel sub/unsub?为什么使用通道 sub/unsub 的以下代码会出现内存泄漏？
【发布时间】：2023-03-06 17:04:01
【问题描述】：

我正在使用[org.clojure/clojure "1.10.1"],[org.clojure/core.async "1.2.603"] 和最新的 Amazon Corretto 11 JVM，如果它们有任何关系的话。

以下代码是生产中使用的代码的简化版本，它确实会导致内存泄漏。我不知道为什么会发生这种情况，但怀疑这可能是由于频道的子/取消子。谁能帮助指出我的代码可能出错的地方或如何修复内存泄漏？

(ns test-gc.core
  (:require [clojure.core.async :as a :refer [chan put! close! <! go >! go-loop timeout]])
  (:import [java.util UUID]))

(def global-msg-ch (chan (a/sliding-buffer 200)))

(def global-msg-pub (a/pub global-msg-ch :id))

(defn io-promise []
  (let [id (UUID/randomUUID)
        ch (chan)]
    (a/sub global-msg-pub id ch)
    [id (go
          (let [x (<! ch)]
            (a/unsub global-msg-pub id ch)
            (:data x)))]))

(defn -main []
  (go-loop []
    (<! (timeout 1))
    (let [[pid pch] (io-promise)
          cmd {:id   pid
               :data (rand-int 1E5)}]
      (>! global-msg-ch cmd)
      (println (<! pch)))
    (recur))
  (while true
    (Thread/yield)))

例如，快速堆转储提供以下统计信息：

按实例数分类
- java.util.LinkedList 5,157,128 (14.4%)
- java.util.concurrent.atomic.AtomicReference 3,698,382 (10.3%)
- clojure.lang.Atom 3,094,279 (8.6%)
- ...
按实例大小分类
- java.lang.Object[] 210,061,752 B (13.8%)
- java.util.LinkedList 206,285,120 B (13.6%)
- clojure.lang.Atom 148,525,392 B (9.8%)
- clojure.core.async.impl.channels.ManyToManyChannel 132,022,336 B (8.7%)
- ...

【问题讨论】：

这段代码的目标是什么？
@AlanThompson 我承认这是一个糟糕的代码/设计。但主要目标是处理 API 调用的异步响应。客户端的请求被分发到多个模块，完成后，带有额外数据的完成消息将发送到global-msg-ch，然后为每个请求检索（您可能猜到，每个请求都标记为UUID） .
你有什么样的内存泄漏？你能粘贴内存不足异常吗？
@rcastellcastell 它只是说Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "async-dispatch-1" alike。

标签： clojure core.async

【解决方案1】：

我终于知道为什么了。通过查看源代码，我们得到以下片段：

(defn pub
  "Creates and returns a pub(lication) of the supplied channel, ..."
  ...
     (let [mults (atom {}) ;;topic->mult
           ensure-mult (fn [topic]
                         (or (get @mults topic)
                             (get (swap! mults
                                         #(if (% topic) % (assoc % topic (mult (chan (buf-fn topic))))))
                                  topic)))
           p (reify
              Mux
              (muxch* [_] ch)

              Pub
              (sub* [p topic ch close?]
                    (let [m (ensure-mult topic)]
                      (tap m ch close?)))
              (unsub* [p topic ch]
                      (when-let [m (get @mults topic)]
                        (untap m ch)))
              (unsub-all* [_] (reset! mults {}))
              (unsub-all* [_ topic] (swap! mults dissoc topic)))]
       ...
       p)))

我们可以看到mults 存储了所有topic，因此如果我们不清除它，它将单调增加。我们可能会添加类似(a/unsub-all* global-msg-pub pid) 的内容来解决这个问题。

【讨论】：