c# - .Net BlockingCollection.Take(2) : Safely removing two items at a time

Question

在做了一些研究之后，我正在求助于有关如何有效地从并发集合中删除两个项目的任何反馈。我的情况涉及通过 UDP 传入的消息，这些消息当前被放入 BlockingCollection 中。一旦集合中有两个用户，我需要安全地获取两个用户并对其进行处理。我见过几种不同的技术，包括下面列出的一些想法。我当前的实现如下，但我认为有一种更简洁的方法可以做到这一点，同时确保用户以两人为一组进行处理。这是这种情况下的唯一限制。

当前实现:

    private int userQueueCount = 0;
    public BlockingCollection<User> UserQueue = new BlockingCollection<User>();

    public void JoinQueue(User u)
    {
           UserQueue.Add(u);
           Interlocked.Increment(ref userQueueCount);

           if (userQueueCount > 1)
           {
               IEnumerable<User> users = UserQueue.Take(2);
               if(users.Count==2) {
                 Interlocked.Decrement(ref userQueueCount);
                 Interlocked.Decrement(ref userQueueCount);
                 ... do some work with users but if only one 
                 is removed I'll run into problems
               }

           }
    }

我想做的是类似这样的事情，但我目前无法在生产环境中对其进行测试以确保完整性。

 Parallel.ForEach(UserQueue.Take(2), (u) => { ... });

或者更好的是:

    public void JoinQueue(User u)
    {
           UserQueue.Add(u);
           // if needed? increment
           Interlocked.Increment(ref userQueueCount);
           UserQueue.CompleteAdding();
    }

然后在某处实现:

        Task.Factory.StartNew(() =>
        {
            while (userQueueCount > 1) OR (UserQueue.Count > 1) If it's safe?
            {
                IEnumerable<User> users = UserQueue.Take(2);
                ... do stuff
            }

        });

问题是我不确定我能否保证在条件 (Count > 1) 和 Take(2) 之间我确保 UserQueue 至少有两个项目要处理？传入的 UDP 消息是并行处理的，因此我需要一种方法来安全地将项目成对地从阻塞/并发集合中拉出。

有没有更好/更安全的方法来做到这一点？

修改后的评论:这个问题的预期目标实际上只是为了实现一种稳定/线程安全的方法来处理 .Net 4.0 中并发集合的项目。它不必很漂亮，它只需要在并行环境中以无序成对的形式处理项目的任务中保持稳定即可。

Answer 1

这是我在粗略代码中所做的:

ConcurrentQueuequeue = new ConcurrentQueue(); //can use a BlockingCollection too (as it's just a blocking ConcurrentQueue by default anyway)

public void OnUserStartedGame(User joiningUser)
{
   User waitingUser;
   if (this.gameQueue.TryDequeue(out waitingUser)) //if there's someone waiting, we'll get him
      this.MatchUsers(waitingUser, joiningUser);
   else
      this.QueueUser(joiningUser); //it doesn't matter if there's already someone in the queue by now because, well, we are using a queue and it will sort itself out.
}

private void QueueUser(User user)
{
   this.gameQueue.Enqueue(user);
}

private void MatchUsers(User first, User second)
{
   //not sure what you do here
}

基本思路是，如果有人想开始游戏，而您的队列中有人，您匹配他们并开始游戏 - 如果没有人，则将他们添加到队列中。充其量一次队列中只有一个用户，但如果没有，那也不算太糟糕，因为随着其他用户开始游戏，等待的用户将逐渐移除，并且不会添加新用户，直到队列为空再次。