下面列出了java.util.Queue#poll ( ) 实例代码，或者点击链接到github查看源代码，也可以在右侧发表评论。
@Override
public long drainQueue(final Queue<FlowFileRecord> sourceQueue, final List<FlowFileRecord> destination, int maxResults, final Set<FlowFileRecord> expiredRecords) {
long drainedSize = 0L;
FlowFileRecord pulled = null;
final long expirationMillis = expirationPeriod.get().getMillis();
while (destination.size() < maxResults && (pulled = sourceQueue.poll()) != null) {
if (isLaterThan(getExpirationDate(pulled, expirationMillis))) {
expiredRecords.add(pulled);
if (expiredRecords.size() >= MAX_EXPIRED_RECORDS_PER_ITERATION) {
break;
}
} else {
if (pulled.isPenalized()) {
sourceQueue.add(pulled);
break;
}
destination.add(pulled);
}
drainedSize += pulled.getSize();
}
return drainedSize;
}
public void finish() throws IOException {
Queue<IndexRecord> pendingIndexRecord = new LinkedList<IndexRecord>();
for (int i = 0; i < indexBlockBuilders.size(); i++) {
IndexBlock.Builder levelBuilder = indexBlockBuilders.get(i);
if (!pendingIndexRecord.isEmpty()) {
levelBuilder.addRecord(pendingIndexRecord.poll());
}
IndexRecord nextLevelRecord = writeIndexBlock(levelBuilder.build());
pendingIndexRecord.add(nextLevelRecord);
}
IndexRecord rootIndexRecord = pendingIndexRecord.poll();
indexFile.appendInt(rootIndexRecord.blockSize());
indexFile.appendLong(rootIndexRecord.blockOffset());
indexFile.close();
}
void consumeStage(CompletionStage<Boolean> stage) {
Queue<CompletionStage<Boolean>> q = queue;
q.offer(stage);
if (getAndIncrement() == 0) {
do {
stage = q.poll();
stage.whenComplete((b, e) -> {
if (e != null) {
bh.consume(e);
cdl.countDown();
} else
if (b) {
bh.consume(source.current());
consumeStage(source.moveNext(csub));
} else {
bh.consume(false);
cdl.countDown();
}
});
} while (decrementAndGet() != 0);
}
}
public void deleteNode(NodeModel<String> node) {
//设置current的选择
setCurrentSelectedNode(node.getParentNode());
NodeModel<String> parentNode = node.getParentNode();
if (parentNode != null) {
//切断
mTreeModel.removeNode(parentNode, node);
}
//清理碎片
Queue<NodeModel<String>> queue = new ArrayDeque<>();
queue.add(node);
while (!queue.isEmpty()) {
NodeModel<String> poll = queue.poll();
NodeView treeNodeView = (NodeView) findNodeViewFromNodeModel(poll);
removeView(treeNodeView);
for (NodeModel<String> nm : poll.getChildNodes()) {
queue.add(nm);
}
}
}
/**
* Returns a new BTree with the provided set inserting/replacing as necessary any equal items
*
* @param btree the tree to update
* @param comparator the comparator that defines the ordering over the items in the tree
* @param updateWith the items to either insert / update
* @param updateWithIsSorted if false, updateWith will be copied and sorted to facilitate construction
* @param updateF the update function to apply to any pairs we are swapping, and maybe abort early
* @param <V>
* @return
*/
public static <V> Object[] update(Object[] btree,
Comparator<V> comparator,
Iterable<V> updateWith,
int updateWithLength,
boolean updateWithIsSorted,
UpdateFunction<V> updateF)
{
if (btree.length == 0)
return build(updateWith, updateWithLength, comparator, updateWithIsSorted, updateF);
if (!updateWithIsSorted)
updateWith = sorted(updateWith, comparator, updateWithLength);
Queue<Builder> queue = modifier.get();
Builder builder = queue.poll();
if (builder == null)
builder = new Builder();
btree = builder.update(btree, comparator, updateWith, updateF);
queue.add(builder);
return btree;
}
protected FileCacheService()
{
RemovalListener<CacheKey, CacheBucket> onRemove = new RemovalListener<CacheKey, CacheBucket>()
{
@Override
public void onRemoval(RemovalNotification<CacheKey, CacheBucket> notification)
{
CacheBucket bucket = notification.getValue();
if (bucket == null)
return;
// set discarded before deallocating the readers, to ensure we don't leak any
bucket.discarded = true;
Queue<RandomAccessReader> q = bucket.queue;
boolean first = true;
for (RandomAccessReader reader = q.poll() ; reader != null ; reader = q.poll())
{
if (logger.isDebugEnabled() && first)
{
logger.debug("Evicting cold readers for {}", reader.getPath());
first = false;
}
memoryUsage.addAndGet(-1 * reader.getTotalBufferSize());
reader.deallocate();
}
}
};
cache = CacheBuilder.newBuilder()
.expireAfterAccess(AFTER_ACCESS_EXPIRATION, TimeUnit.MILLISECONDS)
.concurrencyLevel(DatabaseDescriptor.getConcurrentReaders())
.removalListener(onRemove)
.initialCapacity(16 << 10)
.build();
}
private static int expandIslandReturnSquareWithoutRecursionBFS(
int[][] islandMatrix, int i, int j) {
if (islandMatrix[i][j] == 0) {
return 0;
}
Queue<Pair> queue = new ArrayDeque<Pair>();
queue.add(new Pair(i, j));
int sum = 0;
while (!queue.isEmpty()) {
Pair element = queue.poll();
int curI = element.getFirst();
int curJ = element.getSecond();
if (islandMatrix[curI][curJ] != 0) {
sum += 1;
islandMatrix[curI][curJ] = 0;
// up, down left, right
addToQueueIfIsland(islandMatrix, queue, curI + 1, curJ);
addToQueueIfIsland(islandMatrix, queue, curI, curJ + 1);
addToQueueIfIsland(islandMatrix, queue, curI - 1, curJ);
addToQueueIfIsland(islandMatrix, queue, curI, curJ - 1);
// south-west, south-east, north-west, north-east
addToQueueIfIsland(islandMatrix, queue, curI + 1, curJ + 1);
addToQueueIfIsland(islandMatrix, queue, curI + 1, curJ - 1);
addToQueueIfIsland(islandMatrix, queue, curI - 1, curJ + 1);
addToQueueIfIsland(islandMatrix, queue, curI - 1, curJ - 1);
}
}
return sum;
}
/**
* @param topN {@link Queue} of items from which to take top n
* @param n how many top values to choose
* @return order list of top results
*/
public static List<RecommendedItem> selectTopNFromQueue(Queue<MutableRecommendedItem> topN, int n) {
if (topN.isEmpty()) {
return Collections.emptyList();
}
while (topN.size() > n) {
topN.poll();
}
List<RecommendedItem> result = Lists.<RecommendedItem>newArrayList(topN);
Collections.sort(result, Collections.reverseOrder(ByValueAscComparator.INSTANCE));
return result;
}
private void resizeOnce(int capacity) {
int nextNumberToPut = 0;
Queue<Integer> referenceQueue = new ArrayBlockingQueue<>(capacity);
CircularBuffer<Integer> buffer = new CircularBuffer<>(capacity);
// Fill full, so the next add will wrap
for (int i = 0; i < capacity; i++, nextNumberToPut++) {
buffer.add(nextNumberToPut);
referenceQueue.add(nextNumberToPut);
}
int gets = r.nextInt(capacity); // [0, capacity)
for (int i = 0; i < gets; i++) {
referenceQueue.poll();
buffer.remove();
}
int puts = r.nextInt(gets + 1); // [0, gets]
for (int i = 0; i < puts; i++, nextNumberToPut++) {
buffer.add(nextNumberToPut);
referenceQueue.add(nextNumberToPut);
}
Integer[] expected = referenceQueue.toArray(new Integer[0]);
buffer.resize(expected.length);
assertEquals(buffer.elements, expected);
}
public boolean canReach(int[] arr, int start) {
if (arr[start] == 0) {
return true;
}
int len = arr.length;
boolean[] visited = new boolean[len];
Queue<Integer> queue = new LinkedList<>();
queue.offer(start);
visited[start] = true;
while (!queue.isEmpty()) {
Integer top = queue.poll();
int right = top + arr[top];
int left = top - arr[top];
if (right < len && !visited[right]) {
visited[right] = true;
queue.offer(right);
if (arr[right] == 0) {
return true;
}
}
if (left >= 0 && !visited[left]) {
visited[left] = true;
queue.offer(left);
if (arr[left] == 0) {
return true;
}
}
}
return false;
}
public StringBuffer doBFS(TreeNode root)
{
StringBuffer sBuffer=new StringBuffer();
Queue<TreeNode> queue=new LinkedList<TreeNode>();
queue.add(root);
root.visited=true;
while(!queue.isEmpty())
{
TreeNode n=queue.poll();
sBuffer.append(n.data);
sBuffer.append(",");
//System.out.print(n.data+",");
TreeNode left=n.left;
TreeNode right=n.right;
if(n.left!=null&&n.left.visited==false)
{
queue.add(left);
}
if(n.right!=null&&n.right.visited==false)
{
queue.add(right);
}
n.visited=true;
}
return sBuffer;
}
private long bfs() {
// Initialize BFS queue and add starting source node.
Queue<Integer> q = new ArrayDeque<>(n);
visit(s);
q.offer(s);
// Perform BFS from source to sink
Edge[] prev = new Edge[n];
while (!q.isEmpty()) {
int node = q.poll();
if (node == t) break;
for (Edge edge : graph[node]) {
long cap = edge.remainingCapacity();
if (cap > 0 && !visited(edge.to)) {
visit(edge.to);
prev[edge.to] = edge;
q.offer(edge.to);
}
}
}
// Sink not reachable!
if (prev[t] == null) return 0;
// Find augmented path and bottle neck
long bottleNeck = Long.MAX_VALUE;
for (Edge edge = prev[t]; edge != null; edge = prev[edge.from])
bottleNeck = min(bottleNeck, edge.remainingCapacity());
// Retrace augmented path and update flow values.
for (Edge edge = prev[t]; edge != null; edge = prev[edge.from]) edge.augment(bottleNeck);
// Return bottleneck flow
return bottleNeck;
}
private String extractTargetTableName(Queue<String> tokens) throws SQLException {
/* Drops FROM keyword */
tokens.poll();
if (!Constants.TABLE.equals(tokens.peek())) {
throw new SQLException("'TABLE' keyword is expected");
}
tokens.poll();
if (!ParserUtil.isStringLiteral(tokens.peek())) {
throw new SQLException("Syntax Error : String literal is expected");
}
return tokens.poll();
}
private void assertSaveMessageWithAttachments(Message msg, int num) throws IOException {
store.save(msg);
List<BodyPart> attList = new LinkedList<BodyPart>();
MessageStoreImpl.recursiveMultipartProcessing((Multipart) msg.getBody(), new StringBuilder(), new StringBuilder(), false, attList);
@SuppressWarnings("unchecked")
Queue<BodyPart> attachmentsMsg = (Queue<BodyPart>) attList;
assertTrue("No attachments found", attachmentsMsg.size() > 0);
assertEquals("", num, attachmentsMsg.size());
final Resource r = resolver.getResource(getResourcePath(msg, store));
assertNotNull("Expecting non-null Resource", r);
for (Resource aRes : r.getChildren()) {
final ModifiableValueMap aMap = aRes.adaptTo(ModifiableValueMap.class);
BodyPart aMsg = attachmentsMsg.poll();
assertNotNull("JCR contains more attachments", aMsg);
for (Field f : aMsg.getHeader().getFields()) {
String name = f.getName();
assertEquals("Field "+name+" is different", (aMap.get(name, String.class)), f.getBody());
}
if (aMsg.getBody() instanceof TextBody) {
assertEquals("Content is not the same", MessageStoreImpl.getTextPart(aMsg), aMap.get(CONTENT, String.class));
} else if (aMsg.getBody() instanceof BinaryBody) {
assertEquals("Content is not the same", getBinPart(aMsg), aMap.get(CONTENT, String.class));
} else {
fail("Unknown type of attachment body");
}
}
assertEquals("Message contains more attachments", attachmentsMsg.poll(), null);
}
@Override
public void handleSplitsChanges(Queue<SplitsChange<MockSourceSplit>> splitsChanges) {
do {
SplitsChange<MockSourceSplit> splitsChange = splitsChanges.poll();
if (splitsChange instanceof SplitsAddition) {
splitsChange.splits().forEach(s -> splits.put(s.splitId(), s));
}
} while (handleSplitsInOneShot && !splitsChanges.isEmpty());
}
/**
* Extract the JSON field key set from the response class. The keys are identified in two ways.
* <ul>
* <li>Any (static) field in response class or its super class (if exists) and is annotated with {@link JsonResponseField}.</li>
* <li>If the response class is annotated with {@link JsonResponseExternalFields} , then any (static) field in the class
* referenced by {@link JsonResponseExternalFields#value()} and is annotated with {@link JsonResponseField}.</li>
* </ul>
*
* @param responseClass The Java response class.
* @return A map of field key to its necessity (i.e. whether it is a required field in JSON response).
*/
private Map<String, Boolean> extractFieldKeys(Class responseClass) {
Map<String, Boolean> fields = new HashMap<>();
Queue<Class> classToScan = new LinkedList<>();
classToScan.add(responseClass);
while (!classToScan.isEmpty()) {
Class currentClass = classToScan.poll();
Arrays.stream(currentClass.getDeclaredFields())
.filter(f -> Modifier.isStatic(f.getModifiers()) && f.isAnnotationPresent(JsonResponseField.class))
.forEach(f -> {
f.setAccessible(true);
try {
fields.put(f.get(null).toString(), f.getAnnotation(JsonResponseField.class).required());
} catch (IllegalAccessException e) {
// Not reach here.
}
});
// Scan for super class.
Class superClass = currentClass.getSuperclass();
if (superClass != null && superClass.isAnnotationPresent(JsonResponseClass.class)) {
classToScan.offer(superClass);
}
// Scan for referenced class.
if (currentClass.isAnnotationPresent(JsonResponseExternalFields.class)) {
classToScan.offer(((JsonResponseExternalFields) currentClass.getAnnotation(JsonResponseExternalFields.class)).value());
}
}
return fields;
}
/**
* Fuses a {@link Pipeline} into a collection of {@link ExecutableStage}.
*
* <p>The input is the initial collection of siblings sets which will be fused into {@link
* ExecutableStage stages}. A sibling in this context represents a pair of (PCollection,
* PTransform), where the PTransform consumes input elements on a per-element basis from the
* PCollection, represented by a {@link CollectionConsumer}. A sibling set is a collection of
* siblings which can execute within a single {@link ExecutableStage}, determined by {@link
* GreedyPCollectionFusers#isCompatible(PTransformNode, PTransformNode, QueryablePipeline)}.
*
* <p>While a pending sibling set exists:
*
* <ul>
* <li>Retrieve a pending sibling set from the front of the queue.
* <li>If the pending sibling set has already been created, continue. Each materialized {@link
* PTransformNode} can be consumed by any number of {@link ExecutableStage stages}, but each
* {@link PTransformNode} may only be present in a single stage rooted at a single {@link
* PCollectionNode}, otherwise it will process elements of that {@link PCollectionNode}
* multiple times.
* <li>Create a {@link GreedyStageFuser} with those siblings as the initial consuming transforms
* of the stage
* <li>For each materialized {@link PCollectionNode}, find all of the descendant in-environment
* consumers. See {@link #getDescendantConsumers(PCollectionNode)} for details.
* <li>Construct all of the sibling sets from the descendant in-environment consumers, and add
* them to the queue of sibling sets.
* </ul>
*/
private FusedPipeline fusePipeline(
Collection<PTransformNode> initialUnfusedTransforms,
NavigableSet<NavigableSet<CollectionConsumer>> initialConsumers,
Set<String> requirements) {
Map<CollectionConsumer, ExecutableStage> consumedCollectionsAndTransforms = new HashMap<>();
Set<ExecutableStage> stages = new LinkedHashSet<>();
Set<PTransformNode> unfusedTransforms = new LinkedHashSet<>(initialUnfusedTransforms);
Queue<Set<CollectionConsumer>> pendingSiblingSets = new ArrayDeque<>(initialConsumers);
while (!pendingSiblingSets.isEmpty()) {
// Only introduce new PCollection consumers. Not performing this introduces potential
// duplicate paths through the pipeline.
Set<CollectionConsumer> candidateSiblings = pendingSiblingSets.poll();
Set<CollectionConsumer> siblingSet =
Sets.difference(candidateSiblings, consumedCollectionsAndTransforms.keySet());
checkState(
siblingSet.equals(candidateSiblings) || siblingSet.isEmpty(),
"Inconsistent collection of siblings reported for a %s. Initial attempt missed %s",
PCollectionNode.class.getSimpleName(),
siblingSet);
if (siblingSet.isEmpty()) {
LOG.debug("Filtered out duplicate stage root {}", candidateSiblings);
continue;
}
// Create the stage with these siblings as the initial consuming transforms
ExecutableStage stage = fuseSiblings(siblingSet);
// Mark each of the root transforms of the stage as consuming the input PCollection, so we
// don't place them in multiple stages.
for (CollectionConsumer sibling : siblingSet) {
consumedCollectionsAndTransforms.put(sibling, stage);
}
stages.add(stage);
for (PCollectionNode materializedOutput : stage.getOutputPCollections()) {
// Get all of the descendant consumers of each materialized PCollection, and add them to the
// queue of pending siblings.
DescendantConsumers descendantConsumers = getDescendantConsumers(materializedOutput);
unfusedTransforms.addAll(descendantConsumers.getUnfusedNodes());
NavigableSet<NavigableSet<CollectionConsumer>> siblings =
groupSiblings(descendantConsumers.getFusibleConsumers());
pendingSiblingSets.addAll(siblings);
}
}
// TODO: Figure out where to store this.
DeduplicationResult deduplicated =
OutputDeduplicator.ensureSingleProducer(pipeline, stages, unfusedTransforms);
// TODO: Stages can be fused with each other, if doing so does not introduce duplicate paths
// for an element to take through the Pipeline. Compatible siblings can generally be fused,
// as can compatible producers/consumers if a PCollection is only materialized once.
return FusedPipeline.of(
deduplicated.getDeduplicatedComponents(),
stages.stream()
.map(stage -> deduplicated.getDeduplicatedStages().getOrDefault(stage, stage))
.map(GreedyPipelineFuser::sanitizeDanglingPTransformInputs)
.collect(Collectors.toSet()),
Sets.union(
deduplicated.getIntroducedTransforms(),
unfusedTransforms.stream()
.map(
transform ->
deduplicated
.getDeduplicatedTransforms()
.getOrDefault(transform.getId(), transform))
.collect(Collectors.toSet())),
requirements);
}
void drainRegular(CoreSubscriber<? super T> a) {
int missed = 1;
final Queue<T> q = queue;
for (;;) {
long r = requested;
long e = 0L;
while (r != e) {
boolean d = done;
T t = q.poll();
boolean empty = t == null;
if (checkTerminated(d, empty, a, q, t)) {
return;
}
if (empty) {
break;
}
a.onNext(t);
e++;
}
if (r == e) {
if (checkTerminated(done, q.isEmpty(), a, q, null)) {
return;
}
}
if (e != 0 && r != Long.MAX_VALUE) {
REQUESTED.addAndGet(this, -e);
}
missed = WIP.addAndGet(this, -missed);
if (missed == 0) {
break;
}
}
}
public Intention selectIntention(Queue<Intention> intentions) {
// make sure the selected Intention is removed from 'intentions'
// and make sure no intention will "starve"!!!
return intentions.poll();
}
/**
* Collect cuboid from bottom up, considering all factor including black list
* Build tree steps:
* 1. Build tree from bottom up considering dim capping
* 2. Kick out blacked-out cuboids from the tree
* 3. Make sure all the cuboids have proper "parent", if not add it to the tree.
* Direct parent is not necessary, can jump *forward* steps to find in-direct parent.
* For example, forward = 1, grandparent can also be the parent. Only if both parent
* and grandparent are missing, add grandparent to the tree.
* @return Cuboid collection
*/
protected Pair<Set<Long>, Map<Long, List<Long>>> buildTreeBottomUp() {
int forward = cubeDesc.getParentForward();
KylinConfig config = cubeDesc.getConfig();
Set<Long> cuboidHolder = new HashSet<>();
// build tree structure
Set<Long> children = getLowestCuboids();
long maxCombination = config.getCubeAggrGroupMaxCombination() * 10;
maxCombination = maxCombination < 0 ? Long.MAX_VALUE : maxCombination;
while (!children.isEmpty()) {
cuboidHolder.addAll(children);
if (cuboidHolder.size() > maxCombination) {
throw new IllegalStateException("Too many cuboids for the cube. Cuboid combination reached "
+ cuboidHolder.size() + " and limit is " + maxCombination + ". Abort calculation.");
}
children = getOnTreeParentsByLayer(children);
}
cuboidHolder.add(Cuboid.getBaseCuboidId(cubeDesc));
// kick off blacked
cuboidHolder = Sets.newHashSet(Iterators.filter(cuboidHolder.iterator(), new Predicate<Long>() {
@Override
public boolean apply(@Nullable Long cuboidId) {
return !cubeDesc.isBlackedCuboid(cuboidId);
}
}));
// fill padding cuboids
Map<Long, List<Long>> parent2Child = Maps.newHashMap();
Queue<Long> cuboidScan = new ArrayDeque<>();
cuboidScan.addAll(cuboidHolder);
while (!cuboidScan.isEmpty()) {
long current = cuboidScan.poll();
long parent = getParentOnPromise(current, cuboidHolder, forward);
if (parent > 0) {
if (!cuboidHolder.contains(parent)) {
cuboidScan.add(parent);
cuboidHolder.add(parent);
}
if (parent2Child.containsKey(parent)) {
parent2Child.get(parent).add(current);
} else {
parent2Child.put(parent, Lists.newArrayList(current));
}
}
}
return Pair.newPair(cuboidHolder, parent2Child);
}