在 Lparallel 库中使用队列 (Common Lisp)

lparallel 库中队列的基本讨论位于https://z0ltan.wordpress.com/2016/09/09/basic-concurrency-and-parallelism-in-common-lisp-part-4a-parallelism-using-lparallel-fundamentals/#channels https://z0ltan.wordpress.com/2016/09/09/basic-concurrency-and-parallelism-in-common-lisp-part-4a-parallelism-using-lparallel-fundamentals/#channels说队列“启用工作线程之间的消息传递”。下面的测试使用共享队列来协调主线程和从属线程，其中主线程在退出之前只是等待从属线程完成：

(defun foo (q)
  (sleep 1)
  (lparallel.queue:pop-queue q))  ;q is now empty

(defun test ()
  (setf lparallel:*kernel* (lparallel:make-kernel 1))
  (let ((c (lparallel:make-channel))
        (q (lparallel.queue:make-queue)))
    (lparallel.queue:push-queue 0 q)
    (lparallel:submit-task c #'foo q)
    (loop do (sleep .2)
             (print (lparallel.queue:peek-queue q))
          when (lparallel.queue:queue-empty-p q)
            do (return)))
  (lparallel:end-kernel :wait t))

这按预期产生输出：

* (test)

0
0
0
0
NIL
(#<SB-THREAD:THREAD "lparallel" FINISHED values: NIL {10068F2B03}>)

我的问题是我是否正确或完全使用了 lparallel 的队列功能。队列似乎只是使用全局变量来保存线程共享对象的替代品。使用队列的设计优势是什么？为每个提交的任务分配一个队列（假设任务需要通信）通常是一种好的做法吗？感谢您提供更深入的见解。

多线程工作是通过管理对可变对象的并发访问来完成的 共享状态，即你有一个公共数据结构的锁， 每个线程都对其进行读取或写入。

然而，建议尽量减少数据的数量 同时访问。队列是一种将工作人员与每个工作人员解耦的方法 其他，通过让每个线程管理其本地状态并交换数据 只能通过消息；这是线程安全的，因为访问 队列由以下控制锁和状况 变量 https://en.wikipedia.org/wiki/Monitor_(synchronization).

你在主线程中正在做的是polling当队列 是空的;这可能有效，但这会适得其反，因为队列 用作同步机制，但在这里你正在做 自己同步。

(ql:quickload :lparallel)
(defpackage :so (:use :cl
                      :lparallel
                      :lparallel.queue
                      :lparallel.kernel-util))
(in-package :so)

让我们改变foo这样它就有两个队列，一个用于传入 一个用于请求，一个用于回复。在这里，我们进行一个简单的变换 正在发送的数据，对于每条输入消息，只有一个 输出消息，但情况并不总是如此。

(defun foo (in out)
  (push-queue (1+ (pop-queue in)) out))

Change test这样控制流仅基于对队列的读/写：

(defun test ()
  (with-temp-kernel (1)
    (let ((c (make-channel))
          (foo-in (make-queue))
          (foo-out (make-queue)))
      (submit-task c #'foo foo-in foo-out)
      ;; submit data to task (could be blocking)
      (push-queue 0 foo-in)
      ;; wait for message from task (could be blocking too)
      (pop-queue foo-out))))

但是，如果有多个任务正在运行，如何避免测试中的轮询呢？您是否不需要不断检查其中任何一项何时完成，以便可以将更多工作推送到队列中？

您可以使用不同的并发机制，类似于listen http://man7.org/linux/man-pages/man2/listen.2.html and 轮询/epoll http://man7.org/linux/man-pages/man7/epoll.7.html，您可以在其中观看多个 事件源并在其中之一准备好时做出反应。有类似的语言Go (select https://tour.golang.org/concurrency/5) and Erlang (receive http://erlangbyexample.org/send-receive） 在哪里 这是很自然的表达。在 Lisp 方面，Calispel https://www.cliki.net/Calispel库提供了类似的交替机制（pri-alt and fair-alt）。例如，以下内容取自 Calispel 的测试代码：

(pri-alt ((? control msg)
          (ecase msg
            (:clean-up (setf cleanup? t))
            (:high-speed (setf slow? nil))
            (:low-speed (setf slow? t))))
         ((? channel msg)
          (declare (type fixnum msg))
          (vector-push-extend msg out))
         ((otherwise :timeout (if cleanup? 0 nil))
          (! reader-results out)
          (! thread-expiration (bt:current-thread))
          (return)))

如果是并行，没有这样的机制，但是只要您使用标识符标记消息，您就可以仅使用队列走得很远。

如果您需要尽快做出反应either a task t1 or t2给出结果，然后使这两个任务写入同一结果通道：

(let ((t1 (foo :id 1 :in i1 :out res))
      (t2 (bar :id 2 :in i2 :out res)))
   (destructuring-bind (id message) (pop-queue res)
     (case id
       (1 ...)
       (2 ...))))

如果您需要同步代码t1 and t2发出结果，让他们写入不同的通道：

(let ((t1 (foo :id 1 :in i1 :out o1))
      (t2 (bar :id 2 :in i2 :out o2)))
   (list (pop-queue o1)
         (pop-queue o2)))