标签: concurrent-queue

如果Queue中没有项目,ConcurrentQueue中的TryDequeue将返回false.

如果队列是空的,我需要我的队列将等待,直到新项目被添加到队列中并且它将新队列出列,并且该过程将继续这样.

我应该在C#4.0中使用monitor.enter,wait,pulse或任何更好的选项

我之前没有使用过并发队列.

可以在while循环中使用如下所示的TryDequeue吗？难道这不会永远被卡住吗？

var cq = new ConcurrentQueue<string>();
cq.Enqueue("test");

string retValue;

while(!cq.TryDequeue(out retValue))
{
    // Maybe sleep?
}

//Do rest of code

ConcurrentQueue<T>我最近使用和构建了一个消费者/生产者系统SemaphoreSlim。然后利用新System.Threading.Channel类制作了另一个替代系统。

使用 BenchmarkDotNet 对两个系统进行基准测试后，将 1000 个项目写入两个系统 1000 次（并等待读者完成），我得到以下结果：

|      Method | ItemsCount | Iterations |        Mean |       Error |      StdDev |      Median |  Allocated |
|------------ |----------- |----------- |------------:|------------:|------------:|------------:|-----------:|
|     MyQueue |       1000 |       1000 | 19,379.4 us | 1,230.30 us | 3,569.33 us | 18,735.6 us | 8235.02 KB |
|   MyChannel |       1000 |       1000 | 45,858.2 us | 1,298.42 us | 3,704.46 us | 45,689.2 us |   72.11 KB |

实施ConcurrentQueue …

我想使用Visual Studio 2010的并发库来在线程之间传递操作.我有我的课程,SimpleAction并指向它存储在Concurrency::concurrent_queue.

使用这个定义和"消费"逻辑它可以工作:

typedef Concurrency::concurrent_queue<SimpleAction *> ActionQueue;

while (true)
   {
   SimpleAction *action = nullptr;
   while (m_queue.try_pop(action))
      {
      action->process();
      delete action;
      }
   Sleep(100);
   }

但是,当我将其更改为std :: unique_ptr时,如下所示:

typedef Concurrency::concurrent_queue<std::unique_ptr<SimpleAction>> ActionQueue;

while (true)
   {
   std::unique_ptr<SimpleAction> action;
   while (m_queue.try_pop(action))
      {
      action->process();
      }
   Sleep(100);
   }

编译器给出以下错误消息:

F:\DevStudio\Vs2010\VC\INCLUDE\concurrent_queue.h(366) : error C2248: 'std::unique_ptr<_Ty>::unique_ptr' : cannot access private member declared in class 'std::unique_ptr<_Ty>'
    with
    [
        _Ty=`anonymous-namespace'::SimpleAction
    ]
F:\DevStudio\Vs2010\VC\INCLUDE\memory(2347) : see declaration of 'std::unique_ptr<_Ty>::unique_ptr'
    with
    [
        _Ty=`anonymous-namespace'::SimpleAction
    ]
F:\DevStudio\Vs2010\VC\INCLUDE\concurrent_queue.h(365) : while compiling class …

Run Code Online (Sandbox Code Playgroud)

默认情况下，C++ 容器应该是线程安全的。我一定是queue错误地使用了多线程，因为对于这段代码：

#include <thread>
using std::thread;
#include <iostream>
using std::cout;
using std::endl;
#include <queue>
using std::queue;
#include <string>
using std::string;
using std::to_string;
#include <functional>
using std::ref;


void fillWorkQueue(queue<string>& itemQueue) {
    int size = 40000;
    for(int i = 0; i < size; i++)
        itemQueue.push(to_string(i));
}

void doWork(queue<string>& itemQueue) {
    while(!itemQueue.empty()) {
        itemQueue.pop();
    }   
}

void singleThreaded() {
    queue<string> itemQueue;
    fillWorkQueue(itemQueue);
    doWork(itemQueue);
    cout << "done\n";
}

void multiThreaded() {
    queue<string> itemQueue;
    fillWorkQueue(itemQueue);
    thread t1(doWork, ref(itemQueue));
    thread t2(doWork, ref(itemQueue)); …

我有一个非常基本的问题,更多的是关于概念ConcurrentQueue.队列是FIFO.当多个线程开始访问它时,我们如何保证FIFO？假设,我已经加入Apple,Oranges,Lemon,Peach和Apricot-的顺序.第一个TryTake应该回来Apple.但是当多个线程开始提供自己的TryTake请求时会发生什么？当一个线程Lemon甚至可以在另一个线程返回之前返回时,是不是有可能Apple？我假设其他项目也将被返回,直到队列为空.但这些回报是否会围绕FIFO的基本原则进行管理？

ConcurrentQueue有TryDequeue方法。

Queue就有Dequeue方法了。

没有ConcurrentDictionary方法Add，但我们有TryAdd。

我的问题是：

这些并发收集方法有什么区别？为什么它们对于并发集合不同？

JavadocConcurrentLinkedQueue明确指出 size() 方法的复杂度是 O(n)。我觉得这很令人惊讶。我会遵循库存LinkedList.size()方法，在柜台中累积规模。鉴于操作的异步性质ConcurrentLinkedQueue，计数器自然必须是AtomicInteger。这样做是不是因为它会违反非阻塞保证？AtomicInteger如果是这样，计数器会带来多少开销？

在类中，定义了ConcurrentQeueue<>额外的方法。TryDequeue()但是，当它实现时IProducerConsumerCollection<>，它也有一个TryTake()方法。根据文档，他们都做同样的事情：

尝试出队：

尝试删除并返回并发队列开头的对象。

尝试一下：

对于 ConcurrentQueue，此操作将尝试从 ConcurrentQueue 的开头删除对象。

为什么要费心实施该TryDequeue方法呢？

谁能帮助我理解我创建的这段代码：

let cq = DispatchQueue(label: "downloadQueue", attributes: .concurrent)
cq.sync {
for i in 0..<10 {
    sleep(2)
    print(i)
  }
}

print("all finished!")

输出是串行顺序 1->10，中间等待 2 秒。最后会打印出all finished

我明白最后一部分。然而我的问题是：

并发队列不应该同时启动多个任务吗？所以我最初的想法是：应该按1-10打印concurrently，不一定按序列顺序。

谁能解释一下sync并发队列调用的目的，并给我一个例子，为什么以及何时需要它？