如何在Clojure core.async UDP读取包装中正确停止接收线程

Great question! Stopping a blocking I/O thread gracefully is always a bit tricky, but with Clojure's tools and some Java I/O/NIO tricks, you've got a few solid, idiomatic options. Let's break them down:

Option 1: Use Thread Interruption (for DatagramSocket) #

Blocking calls like DatagramSocket.receive() will immediately throw an InterruptedException when the thread is interrupted. This is a clean way to signal the thread to shut down, and you can use this to exit your loop and clean up resources.

Here's a concrete example:

(ns udp.async-wrapper
  (:require [clojure.core.async :as async])
  (:import [java.net DatagramSocket DatagramPacket]))

(defn start-udp-receiver [port output-ch]
  (let [socket (DatagramSocket. port)
        buffer (byte-array 1024)
        packet (DatagramPacket. buffer (alength buffer))
        ;; Define the receiver thread with a loop that reads packets
        receiver-thread (Thread.
                         (fn []
                           (try
                             (loop []
                               ;; Block until a packet arrives
                               (.receive socket packet)
                               ;; Send the parsed packet to the async channel
                               (async/go
                                 (async/>! output-ch
                                           {:data (vec (.getData packet))
                                            :length (.getLength packet)
                                            :sender-address (.getAddress packet)
                                            :sender-port (.getPort packet)}))
                               ;; Reset the packet for the next receive
                               (.setLength packet (alength buffer))
                               (recur))
                             ;; Catch the interruption to handle shutdown
                             (catch InterruptedException _
                               (println "Receiver thread interrupted, initiating shutdown"))
                             ;; Always clean up resources, even if something goes wrong
                             (finally
                               (.close socket)
                               (async/close! output-ch)))))]
    ;; Start the thread and return a reference to it for later interruption
    (.start receiver-thread)
    receiver-thread))

;; --- Usage ---
(def packet-channel (async/chan))
(def my-receiver-thread (start-udp-receiver 5000 packet-channel))

;; To stop the receiver later:
(.interrupt my-receiver-thread)

How this works: #

• We store a reference to the receiver thread so we can interrupt it later.
• When .interrupt() is called, the blocking .receive() call throws InterruptedException, which we catch to trigger shutdown.
• The finally block ensures we close the socket and async channel to avoid resource leaks.

Option 2: Use Selector.wakeup() (for DatagramChannel with NIO) #

If you're using DatagramChannel with a Selector (for non-blocking I/O), you can't rely solely on thread interruption (though it works, Selector.wakeup() is more idiomatic for NIO). Here's how to do it with a stop flag and wakeup:

(ns udp.nio.async-wrapper
  (:require [clojure.core.async :as async])
  (:import [java.nio.channels DatagramChannel Selector SelectionKey]
           [java.net InetSocketAddress]))

(defn start-nio-udp-receiver [port output-ch]
  (let [channel (DatagramChannel/open)
        selector (Selector/open)
        buffer (java.nio.ByteBuffer/allocate 1024)
        ;; Atom to track if we should stop the loop
        stop? (atom false)]
    ;; Configure the channel for non-blocking mode and register with selector
    (.bind channel (InetSocketAddress. port))
    (.configureBlocking channel false)
    (.register channel selector SelectionKey/OP_READ)

    (let [receiver-thread (Thread.
                           (fn []
                             (try
                               (loop []
                                 (when-not @stop?
                                   ;; Block until ready to read, or woken up
                                   (.select selector)
                                   ;; Process all ready keys
                                   (doseq [key (.selectedKeys selector)]
                                     (when (.isReadable key)
                                       ;; Read the packet into the buffer
                                       (.receive channel buffer)
                                       (.flip buffer)
                                       ;; Convert buffer to a byte array for Clojure
                                       (let [data (byte-array (.remaining buffer))]
                                         (.get buffer data)
                                         (async/go (async/>! output-ch {:data data}))
                                         (.clear buffer))
                                       ;; Remove the key from the selected set
                                       (.remove (.selectedKeys selector) key)))
                                   (recur)))
                             (catch Exception e
                               (println "Receiver error occurred:" e))
                             (finally
                               (.close channel)
                               (.close selector)
                               (async/close! output-ch)))))]
      (.start receiver-thread)
      ;; Return a function to trigger shutdown (cleaner than passing thread refs)
      (fn []
        (reset! stop? true)
        (.wakeup selector)))))

;; --- Usage ---
(def nio-packet-channel (async/chan))
(def stop-receiver (start-nio-udp-receiver 5001 nio-packet-channel))

;; To stop the receiver later:
(stop-receiver)

How this works: #

• We use an atom stop? to track whether the loop should exit.
• When we want to stop, we set stop? to true and call .wakeup() on the selector, which immediately unblocks the .select() call.
• The loop checks stop? and exits, then the finally block cleans up all resources.

Bonus Best Practices #

• Never ignore InterruptedException: Always handle it to trigger proper shutdown (don't just catch and swallow it).
• Clean up in finally: This ensures sockets, channels, and async channels are closed even if an unexpected error occurs.
• Prefer returning a stop function: For the NIO example, returning a closure that handles stopping is cleaner than passing thread references around (avoids mutable state leaks).
• Alternative for DatagramSocket: You can also call .close() on the socket directly, which will make .receive() throw a SocketException. Interruption is more explicit, though, since it signals an intentional shutdown rather than an error.

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