Rust go 重并堆积MPSC通道消息

发布于06月27日

当使用channel时，假设在给定时刻，给定接收方的队列中有以下消息:

abc
blah
abc
something
something
blah
something

通常，消息的处理方式如下:

let (sender, receiver) = mpsc::channel::<Msg>();
for msg in receiver {
   // Process a message...
}

对已经堆积的邮件进行重复数据删除最简单的方法是什么？

在上面的例子中，我只想处理3条(而不是7条)消息:

abc
blah
something

注意:在处理完一条消息abc之后，以后可以再次接收(并处理)另一条消息abc.其目标是仅对已在队列中堆积的邮件进行重复数据消除.

解决方案1

以下是一个解决方案示例:

loop {
    let deduped_msgs : HashSet<Msg> = HashSet::from_iter(receiver.try_iter());
    for msg in deduped_msgs {
        // Process a message...                
    }
    thread::sleep(time::Duration::from_millis(10));
}

上面使用的是非阻塞的try_iter方法.我添加了一个短sleep()，以避免对CPU造成过大的压力.

解决方案2

根据Chayim的建议，这里有一个类似的解决方案，它避免了每X毫秒进行一次定期判断的需要:

let mut deduped_msgs = HashSet::new();
loop {
    deduped_msgs.insert(receiver.recv().unwrap());
    deduped_msgs.extend(receiver.try_iter());               
    for msg in deduped_msgs.drain() {
        // Process a message... 
    }
}

处理顺序

如果处理消息的顺序很重要，那么HashSet(单独)可能不是最佳 Select .

要按照消息到达的顺序处理消息(只考虑每条消息第一次出现的情况)，例如可以使用IndexSet.
在某些情况下，消息的自然顺序可能更有意义，然后BTreeSet可能是最简单的 Select .
要应用某种自定义优先级顺序，可以为例如，使用一个向量和sort it个VIA sort_unstable_by 或一百零二 (首先用HashSet对他们进行重复数据删除可能更简单，但另一个选项可能是 dedup 已经排序的向量).