co.cask.coopr.common.queue.internal.ZKQueueGroup.()方法的使用及代码示例

转载 作者：知者 更新时间：2024-03-13 12:36:58 26 4

@Inject
private ZKQueueService(ZKClient zkClient) {
 ImmutableMap.Builder<QueueType, QueueGroup> builder = ImmutableMap.builder();
 for (QueueType type : QueueType.GROUP_TYPES) {
  builder.put(type, new ZKQueueGroup(zkClient, type));
 }
 this.queueGroups = builder.build();
}

@Test
public void testInstanceInitializedWithExistingData() throws Exception {
 QueueGroup instance1 = new ZKQueueGroup(zkClient, QueueType.PROVISIONER);
 instance1.startAndWait();
 instance1.add("tenant1", new Element("val1"));
 instance1.add("tenant2", new Element("val2"));
 QueueGroup instance2 = new ZKQueueGroup(zkClient, QueueType.PROVISIONER);
 instance2.startAndWait();
 waitForQueueNames(Sets.newHashSet("tenant1", "tenant2"), instance2);
 instance1.stop();
 instance2.stop();
}

@Test
public void testChangesSeenAcrossInstances() throws Exception {
 QueueGroup instance1 = new ZKQueueGroup(zkClient, QueueType.PROVISIONER);
 QueueGroup instance2 = new ZKQueueGroup(zkClient, QueueType.PROVISIONER);
 instance1.startAndWait();
 instance2.startAndWait();
 // add a queue for tenant3 with 2 elements
 String tenant = "tenantX";
 Set<String> expectedQueueNames = Sets.newHashSet(tenant);
 instance1.add(tenant, new Element("id3-1", "val1"));
 instance1.add(tenant, new Element("id3-2", "val2"));
 // check both instances see tenant3
 Assert.assertEquals(expectedQueueNames, instance1.getQueueNames());
 waitForQueueNames(expectedQueueNames, instance2);
 // make sure each instance gets an accurate picture of the queue
 Iterator<GroupElement> queuesIter1 = instance1.takeIterator("consumer1");
 Iterator<GroupElement> queuesIter2 = instance1.takeIterator("consumer2");
 GroupElement gelement = queuesIter1.next();
 Assert.assertEquals(tenant, gelement.getQueueName());
 Assert.assertEquals("id3-1", gelement.getElement().getId());
 Assert.assertEquals("val1", gelement.getElement().getValue());
 gelement = queuesIter2.next();
 Assert.assertEquals(tenant, gelement.getQueueName());
 Assert.assertEquals("id3-2", gelement.getElement().getId());
 Assert.assertEquals("val2", gelement.getElement().getValue());
 Assert.assertFalse(queuesIter1.hasNext());
 Assert.assertFalse(queuesIter2.hasNext());
 instance1.stop();
 instance2.stop();
}

26 4 0

