基于MVar的Conduit Source实现问题：线程无限阻塞异常求助

你碰到的问题核心其实是MVar的特性和Conduit Source持续产出的需求不匹配。MVar本质是个二元容器：要么存着单个值，要么是空的。当你用takeMVar取出值后，它就变成空状态了，如果此时客户端还没把新消息通过putMVar塞进来，下一次读取操作就会无限阻塞，最终触发thread blocked indefinitely in an MVar operation这个异常。

最优解决方案：用Chan替代MVar #

Chan是Haskell标准库中的线程安全FIFO队列，完美适配这种“持续接收消息并通过Conduit产出”的场景——它能存储多条消息，readChan会自动等待新消息到来，不会因为单次读取就陷入阻塞。

改造步骤如下：

1. 替换MVar为Chan：

   import Control.Concurrent.Chan
   import Data.Conduit
   
   -- 先创建一个空的消息队列
   messageChan <- newChan
   

2. 修改客户端订阅逻辑：
   把原来的putMVar换成writeChan，这样每条新消息都会被追加到队列末尾：

   hookToChan $ \topic msg -> writeChan messageChan (show msg)
   

3. 实现基于Chan的Conduit Source：
   这个Source可以持续读取队列中的消息并向外产出：

   sourceQueue :: Chan String -> Source IO String
   sourceQueue chan = do
     msg <- lift $ readChan chan
     yield msg
     sourceQueue chan  -- 循环读取，持续处理新消息
   

为什么MVar不适合这个场景？ #

再明确一下：MVar只能保存单个值，取出值后就会变成空状态。如果客户端推送消息的速度跟不上Source的读取速度，或者某次读取后没有新消息写入，Source就会卡在takeMVar操作上，永远等不到新值，最终抛出阻塞异常。而Chan是专门为多消息的线程间通信设计的，天生就适合这种持续订阅的需求。

（可选）如果一定要用MVar怎么办？ #

如果因为某些限制必须使用MVar，你得确保每次读取MVar后，客户端能及时填充新的空MVar（或者用swapMVar这类操作），但这种方式会增加同步逻辑的复杂度，远不如Chan简洁。比如可以这样实现：

sourceMVarLoop :: MVar String -> Source IO String
sourceMVarLoop mvar = do
  msg <- lift $ takeMVar mvar
  yield msg
  -- 这里必须保证客户端会再次调用putMVar传入新消息，否则还是会阻塞
  sourceMVarLoop mvar

但这种方案依然依赖客户端的严格同步，很容易再次触发相同的阻塞问题，所以还是优先推荐使用Chan。

内容的提问来源于stack exchange，提问作者Nikita Tchayka