如何构建支持异步的Monad Transformer栈并实现隐式异步调用？

原async库的局限性 #

async库允许我们编写这样的异步代码：

Async.withAsync (getUrl url1) $ \getHandler1 -> do
  Async.withAsync (getUrl url1) $ \getHandler2 -> do
    url1Content <- Async.wait getHandler1
    url2Content <- Async.wait getHandler2
    pure $ url1Content <> url2Content

但withAsync有个明显局限：它只能用于纯IO动作，类型签名为withAsync :: IO a -> (Async a -> IO b) -> IO b。我们需要在包含StateT、ExceptT和ReaderT的Monad Transformer栈中实现类似的异步能力，同时满足以下核心需求：

核心需求 #

• 将异步操作抛出的异常捕获到栈中的ExceptT里，尽早终止后续计算；
• 异步动作能完整访问整个Monad栈的状态：这里确实需要把State放在TVar（通过STM模块）中，保证多线程下的状态原子性；
• 无需显式调用withAsync，在栈的do块中调用类似async的接口时，底层隐式处理异步动作的生命周期，让do块剩余逻辑自动确保异步动作完成，同时返回可后续wait的句柄。

目标伪代码（Context栈） #

我们希望最终能写出这样的业务代码：

main = Context.runContext $ do
  firstPageContentHandler <- Context.getAsync "https://www.google.com"
  secondPageContentHandler <- Context.getAsync "https://www.yahoo.com"
  part1 <- Context.doSync doSomeStuffThatBuildsPartOfAUrl
  part2 <- doSomeMoreStuffThatBuildsPartOfAUrlMaybePurely "with some data"
  let anotherUrl = part1 <> part2
  firstPageContent <- Context.wait firstPageContentHandler
  thirdPageContentHandler <- Context.getAsync (buildUrlFromContent firstPageContent)
  fourthPageContent <- Context.get anotherUrl
  secondPageContent <- Context.wait secondPageContentHandler
  let joinedContent = secondPageContent <> thirdPageContentHandler <> fourthPageContent
  Context.setResponseAs joinedContent
  pure ()

实现方案 #

1. 定义Monad栈结构 #

我们需要构建一个包含ReaderT（全局环境）、ExceptT（异常处理）、StateT（基于TVar的共享状态）和IO的栈，同时利用unliftIO实现跨栈异步操作：

{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE TypeApplications #-}

import Control.Concurrent.Async (Async)
import qualified Control.Concurrent.Async as Async
import Control.Concurrent.STM (TVar, atomically, readTVar, writeTVar, newTVar)
import Control.Monad.Except (ExceptT, MonadError, catchError, throwError, runExceptT)
import Control.Monad.Reader (ReaderT, MonadReader, ask, runReaderT)
import Control.Monad.State (StateT, MonadState, get, runStateT)
import Control.Monad.IO.Unlift (MonadUnliftIO, withRunInIO)
import Control.Exception (SomeException)
import Data.Text (Text)
import qualified Data.Text as T

-- 自定义异常类型
data ContextError = AsyncError SomeException
                  | BusinessError Text
                  deriving (Show)

-- 共享状态：用TVar包裹确保线程安全
data ContextState = ContextState
  { csResponse :: Text
  -- 可添加更多业务状态字段
  }

-- 全局环境配置
data ContextEnv = ContextEnv
  { ceHttpTimeout :: Int
  -- 可添加更多环境参数
  }

-- 最终的Monad栈类型
newtype Context a = Context
  { unContext :: ReaderT ContextEnv (ExceptT ContextError (StateT (TVar ContextState) IO)) a
  } deriving (Functor, Applicative, Monad, MonadIO, MonadReader ContextEnv, MonadError ContextError, MonadState (TVar ContextState))

-- 实现MonadUnliftIO：跨栈异步操作的核心
instance MonadUnliftIO Context where
  withRunInIO inner = Context $ withRunInIO $ \run -> inner (run . unContext)

2. 实现核心异步操作函数 #

接下来实现getAsync、wait等核心函数，同时处理异常捕获：

-- 异步执行Context动作，返回Async句柄
getAsync :: (MonadUnliftIO m, MonadError ContextError m) => m a -> m (Async a)
getAsync action = do
  let wrappedAction = action `catchError` throwError
  withRunInIO $ \run -> Async.async (run wrappedAction)

-- 等待异步操作完成，捕获异常并转换为ContextError
wait :: (MonadIO m, MonadError ContextError m) => Async a -> m a
wait asyncHandle = liftIO (Async.wait asyncHandle) `catchError` (throwError . AsyncError)

-- 语义化的同步执行函数（实际就是普通Context动作）
doSync :: m a -> m a
doSync = id

-- 示例：同步获取URL内容
get :: Text -> Context Text
get url = do
  timeout <- asks ceHttpTimeout
  -- 替换为真实HTTP客户端逻辑
  liftIO $ putStrLn $ "Fetching " <> T.unpack url
  pure $ T.pack $ "Content of " <> T.unpack url

-- 示例：更新响应状态
setResponseAs :: Text -> Context ()
setResponseAs content = do
  stateVar <- get
  liftIO $ atomically $ writeTVar stateVar (ContextState content)

3. 实现runContext入口函数 #

最后实现栈的执行入口，初始化状态和环境：

runContext :: ContextEnv -> ContextState -> Context a -> IO (Either ContextError (a, ContextState))
runContext env initialState action = do
  stateVar <- atomically $ newTVar initialState
  (eitherResult, finalStateVar) <- runStateT (runExceptT (runReaderT (unContext action) env)) stateVar
  finalState <- atomically $ readTVar finalStateVar
  pure $ eitherResult >>= \a -> Right (a, finalState)

关键说明 #

• TVar的必要性：异步动作在独立线程执行，共享状态必须用TVar（STM）保证原子性，避免竞态条件；
• MonadUnliftIO的作用：它允许将带栈的动作转换为纯IO动作，是Transformer栈中实现异步的核心；
• 异常处理：通过catchError将异步异常转换为ExceptT中的自定义错误，确保异常能被上层捕获并终止计算；
• 隐式生命周期管理：如果需要完全替代withAsync的自动清理，可以在ContextState中添加字段跟踪所有异步句柄，在runContext结束时自动等待或取消未完成的异步动作，避免资源泄漏。

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