mapConcurrently异常触发全任务取消：求替代方案或优化实现

嘿，这个问题我之前踩过坑！mapConcurrently的“一失败全取消”特性在处理输入EOF、网络错误这类个别失败场景时确实挺闹心的，而且你代码里还有个容易忽略的小问题——多个并发线程同时调用getLine会导致输入读取混乱（标准输入Handle的读取不是线程安全的，多线程读可能把一行拆成好几段）。咱们一步步来解决：

先分析核心问题 #

1. mapConcurrently的异常行为：只要任一任务抛出未捕获的异常，它就会立即取消所有剩余任务并重新抛出异常，这就是你遇到EOF后所有任务都被取消的原因。
2. 多线程读输入的风险：多个线程同时调用getLine会导致输入数据交错，比如两个线程各读一行的一部分，生成错误的请求参数。

最优解决方案：单线程读输入，并发处理请求 #

最稳妥的方式是用单线程读取所有输入（直到EOF），再用带并发限制的工具处理请求。async库的pooledMapConcurrentlyN正好满足需求——它自带并发数限制，而且单个任务的异常不会影响其他任务（只要你在任务内部捕获异常）。

修改后的代码 #

{-# LANGUAGE OverloadedStrings #-}
module Main where
import Control.Monad
import Network.Wreq
import Control.Concurrent.Async (pooledMapConcurrentlyN)
import Control.Exception (try, SomeException)
import qualified Data.ByteString.Lazy as BL

main :: IO ()
main = do
  -- 单线程读取所有输入行，遇到EOF自动停止
  params <- lines <$> getContents
  -- 用4个并发处理每个参数，单个任务失败不影响其他
  void $ pooledMapConcurrentlyN 4 processUrl params

processUrl :: String -> IO ()
processUrl param = do
  -- 捕获HTTP请求的异常，避免冒泡到上层取消所有任务
  result <- try (get (url ++ param)) :: IO (Either SomeException (Response BL.ByteString))
  case result of
    Left e -> putStrLn $ "请求参数 '" ++ param ++ "' 失败: " ++ show e
    Right response -> print response

url :: String
url = "http://example.com/resources?param="

为什么这个方案更好 #

• 输入读取安全：单线程读取所有输入，避免多线程读输入的混乱问题。
• 异常隔离：每个请求的异常都被内部捕获，不会触发pooledMapConcurrentlyN的全局取消。
• 简洁高效：pooledMapConcurrentlyN已经帮你处理了线程池的管理，不用自己手动维护信号量。

进阶方案：边读边处理（适合超大输入） #

如果你的输入非常大，没法一次性读到内存里，可以用生产者-消费者模式：单线程读输入（生产者），把参数放到队列里，多个消费者线程从队列取参数处理。这样既能边读边处理，又能保证输入读取的安全性。

代码示例 #

{-# LANGUAGE OverloadedStrings #-}
module Main where
import Control.Monad
import Network.Wreq
import Control.Concurrent.Async (async, wait)
import Control.Concurrent.STM.TQueue
import Control.Concurrent.STM
import Control.Exception (try, SomeException)
import qualified Data.ByteString.Lazy as BL

main :: IO ()
main = do
  queue <- newTQueueIO
  -- 启动生产者线程：读取输入直到EOF，把参数放入队列
  producer <- async $ do
    let readLoop = do
          result <- try getLine :: IO (Either SomeException String)
          case result of
            Left _ -> do -- 遇到EOF或读取错误，写入结束标记
              atomically $ writeTQueue queue Nothing
            Right param -> do
              atomically $ writeTQueue queue (Just param)
              readLoop
    readLoop
  -- 启动4个消费者线程
  consumers <- replicateM 4 $ async $ consumer queue
  -- 等待生产者结束，再等待所有消费者处理完剩余任务
  wait producer
  mapM_ wait consumers

consumer :: TQueue (Maybe String) -> IO ()
consumer queue = do
  mParam <- atomically $ readTQueue queue
  case mParam of
    Nothing -> return () -- 收到结束标记，退出线程
    Just param -> do
      result <- try (get (url ++ param)) :: IO (Either SomeException (Response BL.ByteString))
      case result of
        Left e -> putStrLn $ "请求参数 '" ++ param ++ "' 失败: " ++ show e
        Right response -> print response
      consumer queue -- 继续处理下一个参数

url :: String
url = "http://example.com/resources?param="

总结 #

• 优先用pooledMapConcurrentlyN处理有限任务，它是mapConcurrently的带并发限制版本，且异常隔离更友好。
• 永远不要让多个并发线程读取同一个输入Handle，单线程读输入再分发是最安全的方式。
• 超大输入场景用生产者-消费者模式，通过队列传递参数，实现边读边处理。

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