我
目前正在尝试为以下方面找到一个lispy/工作解决方案的问题描述：
作业队列提供一组相等的（通过其代码）线程，其中包含它们应处理的任务。如果队列为空，线程将等待，直到创建新条目，但我也想提供一个干净的关闭。因此，即使在等待队列时，母线程也必须能够设置一些变量/来调用线程并告诉它们关闭。他们不直接遵守的唯一原因应该是线程当前正在评估任务，因此在任务完成之前忙碌/无法进行干净关闭。
我目前有两个解决方案，我不太相信：
(defparameter *kill-yourself* nil)
(defparameter *mutex* (sb-thread:make-mutex))
(defparameter *notify* (sb-thread:make-waitqueue))
#|the queue is thread safe|#
(defparameter *job-queue* (make-instance 'queue))
(defun fill-queue (with-data)
(fill-stuff-in-queue)
(sb-thread:with-mutex (*mutex*)
(sb-thread:condition-notify *notify*)))
#|solution A|#
(with-mutex (*mutex*)
(do ((curr-job nil))
(*kill-yourself* nil)
(if (is-empty *job-queue*)
(sb-thread:condition-wait *notify* *mutex*)
(progn
(setf curr-job (dequeue *job-queue*))
(do-stuff-with-job)))))
#|solution B|#
(defun helper-kill-yourself-p ()
(sb-thread:with-mutex (*mutex*)
*kill-yourself*))
(do ((job (dequeue-* *job-queue* :timeout 0)
(dequeue-* *job-queue* :timeout 0)))
((if (helper-kill-yourself-p)
t
(sb-thread:with-mutex (*mutex*)
(sb-thread:condition-wait *notify* *mutex*)
(if (helper-kill-yourself-p)
t
nil)))
(progn
nil))
(do-stuff-with-job))
这两个 do 循环都可用于启动线程。但是如果有多个线程，A 将无法真正工作（因为互斥锁会阻止发生任何并行操作），并且 B 解决方案看起来/非常脏，因为可能存在提取的作业为零的侧情况。此外，我并不真正相信停止条件，因为它太长而且似乎很复杂。
实现（do）循环的正确方法是什么，只要队列应该对队列提供的数据起作用，并且只要没有新数据并且只要它不应该关闭，它就可以休眠？最后但并非最不重要的一点是，必须能够在无限数量的多个并行线程中使用此（do）循环。
解决方案 A
是的，您对解决方案 A 的看法是正确的，互斥锁不会让工作成为并行执行。
解决方案 B
我认为do循环不是适合这项工作的工具。特别是，在您的代码 有可能从队列和线程中提取作业将退出而不执行它。这种情况是可能的，因为您取消排队之前应终止检查。还因为您在 do 的变量中定义了job块你忽略从dequeue返回的多个值，这也是不好的因为您无法有效地检查队列是否为空。同样在场景中检查线程是否应该在结束测试表单中停止do您必须获取*mutex*两次，以检查线程是否应停止并取消排队（或你可以发明奇怪的结束测试形式，它将完成循环体的工作）。
所以，话虽如此，你必须把所有的代码都放在do的身体里，然后将 var 和结束测试留空。这就是为什么我认为loop在这方面更好箱。
如果您必须使用do循环，您可以轻松地将loop身体包裹在其中，例如 (do nil (nil nil) *loop-body*) .
我的解决方案
(require :sb-concurrency)
(use-package :sb-concurrency)
(use-package :sb-thread)
(defparameter *kill-yourself* nil)
(defparameter *mutex* (make-mutex))
(defparameter *notify* (make-waitqueue))
#|the queue is thread safe|#
(defparameter *job-queue* (make-queue :name "job-queue"))
(defparameter *timeout* 10)
(defparameter *output-lock* (make-mutex))
(defun output (line)
(with-mutex (*output-lock*)
(write-line line)))
(defun fill-queue (with-data)
(enqueue with-data *job-queue*)
(with-mutex (*mutex*)
(condition-notify *notify*)))
(defun process-job (thread-name job)
(funcall job thread-name))
(defun run-worker (name)
(make-thread
(lambda ()
(output (format nil "starting thread ~a" name))
(loop (with-mutex (*mutex*)
(condition-wait *notify* *mutex* :timeout *timeout*)
(when *kill-yourself*
(output (format nil "~a thread quitting" name))
(return-from-thread nil)))
;; release *mutex* before starting the job,
;; otherwise it won't allow other threads wait for new jobs
;; you don't want to make 2 separate calls (queue-empty-p, dequeue)
;; since inbetween queue can become empty
(multiple-value-bind (job has-job) (dequeue *job-queue*)
(if has-job
(process-job name job)))))
:name name))
(defun stop-work ()
(with-mutex (*mutex*)
(setf *kill-yourself* t)
(condition-broadcast *notify*)))
(defun add-job (job)
;; no need to put enqueue in critical section
(enqueue job *job-queue*)
(with-mutex (*mutex*)
(condition-notify *notify*)))
(defun make-job (n)
(lambda (thread-name)
(loop for i upto 1000 collecting i)
(output (format nil "~a thread executes ~a job" thread-name n))))
(defun try-me ()
(run-worker "worker1")
(run-worker "worker2")
(loop for i upto 1000 do
(add-job (make-job i)))
(loop for i upto 2000 collecting i)
(stop-work))
在 REPL 中调用try-me应该会给你类似于以下输出的内容
starting thread worker1
worker1 thread executes 0 job
worker1 thread executes 1 job
worker1 thread executes 2 job
worker1 thread executes 3 job
starting thread worker2
worker2 thread executes 4 job
worker1 thread executes 5 job
worker2 thread executes 6 job
worker1 thread executes 7 job
worker1 thread executes 8 job
...
worker2 thread executes 33 job
worker1 thread executes 34 job
worker2 thread executes 35 job
worker1 thread executes 36 job
worker1 thread executes 37 job
worker2 thread executes 38 job
0
worker1 thread executes 39 job
worker2 thread quitting
worker1 thread quitting
附言我找不到旧SBCL的文档，所以我留下了翻译到由您决定的旧 API。希望它会有所帮助。
编辑类解决方案
在对你（已删除的）答案的评论中，我们发现你想要一个事件循环的类。我想出以下内容
(defclass event-loop ()
((lock
:initform (make-mutex))
(queue
:initform (make-waitqueue))
(jobs
:initform (make-queue))
(stopped
:initform nil)
(timeout
:initarg :wait-timeout
:initform 0)
(process-job
:initarg :process-job
:initform #'identity)
(worker-count
:initarg :worker-count
:initform (error "Must supply worker count"))))
(defmethod initialize-instance :after ((eloop event-loop) &key)
(with-slots (worker-count timeout lock queue jobs process-job stopped) eloop
(dotimes (i worker-count)
(make-thread
(lambda ()
(loop (with-mutex (lock)
(condition-wait queue lock :timeout timeout)
(when stopped
(return-from-thread nil)))
;; release *mutex* before starting the job,
;; otherwise it won't allow other threads wait for new jobs
;; you don't want to make 2 separate calls (queue-empty-p, dequeue)
;; since inbetween queue can become empty
(multiple-value-bind (job has-job) (dequeue jobs)
(if has-job
(funcall process-job job)))))))))
(defun push-job (job event-loop )
(with-slots (lock queue jobs) event-loop
(enqueue job jobs)
(with-mutex (lock)
(condition-notify queue))))
(defun stop-loop (event-loop)
(with-slots (lock queue stopped) event-loop
(with-mutex (lock)
(setf stopped t)
(condition-broadcast queue))))
你可以这样使用它
> (defparameter *el* (make-instance 'event-loop :worker-count 10 :process-job #'funcall))
> (defparameter *oq* (make-queue))
> (dotimes (i 100)
(push-job (let ((n i)) (lambda ()
(sleep 1)
(enqueue (format nil "~a job done" n) *oq*))) *el*))
它使用sb-thread:queue作为输出以避免奇怪的结果。当这起作用时，您可以在 REPL 中检查*oq*。
> *oq*
#S(QUEUE
:HEAD (SB-CONCURRENCY::.DUMMY. "7 job done" "1 job done" "9 job done"
"6 job done" "2 job done" "11 job done" "10 job done" "16 job done"
"12 job done" "4 job done" "3 job done" "17 job done" "5 job done"
"0 job done" "8 job done" "14 job done" "25 job done" "15 job done"
"21 job done" "28 job done" "13 job done" "23 job done" "22 job done"
"19 job done" "27 job done" "18 job done")
:TAIL ("18 job done")
:NAME NIL)
我使用了库chanl，它提供了消息队列机制。 当我希望线程关闭时，我只是将关键字:stop发送到队列。 当然，这不会在队列中:stop之前的所有事情都完成之前停止。 如果要提前停止，可以创建另一个队列（控制队列），该队列在数据队列之前进行检查。