标签: queue

我有一个线程将事件推送到第二个线程的传入队列的经典问题.只有这一次,我对表现很感兴趣.我想要实现的是:

• 我希望并发访问队列,生产者推送,接收器弹出.
• 当队列为空时,我希望使用者阻塞队列,等待生产者.

我的第一个想法是使用a LinkedBlockingQueue,但我很快意识到它不是并发的,而且性能受到了影响.在另一方面,我现在使用ConcurrentLinkedQueue,但还是我付出的成本wait()/ notify()每本出版物上.由于消费者在找到空队列时没有阻止,因此我必须同步并wait()锁定.另一方面,制作人必须获得锁定并notify()在每一个出版物上.总体结果是sycnhronized (lock) {lock.notify()},即使不需要,我也会支付每一份出版物的费用 .

我猜这里需要的是一个阻塞和并发的队列.我想象一个push()操作工作ConcurrentLinkedQueue,notify()当推送元素是列表中的第一个时,对象的额外内容.这种检查我认为已经存在ConcurrentLinkedQueue,因为推动需要连接下一个元素.因此,这比每次在外部锁上同步要快得多.

是这样的/合理的吗？

我正在Albahari兄弟的Nutshell中阅读C#4.0,我发现了这个:

堆栈在内部实现,其数组根据需要调整大小,与Queue和List一样.(第288页,第4段)

我不禁想知道为什么.LinkedList提供O(1)头尾插入和删除(这应该适用于堆栈或队列).可调整大小的数组有O(1)缓冲插入(如果我没记错的话),但O(n)最坏的情况(我不确定删除).它可能比链表使用更多的空间(对于大型堆栈/队列).

还有更多吗？双链表实现的缺点是什么？

我有兴趣知道实现基于线程的队列的最佳方法是什么.

例如:

我有10个动作,我想用4个线程执行.我想创建一个队列,其中所有10个动作都是线性放置的,并且用4个线程开始前4个动作,一旦一个线程完成执行,下一个将启动等 - 所以一次,线程的数量是4或小于4.

我正在使用队列编写基数排序算法,我想在开始向队列添加内容之前让STL队列分配空间,这样我就可以避免持续的动态调整大小操作.

即使这不存在,我想要的东西具有...的效果

queue<int> qs(N);
for(int i=0;i<N;++i)
  qs.push(rand());

以这种方式,它不会在循环期间动态分配任何内存.

有问题的实际代码......

void radix_sort()
{
// Biggest number?
int max=-1;
for(int i=0;i<N;++i)
    if(a[i]>max)
        max = a[i];

// How many digits in it
int maxdigits=1;
while(max /= 10) maxdigits++;

// Create some buckets.
deque<int> b[10];
for(int i=0;i<10;++i)
    b[i] = deque<int>(N);

int div=1;
// Radix Sort by digits
for(int d=1;d<=maxdigits;++d)
{
    if(d>1)
        div*=10;

    // Queue
    for(int i=0;i<N;++i)
        b[ (a[i]/div) % 10 ].push_front(a[i]);

    // Dequeue
    int k=0;    
    for(int q=0;q<10;++q)
        while(b[q].size() > 0)
        {
            a[k++] = b[q].back(); …

我正在尝试找出使用代理从消息队列(Amazon SQS)中使用项目的最佳方法.现在我有一个函数(process-queue-item),它从队列中获取项目并对其进行处理.

我想同时处理这些项目,但我无法理解如何控制代理.基本上我想尽可能地让所有代理忙,而不是从队列中抽取很多项目并开发积压(我会在几台机器上运行它,所以项目需要留在队列中直到它们真的需要).

谁能给我一些关于改进我的实现的指示？

(def active-agents (ref 0))

(defn process-queue-item [_]
  (dosync (alter active-agents inc))
  ;retrieve item from Message Queue (Amazon SQS) and process
  (dosync (alter active-agents dec)))

(defn -main []
  (def agents (for [x (range 20)] (agent x)))

  (loop [loop-count 0]

    (if (< @active-agents 20)
      (doseq [agent agents]
        (if (agent-errors agent)
          (clear-agent-errors agent))
        ;should skip this agent until later if it is still busy processing (not sure how)
        (send-off agent process-queue-item)))

    ;(apply await-for (* 10 1000) agents)
    (Thread/sleep  10000) …

如何检查消息队列是否已存在？

我有2个不同的应用程序,一个创建队列,另一个从该队列读取.

因此,如果我首先运行从队列中读取的客户端,而不是崩溃.
所以为了避免我想首先检查队列是否存在.

这是我如何读取队列的代码片段:

QueueingBasicConsumer <ConsumerName> = new QueueingBasicConsumer(<ChannelName>); 
<ChannelName>.BasicConsume("<queuename>", null, <ConsumerName>); 
BasicDeliverEventArgs e = (BasicDeliverEventArgs)<ConsumerName>.Queue.Dequeue();

无论是使用堆栈还是DFS或BFS的队列,我总是混淆.有人可以提供一些关于如何记住哪种算法使用哪种数据结构的直觉？

我对使用队列的任务执行感到困惑.我已经阅读了文档,我认为我理解了bucket_size和rate,但是当我将20个任务发送到设置为5/h,大小为5的队列时,所有20个任务尽可能快地执行,最后完成1分钟.

deferred.defer(spam.cookEggs, 
               egg_keys, 
               _queue="tortoise")  

- name: tortoise
  rate: 5/h  
  bucket_size: 5  

我想要的是我是创建10个还是100个任务,我只想要每小时运行5个任务.所以完成需要大约4个小时的20个任务.我希望他们的执行分散开来.

UPDATE

问题是我假设在本地运行时遵循了任务执行率规则,但事实并非如此.您无法在本地测试执行率.当我部署到生产时,我设置的速率和桶大小按我的预期执行.

创建这两个对象有什么区别

Queue<String> test = new LinkedList<String>();

和

List<String> test2 = new LinkedList<String>();

test和之间的实际差异是test2什么？他们俩都是LinkedList？是否存在使用其中一个的性能差异或原因？

在无限循环中使用队列中的值时 - 更有效:

1)阻止队列,直到通过take()获得值

while (value = queue.take()) { doSomething(value); }

2)睡眠n毫秒并检查物品是否可用

while (true) {

    if ((value = queue.poll()) != null) { doSomething(value); }

    Thread.sleep(1000);
}