• python - multiprocessinq.Queue 作为 Queue.Queue 子级的属性

  我试图弄清楚以下模块正在做什么。 import Queue import multiprocessing import threading class BufferedReadQueue(Queue.

• python - multiprocessing.Queue 和 Queue.Queue 有什么不同？

  如果我使用 Queue.Queue，那么我的 read() 函数不起作用，为什么？但是，如果我使用 multiprocessing.Queue，它运行良好: from multiprocessing

• python - multiprocessing.Queue 和 queue.Queue 的实现

  我正在寻找比我在文档中找到的更多关于 Python 队列实现的见解。 根据我的理解，如果我在这方面有误，请原谅我的无知: queue.Queue():通过内存中的基本数组实现，因此不能在多个进程之间共

• python - python Queue.Queue 和 multiprocessing.Queue 的区别

  当我使用多处理模块(Windows 上的 Python 2.7)中的队列代替 Queue.Queue 时，我的程序没有完全关闭。 最终，我想使用 multiprocessing.Process 处理

• JavaScript 事件循环 : Queue vs Message Queue vs Event Queue

  阅读了大量的 JavaScript 事件循环教程，我看到了不同的术语来标识队列存储消息，当调用堆栈为空时，事件循环准备好获取消息: 队列 消息队列 事件队列 我找不到规范的术语来识别它。 甚至 MDN

• java - 错误: Type Queue Does Not Take Parameter - What's the difference between Queues and Priority Queues?

  我收到错误消息“类型队列不接受参数”。当我将更改队列行替换为 PriorityQueue 时，此错误消失并且编译正常。有什么区别以及如何将其更改为编译队列和常规队列？ import java.util

• python - 如何将项目放回 queue.Queue

  如何将项目返回到 queue.Queue？如果任务失败，这在线程或多处理中很有用，这样任务就不会丢失。 docs for queue.Queue.get()说函数可以“从队列中删除并返回一个项目”，但

• python - queue.Queue 上的多路复用？

  如何在多个 queue.Queue 上进行“选择”同时？ Golang 有 desired feature及其 channel : select { case i1 = 声明。 线程:queue 模

• python - python Queue.queue 获取并放置线程安全吗？

  http://docs.python.org/2/library/queue.html#Queue.Queue.put 这似乎是一个幼稚的问题，但我在文档和谷歌搜索中都没有找到答案，那么这些方法是线程

• javascript - .queue() 和 jquery.queue() 的区别

  这可能是个愚蠢的问题，但我对与 .dequeue() 和 $.queue() 一起使用的 .queue() 感到困惑> 或 jquery.queue()。 它们是否相同，如果是，为什么 jquery

• Python Queue.Queue 不能在线程化的 TCP 流处理程序中工作

  我正在尝试创建一个线程化的 tcp 流处理程序类线程和主线程对话，但是 Queue.Queue 也没有做我需要的，服务器从另一个程序接收数据，我只想传递它进入主线程进行处理这里是我到目前为止的代码:

• python - Queue.Queue vs 多线程Python代码中的信号量、锁等

  The principal challenge of multi-threaded applications is coordinating threads that share data or ot

• Python:为什么一些 Queue.queue 的方法是 "unreliable"？

  在Queue模块的queue类中，有几个方法，分别是qsize、empty 和 full，其文档声称它们“不可靠”。 他们到底有什么不可靠的地方？ 我确实注意到 on the Python docs网

• python - Queue.Queue 与 collections.deque

  我需要一个队列，多个线程可以将内容放入其中，并且多个线程可以从中读取。 Python 至少有两个队列类，Queue.Queue 和 collections.deque，前者似乎在内部使用后者。两者都在

• message-queue - 为什么是 ActiveMQ，而不是简单的 Queue/Mutex？

  明天我将介绍我选择进程内消息队列实现的基本原理，但我无法阐明我的推理。我的合作设计者提议我们实现一个简单的异步队列，只使用基本的作业列表和互斥锁来控制访问，我建议在嵌入式模式下使用 ActiveMQ。

• scala - 对于 "trait Queue[T]"， `Queue` 是一种类型吗？

  在 scala 中定义了一个特征: trait Queue[T] Queue 是一种类型吗？或其他东西，例如类型构造函数？ 来自 http://artima.com/pins1ed/type-para

• queue - 我如何在 SML/NJ 中使用 Queue 库

  我看到 SML/NJ 包含一个队列结构。我不知道如何使用它。如何使用 SML/NJ 提供的附加库？ 最佳答案 Queue structure SML '97 未指定，但它存在于 SML/NJ 的顶级环

• javascript - queue.await() 和 queue.awaitAll() 的区别

  我是 D3 和 JavaScript 的新手。 我试图理解其中的 queue.js。 我已经完成了 this关联。但是仍然无法清楚地了解 queue.await() 和 queue.awaitAll(

• c++ - 错误 : no matching function for call to "Queue::Queue()"

  所以我试图在我的 main.cpp 文件中调用一个函数，但我得到“错误:没有匹配函数来调用‘Queue::Queue()。” 队列.h #ifndef QUEUE_H #define QUEUE_H

• Python - 如何将整个 numpy 数组一次全部放入 Queue.Queue 但分别检索每一行

  假设我有一个 10 行的二维 numpy 数组 例如 array([[ 23425. , 521331.40625], [ 23465. , 521246.03125],