这个问题有点代码高尔夫，而且很多新手。
我正在使用 Haskell 中很棒的pipes库，我想拆分管道以沿多个通道发送相同的数据（进行广播）。Pipes.Concurrent教程建议使用 spawn 创建邮箱，利用Output的幺半群状态。例如，我们可以做这样的事情：
main = do
(output1, input1) <- spawn Unbounded
(output2, input2) <- spawn Unbounded
let effect1 = fromInput input1 >-> pipe1
let effect2 = fromInput input2 >-> pipe2
let effect3 = P.stdinLn >-> toOutput (output1 <> output2)
...
通过邮箱进行这种间接处理是否真的必要？我们可以写这样的东西吗？
main = do
let effect3 = P.stdinLn >-> (pipe1 <> pipe2)
...
上面没有编译，因为Pipe没有Monoid实例。这有什么充分的理由吗？第一种方法真的是拆分管道的最干净方法吗？
有两种方法可以在不使用并发的情况下执行此操作，这两种方法都有注意事项。
第一种方法是，如果pipe1和pipe2只是简单的Consumer那么它就会永远循环，就像：
p1 = for cat f -- i.e. p1 = forever $ await >>= f
p2 = for cat g -- i.e. p2 = forever $ await >>= g
。那么解决这个问题的简单方法是只写：
for P.stdinLn $ str -> do
f str
g str
例如，如果p1只是print每个值：
p1 = for cat (lift . print)
。p2只是将该值写入句柄：
p2 = for cat (lift . hPutStrLn h)
。然后，您将像这样组合它们：
for P.stdinLn $ str -> do
lift $ print str
lift $ hPutStrLn h str
但是，这种简化仅适用于琐碎循环Consumer。 还有另一种更通用的解决方案，即为管道定义一个ArrowChoice实例。 我相信基于拉动的Pipe不允许正确的守法实例，但基于推送的Pipe可以：
newtype Edge m r a b = Edge { unEdge :: a -> Pipe a b m r }
instance (Monad m) => Category (Edge m r) where
id = Edge push
(Edge p2) . (Edge p1) = Edge (p1 >~> p2)
instance (Monad m) => Arrow (Edge m r) where
arr f = Edge (push />/ respond . f)
first (Edge p) = Edge $ (b, d) ->
evalStateP d $ (up > unsafeHoist lift . p />/ dn) b
where
up () = do
(b, d) <- request ()
lift $ put d
return b
dn c = do
d <- lift get
respond (c, d)
instance (Monad m) => ArrowChoice (Edge m r) where
left (Edge k) = Edge (bef >=> (up > (k />/ dn)))
where
bef x = case x of
Left b -> return b
Right d -> do
_ <- respond (Right d)
x2 <- request ()
bef x2
up () = do
x <- request ()
bef x
dn c = respond (Left c)
这需要一个 newtype，以便类型参数按照ArrowChoice期望的顺序排列。
如果您不熟悉术语 基于推送的Pipe ，它基本上是一个从最上游的管道而不是最下游的管道开始的Pipe，它们都具有以下形状：
a -> Pipe a b m r
把它想象成一个Pipe，在它从上游收到至少一个值之前，它不能"去"。
这些基于推送的Pipe是传统的基于拉动的Pipe的"双重"，具有自己的组合运算符和标识：
(>~>) :: (Monad m)
=> (a -> Pipe a b m r)
-> (b -> Pipe b c m r)
-> (a -> Pipe a c m r)
push :: (Monad m)
-> a -> Pipe a a m r
。但默认情况下，单向 Pipes API 不会导出此内容。 您只能从Pipes.Core中获取这些运算符（您可能希望更仔细地研究该模块，以建立对它们如何工作的直觉）。 该模块表明，基于推送的 Pipe s 和基于拉的 Pipe 都是更通用的双向版本的特例，理解双向情况是了解为什么它们是彼此对偶的方式。
一旦您拥有基于推送的管道的Arrow实例，您可以编写如下内容：
p >>> bifurcate >>> (p1 +++ p2)
where
bifurcate = Edge $ pull ~> a -> do
yield (Left a) -- First give `p1` the value
yield (Right a) -- Then give `p2` the value
然后，完成后，您将使用runEdge将其转换为基于拉动的管道。
这种方法有一个主要缺点，即您无法自动将基于拉动的管道升级到基于推送的管道（但通常不难弄清楚如何手动执行此操作）。 例如，要将Pipes.Prelude.map升级为基于推送的Pipe，您可以编写：
mapPush :: (Monad m) => (a -> b) -> (a -> Pipe a b m r)
mapPush f a = do
yield (f a)
Pipes.Prelude.map f
然后，它有正确的类型被包裹在Arrow中：
mapEdge :: (Monad m) => (a -> b) -> Edge m r a b
mapEdge f = Edge (mapPush f)
当然，更简单的方法是从头开始编写：
mapEdge f = Edge $ push ~> yield . f
使用最适合您的方法。
事实上，我之所以提出Arrow和ArrowChoice实例，正是因为我试图回答与您完全相同的问题：如何在不使用并发的情况下解决此类问题？ 我在另一个 Stack Overflow 答案中写了一个关于这个更一般主题的长答案，我描述了如何使用这些 Arrow 和 ArrowChoice 实例将并发系统提炼成等效的纯系统。