001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.activemq.broker.region; 018 019import java.io.IOException; 020import java.util.ArrayList; 021import java.util.Collection; 022import java.util.Collections; 023import java.util.Comparator; 024import java.util.HashSet; 025import java.util.Iterator; 026import java.util.LinkedHashMap; 027import java.util.LinkedHashSet; 028import java.util.LinkedList; 029import java.util.List; 030import java.util.Map; 031import java.util.Set; 032import java.util.concurrent.CancellationException; 033import java.util.concurrent.ConcurrentLinkedQueue; 034import java.util.concurrent.CountDownLatch; 035import java.util.concurrent.DelayQueue; 036import java.util.concurrent.Delayed; 037import java.util.concurrent.ExecutorService; 038import java.util.concurrent.TimeUnit; 039import java.util.concurrent.atomic.AtomicBoolean; 040import java.util.concurrent.atomic.AtomicLong; 041import java.util.concurrent.locks.Lock; 042import java.util.concurrent.locks.ReentrantLock; 043import java.util.concurrent.locks.ReentrantReadWriteLock; 044 045import javax.jms.InvalidSelectorException; 046import javax.jms.JMSException; 047import javax.jms.ResourceAllocationException; 048 049import org.apache.activemq.broker.BrokerService; 050import org.apache.activemq.broker.BrokerStoppedException; 051import org.apache.activemq.broker.ConnectionContext; 052import org.apache.activemq.broker.ProducerBrokerExchange; 053import org.apache.activemq.broker.region.cursors.OrderedPendingList; 054import org.apache.activemq.broker.region.cursors.PendingList; 055import org.apache.activemq.broker.region.cursors.PendingMessageCursor; 056import org.apache.activemq.broker.region.cursors.PrioritizedPendingList; 057import org.apache.activemq.broker.region.cursors.QueueDispatchPendingList; 058import org.apache.activemq.broker.region.cursors.StoreQueueCursor; 059import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor; 060import org.apache.activemq.broker.region.group.CachedMessageGroupMapFactory; 061import org.apache.activemq.broker.region.group.MessageGroupMap; 062import org.apache.activemq.broker.region.group.MessageGroupMapFactory; 063import org.apache.activemq.broker.region.policy.DeadLetterStrategy; 064import org.apache.activemq.broker.region.policy.DispatchPolicy; 065import org.apache.activemq.broker.region.policy.RoundRobinDispatchPolicy; 066import org.apache.activemq.broker.util.InsertionCountList; 067import org.apache.activemq.command.ActiveMQDestination; 068import org.apache.activemq.command.ConsumerId; 069import org.apache.activemq.command.ExceptionResponse; 070import org.apache.activemq.command.Message; 071import org.apache.activemq.command.MessageAck; 072import org.apache.activemq.command.MessageDispatchNotification; 073import org.apache.activemq.command.MessageId; 074import org.apache.activemq.command.ProducerAck; 075import org.apache.activemq.command.ProducerInfo; 076import org.apache.activemq.command.RemoveInfo; 077import org.apache.activemq.command.Response; 078import org.apache.activemq.filter.BooleanExpression; 079import org.apache.activemq.filter.MessageEvaluationContext; 080import org.apache.activemq.filter.NonCachedMessageEvaluationContext; 081import org.apache.activemq.selector.SelectorParser; 082import org.apache.activemq.state.ProducerState; 083import org.apache.activemq.store.IndexListener; 084import org.apache.activemq.store.ListenableFuture; 085import org.apache.activemq.store.MessageRecoveryListener; 086import org.apache.activemq.store.MessageStore; 087import org.apache.activemq.thread.Task; 088import org.apache.activemq.thread.TaskRunner; 089import org.apache.activemq.thread.TaskRunnerFactory; 090import org.apache.activemq.transaction.Synchronization; 091import org.apache.activemq.usage.Usage; 092import org.apache.activemq.usage.UsageListener; 093import org.apache.activemq.util.BrokerSupport; 094import org.apache.activemq.util.ThreadPoolUtils; 095import org.slf4j.Logger; 096import org.slf4j.LoggerFactory; 097import org.slf4j.MDC; 098 099/** 100 * The Queue is a List of MessageEntry objects that are dispatched to matching 101 * subscriptions. 102 */ 103public class Queue extends BaseDestination implements Task, UsageListener, IndexListener { 104 protected static final Logger LOG = LoggerFactory.getLogger(Queue.class); 105 protected final TaskRunnerFactory taskFactory; 106 protected TaskRunner taskRunner; 107 private final ReentrantReadWriteLock consumersLock = new ReentrantReadWriteLock(); 108 protected final List<Subscription> consumers = new ArrayList<Subscription>(50); 109 private final ReentrantReadWriteLock messagesLock = new ReentrantReadWriteLock(); 110 protected PendingMessageCursor messages; 111 private final ReentrantReadWriteLock pagedInMessagesLock = new ReentrantReadWriteLock(); 112 private final PendingList pagedInMessages = new OrderedPendingList(); 113 // Messages that are paged in but have not yet been targeted at a subscription 114 private final ReentrantReadWriteLock pagedInPendingDispatchLock = new ReentrantReadWriteLock(); 115 protected QueueDispatchPendingList dispatchPendingList = new QueueDispatchPendingList(); 116 private MessageGroupMap messageGroupOwners; 117 private DispatchPolicy dispatchPolicy = new RoundRobinDispatchPolicy(); 118 private MessageGroupMapFactory messageGroupMapFactory = new CachedMessageGroupMapFactory(); 119 final Lock sendLock = new ReentrantLock(); 120 private ExecutorService executor; 121 private final Map<MessageId, Runnable> messagesWaitingForSpace = new LinkedHashMap<MessageId, Runnable>(); 122 private boolean useConsumerPriority = true; 123 private boolean strictOrderDispatch = false; 124 private final QueueDispatchSelector dispatchSelector; 125 private boolean optimizedDispatch = false; 126 private boolean iterationRunning = false; 127 private boolean firstConsumer = false; 128 private int timeBeforeDispatchStarts = 0; 129 private int consumersBeforeDispatchStarts = 0; 130 private CountDownLatch consumersBeforeStartsLatch; 131 private final AtomicLong pendingWakeups = new AtomicLong(); 132 private boolean allConsumersExclusiveByDefault = false; 133 private final AtomicBoolean started = new AtomicBoolean(); 134 135 private volatile boolean resetNeeded; 136 137 private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() { 138 @Override 139 public void run() { 140 asyncWakeup(); 141 } 142 }; 143 private final Runnable expireMessagesTask = new Runnable() { 144 @Override 145 public void run() { 146 expireMessages(); 147 } 148 }; 149 150 private final Object iteratingMutex = new Object(); 151 152 153 154 class TimeoutMessage implements Delayed { 155 156 Message message; 157 ConnectionContext context; 158 long trigger; 159 160 public TimeoutMessage(Message message, ConnectionContext context, long delay) { 161 this.message = message; 162 this.context = context; 163 this.trigger = System.currentTimeMillis() + delay; 164 } 165 166 @Override 167 public long getDelay(TimeUnit unit) { 168 long n = trigger - System.currentTimeMillis(); 169 return unit.convert(n, TimeUnit.MILLISECONDS); 170 } 171 172 @Override 173 public int compareTo(Delayed delayed) { 174 long other = ((TimeoutMessage) delayed).trigger; 175 int returnValue; 176 if (this.trigger < other) { 177 returnValue = -1; 178 } else if (this.trigger > other) { 179 returnValue = 1; 180 } else { 181 returnValue = 0; 182 } 183 return returnValue; 184 } 185 } 186 187 DelayQueue<TimeoutMessage> flowControlTimeoutMessages = new DelayQueue<TimeoutMessage>(); 188 189 class FlowControlTimeoutTask extends Thread { 190 191 @Override 192 public void run() { 193 TimeoutMessage timeout; 194 try { 195 while (true) { 196 timeout = flowControlTimeoutMessages.take(); 197 if (timeout != null) { 198 synchronized (messagesWaitingForSpace) { 199 if (messagesWaitingForSpace.remove(timeout.message.getMessageId()) != null) { 200 ExceptionResponse response = new ExceptionResponse( 201 new ResourceAllocationException( 202 "Usage Manager Memory Limit reached. Stopping producer (" 203 + timeout.message.getProducerId() 204 + ") to prevent flooding " 205 + getActiveMQDestination().getQualifiedName() 206 + "." 207 + " See http://activemq.apache.org/producer-flow-control.html for more info")); 208 response.setCorrelationId(timeout.message.getCommandId()); 209 timeout.context.getConnection().dispatchAsync(response); 210 } 211 } 212 } 213 } 214 } catch (InterruptedException e) { 215 LOG.debug(getName() + "Producer Flow Control Timeout Task is stopping"); 216 } 217 } 218 }; 219 220 private final FlowControlTimeoutTask flowControlTimeoutTask = new FlowControlTimeoutTask(); 221 222 private final Comparator<Subscription> orderedCompare = new Comparator<Subscription>() { 223 224 @Override 225 public int compare(Subscription s1, Subscription s2) { 226 // We want the list sorted in descending order 227 int val = s2.getConsumerInfo().getPriority() - s1.getConsumerInfo().getPriority(); 228 if (val == 0 && messageGroupOwners != null) { 229 // then ascending order of assigned message groups to favour less loaded consumers 230 // Long.compare in jdk7 231 long x = s1.getConsumerInfo().getAssignedGroupCount(destination); 232 long y = s2.getConsumerInfo().getAssignedGroupCount(destination); 233 val = (x < y) ? -1 : ((x == y) ? 0 : 1); 234 } 235 return val; 236 } 237 }; 238 239 public Queue(BrokerService brokerService, final ActiveMQDestination destination, MessageStore store, 240 DestinationStatistics parentStats, TaskRunnerFactory taskFactory) throws Exception { 241 super(brokerService, store, destination, parentStats); 242 this.taskFactory = taskFactory; 243 this.dispatchSelector = new QueueDispatchSelector(destination); 244 if (store != null) { 245 store.registerIndexListener(this); 246 } 247 } 248 249 @Override 250 public List<Subscription> getConsumers() { 251 consumersLock.readLock().lock(); 252 try { 253 return new ArrayList<Subscription>(consumers); 254 } finally { 255 consumersLock.readLock().unlock(); 256 } 257 } 258 259 // make the queue easily visible in the debugger from its task runner 260 // threads 261 final class QueueThread extends Thread { 262 final Queue queue; 263 264 public QueueThread(Runnable runnable, String name, Queue queue) { 265 super(runnable, name); 266 this.queue = queue; 267 } 268 } 269 270 class BatchMessageRecoveryListener implements MessageRecoveryListener { 271 final LinkedList<Message> toExpire = new LinkedList<Message>(); 272 final double totalMessageCount; 273 int recoveredAccumulator = 0; 274 int currentBatchCount; 275 276 BatchMessageRecoveryListener(int totalMessageCount) { 277 this.totalMessageCount = totalMessageCount; 278 currentBatchCount = recoveredAccumulator; 279 } 280 281 @Override 282 public boolean recoverMessage(Message message) { 283 recoveredAccumulator++; 284 if ((recoveredAccumulator % 10000) == 0) { 285 LOG.info("cursor for {} has recovered {} messages. {}% complete", new Object[]{ getActiveMQDestination().getQualifiedName(), recoveredAccumulator, new Integer((int) (recoveredAccumulator * 100 / totalMessageCount))}); 286 } 287 // Message could have expired while it was being 288 // loaded.. 289 message.setRegionDestination(Queue.this); 290 if (message.isExpired() && broker.isExpired(message)) { 291 toExpire.add(message); 292 return true; 293 } 294 if (hasSpace()) { 295 messagesLock.writeLock().lock(); 296 try { 297 try { 298 messages.addMessageLast(message); 299 } catch (Exception e) { 300 LOG.error("Failed to add message to cursor", e); 301 } 302 } finally { 303 messagesLock.writeLock().unlock(); 304 } 305 destinationStatistics.getMessages().increment(); 306 return true; 307 } 308 return false; 309 } 310 311 @Override 312 public boolean recoverMessageReference(MessageId messageReference) throws Exception { 313 throw new RuntimeException("Should not be called."); 314 } 315 316 @Override 317 public boolean hasSpace() { 318 return true; 319 } 320 321 @Override 322 public boolean isDuplicate(MessageId id) { 323 return false; 324 } 325 326 public void reset() { 327 currentBatchCount = recoveredAccumulator; 328 } 329 330 public void processExpired() { 331 for (Message message: toExpire) { 332 messageExpired(createConnectionContext(), createMessageReference(message)); 333 // drop message will decrement so counter 334 // balance here 335 destinationStatistics.getMessages().increment(); 336 } 337 toExpire.clear(); 338 } 339 340 public boolean done() { 341 return currentBatchCount == recoveredAccumulator; 342 } 343 } 344 345 @Override 346 public void setPrioritizedMessages(boolean prioritizedMessages) { 347 super.setPrioritizedMessages(prioritizedMessages); 348 dispatchPendingList.setPrioritizedMessages(prioritizedMessages); 349 } 350 351 @Override 352 public void initialize() throws Exception { 353 354 if (this.messages == null) { 355 if (destination.isTemporary() || broker == null || store == null) { 356 this.messages = new VMPendingMessageCursor(isPrioritizedMessages()); 357 } else { 358 this.messages = new StoreQueueCursor(broker, this); 359 } 360 } 361 362 // If a VMPendingMessageCursor don't use the default Producer System 363 // Usage 364 // since it turns into a shared blocking queue which can lead to a 365 // network deadlock. 366 // If we are cursoring to disk..it's not and issue because it does not 367 // block due 368 // to large disk sizes. 369 if (messages instanceof VMPendingMessageCursor) { 370 this.systemUsage = brokerService.getSystemUsage(); 371 memoryUsage.setParent(systemUsage.getMemoryUsage()); 372 } 373 374 this.taskRunner = taskFactory.createTaskRunner(this, "Queue:" + destination.getPhysicalName()); 375 376 super.initialize(); 377 if (store != null) { 378 // Restore the persistent messages. 379 messages.setSystemUsage(systemUsage); 380 messages.setEnableAudit(isEnableAudit()); 381 messages.setMaxAuditDepth(getMaxAuditDepth()); 382 messages.setMaxProducersToAudit(getMaxProducersToAudit()); 383 messages.setUseCache(isUseCache()); 384 messages.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark()); 385 store.start(); 386 final int messageCount = store.getMessageCount(); 387 if (messageCount > 0 && messages.isRecoveryRequired()) { 388 BatchMessageRecoveryListener listener = new BatchMessageRecoveryListener(messageCount); 389 do { 390 listener.reset(); 391 store.recoverNextMessages(getMaxPageSize(), listener); 392 listener.processExpired(); 393 } while (!listener.done()); 394 } else { 395 destinationStatistics.getMessages().add(messageCount); 396 } 397 } 398 } 399 400 /* 401 * Holder for subscription that needs attention on next iterate browser 402 * needs access to existing messages in the queue that have already been 403 * dispatched 404 */ 405 class BrowserDispatch { 406 QueueBrowserSubscription browser; 407 408 public BrowserDispatch(QueueBrowserSubscription browserSubscription) { 409 browser = browserSubscription; 410 browser.incrementQueueRef(); 411 } 412 413 void done() { 414 try { 415 browser.decrementQueueRef(); 416 } catch (Exception e) { 417 LOG.warn("decrement ref on browser: " + browser, e); 418 } 419 } 420 421 public QueueBrowserSubscription getBrowser() { 422 return browser; 423 } 424 } 425 426 ConcurrentLinkedQueue<BrowserDispatch> browserDispatches = new ConcurrentLinkedQueue<BrowserDispatch>(); 427 428 @Override 429 public void addSubscription(ConnectionContext context, Subscription sub) throws Exception { 430 LOG.debug("{} add sub: {}, dequeues: {}, dispatched: {}, inflight: {}", new Object[]{ getActiveMQDestination().getQualifiedName(), sub, getDestinationStatistics().getDequeues().getCount(), getDestinationStatistics().getDispatched().getCount(), getDestinationStatistics().getInflight().getCount() }); 431 432 super.addSubscription(context, sub); 433 // synchronize with dispatch method so that no new messages are sent 434 // while setting up a subscription. avoid out of order messages, 435 // duplicates, etc. 436 pagedInPendingDispatchLock.writeLock().lock(); 437 try { 438 439 sub.add(context, this); 440 441 // needs to be synchronized - so no contention with dispatching 442 // consumersLock. 443 consumersLock.writeLock().lock(); 444 try { 445 // set a flag if this is a first consumer 446 if (consumers.size() == 0) { 447 firstConsumer = true; 448 if (consumersBeforeDispatchStarts != 0) { 449 consumersBeforeStartsLatch = new CountDownLatch(consumersBeforeDispatchStarts - 1); 450 } 451 } else { 452 if (consumersBeforeStartsLatch != null) { 453 consumersBeforeStartsLatch.countDown(); 454 } 455 } 456 457 addToConsumerList(sub); 458 if (sub.getConsumerInfo().isExclusive() || isAllConsumersExclusiveByDefault()) { 459 Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer(); 460 if (exclusiveConsumer == null) { 461 exclusiveConsumer = sub; 462 } else if (sub.getConsumerInfo().getPriority() == Byte.MAX_VALUE || 463 sub.getConsumerInfo().getPriority() > exclusiveConsumer.getConsumerInfo().getPriority()) { 464 exclusiveConsumer = sub; 465 } 466 dispatchSelector.setExclusiveConsumer(exclusiveConsumer); 467 } 468 } finally { 469 consumersLock.writeLock().unlock(); 470 } 471 472 if (sub instanceof QueueBrowserSubscription) { 473 // tee up for dispatch in next iterate 474 QueueBrowserSubscription browserSubscription = (QueueBrowserSubscription) sub; 475 BrowserDispatch browserDispatch = new BrowserDispatch(browserSubscription); 476 browserDispatches.add(browserDispatch); 477 } 478 479 if (!this.optimizedDispatch) { 480 wakeup(); 481 } 482 } finally { 483 pagedInPendingDispatchLock.writeLock().unlock(); 484 } 485 if (this.optimizedDispatch) { 486 // Outside of dispatchLock() to maintain the lock hierarchy of 487 // iteratingMutex -> dispatchLock. - see 488 // https://issues.apache.org/activemq/browse/AMQ-1878 489 wakeup(); 490 } 491 } 492 493 @Override 494 public void removeSubscription(ConnectionContext context, Subscription sub, long lastDeliveredSequenceId) 495 throws Exception { 496 super.removeSubscription(context, sub, lastDeliveredSequenceId); 497 // synchronize with dispatch method so that no new messages are sent 498 // while removing up a subscription. 499 pagedInPendingDispatchLock.writeLock().lock(); 500 try { 501 LOG.debug("{} remove sub: {}, lastDeliveredSeqId: {}, dequeues: {}, dispatched: {}, inflight: {}, groups: {}", new Object[]{ 502 getActiveMQDestination().getQualifiedName(), 503 sub, 504 lastDeliveredSequenceId, 505 getDestinationStatistics().getDequeues().getCount(), 506 getDestinationStatistics().getDispatched().getCount(), 507 getDestinationStatistics().getInflight().getCount(), 508 sub.getConsumerInfo().getAssignedGroupCount(destination) 509 }); 510 consumersLock.writeLock().lock(); 511 try { 512 removeFromConsumerList(sub); 513 if (sub.getConsumerInfo().isExclusive()) { 514 Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer(); 515 if (exclusiveConsumer == sub) { 516 exclusiveConsumer = null; 517 for (Subscription s : consumers) { 518 if (s.getConsumerInfo().isExclusive() 519 && (exclusiveConsumer == null || s.getConsumerInfo().getPriority() > exclusiveConsumer 520 .getConsumerInfo().getPriority())) { 521 exclusiveConsumer = s; 522 523 } 524 } 525 dispatchSelector.setExclusiveConsumer(exclusiveConsumer); 526 } 527 } else if (isAllConsumersExclusiveByDefault()) { 528 Subscription exclusiveConsumer = null; 529 for (Subscription s : consumers) { 530 if (exclusiveConsumer == null 531 || s.getConsumerInfo().getPriority() > exclusiveConsumer 532 .getConsumerInfo().getPriority()) { 533 exclusiveConsumer = s; 534 } 535 } 536 dispatchSelector.setExclusiveConsumer(exclusiveConsumer); 537 } 538 ConsumerId consumerId = sub.getConsumerInfo().getConsumerId(); 539 getMessageGroupOwners().removeConsumer(consumerId); 540 541 // redeliver inflight messages 542 543 boolean markAsRedelivered = false; 544 MessageReference lastDeliveredRef = null; 545 List<MessageReference> unAckedMessages = sub.remove(context, this); 546 547 // locate last redelivered in unconsumed list (list in delivery rather than seq order) 548 if (lastDeliveredSequenceId > RemoveInfo.LAST_DELIVERED_UNSET) { 549 for (MessageReference ref : unAckedMessages) { 550 if (ref.getMessageId().getBrokerSequenceId() == lastDeliveredSequenceId) { 551 lastDeliveredRef = ref; 552 markAsRedelivered = true; 553 LOG.debug("found lastDeliveredSeqID: {}, message reference: {}", lastDeliveredSequenceId, ref.getMessageId()); 554 break; 555 } 556 } 557 } 558 559 for (MessageReference ref : unAckedMessages) { 560 // AMQ-5107: don't resend if the broker is shutting down 561 if ( this.brokerService.isStopping() ) { 562 break; 563 } 564 QueueMessageReference qmr = (QueueMessageReference) ref; 565 if (qmr.getLockOwner() == sub) { 566 qmr.unlock(); 567 568 // have no delivery information 569 if (lastDeliveredSequenceId == RemoveInfo.LAST_DELIVERED_UNKNOWN) { 570 qmr.incrementRedeliveryCounter(); 571 } else { 572 if (markAsRedelivered) { 573 qmr.incrementRedeliveryCounter(); 574 } 575 if (ref == lastDeliveredRef) { 576 // all that follow were not redelivered 577 markAsRedelivered = false; 578 } 579 } 580 } 581 if (!qmr.isDropped()) { 582 dispatchPendingList.addMessageForRedelivery(qmr); 583 } 584 } 585 if (sub instanceof QueueBrowserSubscription) { 586 ((QueueBrowserSubscription)sub).decrementQueueRef(); 587 browserDispatches.remove(sub); 588 } 589 // AMQ-5107: don't resend if the broker is shutting down 590 if (dispatchPendingList.hasRedeliveries() && (! this.brokerService.isStopping())) { 591 doDispatch(new OrderedPendingList()); 592 } 593 } finally { 594 consumersLock.writeLock().unlock(); 595 } 596 if (!this.optimizedDispatch) { 597 wakeup(); 598 } 599 } finally { 600 pagedInPendingDispatchLock.writeLock().unlock(); 601 } 602 if (this.optimizedDispatch) { 603 // Outside of dispatchLock() to maintain the lock hierarchy of 604 // iteratingMutex -> dispatchLock. - see 605 // https://issues.apache.org/activemq/browse/AMQ-1878 606 wakeup(); 607 } 608 } 609 610 @Override 611 public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception { 612 final ConnectionContext context = producerExchange.getConnectionContext(); 613 // There is delay between the client sending it and it arriving at the 614 // destination.. it may have expired. 615 message.setRegionDestination(this); 616 ProducerState state = producerExchange.getProducerState(); 617 if (state == null) { 618 LOG.warn("Send failed for: {}, missing producer state for: {}", message, producerExchange); 619 throw new JMSException("Cannot send message to " + getActiveMQDestination() + " with invalid (null) producer state"); 620 } 621 final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo(); 622 final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize() > 0 623 && !context.isInRecoveryMode(); 624 if (message.isExpired()) { 625 // message not stored - or added to stats yet - so chuck here 626 broker.getRoot().messageExpired(context, message, null); 627 if (sendProducerAck) { 628 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); 629 context.getConnection().dispatchAsync(ack); 630 } 631 return; 632 } 633 if (memoryUsage.isFull()) { 634 isFull(context, memoryUsage); 635 fastProducer(context, producerInfo); 636 if (isProducerFlowControl() && context.isProducerFlowControl()) { 637 if (warnOnProducerFlowControl) { 638 warnOnProducerFlowControl = false; 639 LOG.info("Usage Manager Memory Limit ({}) reached on {}, size {}. Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it. See http://activemq.apache.org/producer-flow-control.html for more info.", 640 memoryUsage.getLimit(), getActiveMQDestination().getQualifiedName(), destinationStatistics.getMessages().getCount()); 641 } 642 643 if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) { 644 throw new ResourceAllocationException("Usage Manager Memory Limit reached. Stopping producer (" 645 + message.getProducerId() + ") to prevent flooding " 646 + getActiveMQDestination().getQualifiedName() + "." 647 + " See http://activemq.apache.org/producer-flow-control.html for more info"); 648 } 649 650 // We can avoid blocking due to low usage if the producer is 651 // sending 652 // a sync message or if it is using a producer window 653 if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) { 654 // copy the exchange state since the context will be 655 // modified while we are waiting 656 // for space. 657 final ProducerBrokerExchange producerExchangeCopy = producerExchange.copy(); 658 synchronized (messagesWaitingForSpace) { 659 // Start flow control timeout task 660 // Prevent trying to start it multiple times 661 if (!flowControlTimeoutTask.isAlive()) { 662 flowControlTimeoutTask.setName(getName()+" Producer Flow Control Timeout Task"); 663 flowControlTimeoutTask.start(); 664 } 665 messagesWaitingForSpace.put(message.getMessageId(), new Runnable() { 666 @Override 667 public void run() { 668 669 try { 670 // While waiting for space to free up... the 671 // message may have expired. 672 if (message.isExpired()) { 673 LOG.error("expired waiting for space.."); 674 broker.messageExpired(context, message, null); 675 destinationStatistics.getExpired().increment(); 676 } else { 677 doMessageSend(producerExchangeCopy, message); 678 } 679 680 if (sendProducerAck) { 681 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message 682 .getSize()); 683 context.getConnection().dispatchAsync(ack); 684 } else { 685 Response response = new Response(); 686 response.setCorrelationId(message.getCommandId()); 687 context.getConnection().dispatchAsync(response); 688 } 689 690 } catch (Exception e) { 691 if (!sendProducerAck && !context.isInRecoveryMode() && !brokerService.isStopping()) { 692 ExceptionResponse response = new ExceptionResponse(e); 693 response.setCorrelationId(message.getCommandId()); 694 context.getConnection().dispatchAsync(response); 695 } else { 696 LOG.debug("unexpected exception on deferred send of: {}", message, e); 697 } 698 } 699 } 700 }); 701 702 if (!context.isNetworkConnection() && systemUsage.getSendFailIfNoSpaceAfterTimeout() != 0) { 703 flowControlTimeoutMessages.add(new TimeoutMessage(message, context, systemUsage 704 .getSendFailIfNoSpaceAfterTimeout())); 705 } 706 707 registerCallbackForNotFullNotification(); 708 context.setDontSendReponse(true); 709 return; 710 } 711 712 } else { 713 714 if (memoryUsage.isFull()) { 715 waitForSpace(context, producerExchange, memoryUsage, "Usage Manager Memory Limit reached. Producer (" 716 + message.getProducerId() + ") stopped to prevent flooding " 717 + getActiveMQDestination().getQualifiedName() + "." 718 + " See http://activemq.apache.org/producer-flow-control.html for more info"); 719 } 720 721 // The usage manager could have delayed us by the time 722 // we unblock the message could have expired.. 723 if (message.isExpired()) { 724 LOG.debug("Expired message: {}", message); 725 broker.getRoot().messageExpired(context, message, null); 726 return; 727 } 728 } 729 } 730 } 731 doMessageSend(producerExchange, message); 732 if (sendProducerAck) { 733 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); 734 context.getConnection().dispatchAsync(ack); 735 } 736 } 737 738 private void registerCallbackForNotFullNotification() { 739 // If the usage manager is not full, then the task will not 740 // get called.. 741 if (!memoryUsage.notifyCallbackWhenNotFull(sendMessagesWaitingForSpaceTask)) { 742 // so call it directly here. 743 sendMessagesWaitingForSpaceTask.run(); 744 } 745 } 746 747 private final LinkedList<MessageContext> indexOrderedCursorUpdates = new LinkedList<>(); 748 749 @Override 750 public void onAdd(MessageContext messageContext) { 751 synchronized (indexOrderedCursorUpdates) { 752 indexOrderedCursorUpdates.addLast(messageContext); 753 } 754 } 755 756 private void doPendingCursorAdditions() throws Exception { 757 LinkedList<MessageContext> orderedUpdates = new LinkedList<>(); 758 sendLock.lockInterruptibly(); 759 try { 760 synchronized (indexOrderedCursorUpdates) { 761 MessageContext candidate = indexOrderedCursorUpdates.peek(); 762 while (candidate != null && candidate.message.getMessageId().getFutureOrSequenceLong() != null) { 763 candidate = indexOrderedCursorUpdates.removeFirst(); 764 // check for duplicate adds suppressed by the store 765 if (candidate.message.getMessageId().getFutureOrSequenceLong() instanceof Long && ((Long)candidate.message.getMessageId().getFutureOrSequenceLong()).compareTo(-1l) == 0) { 766 LOG.warn("{} messageStore indicated duplicate add attempt for {}, suppressing duplicate dispatch", this, candidate.message.getMessageId()); 767 } else { 768 orderedUpdates.add(candidate); 769 } 770 candidate = indexOrderedCursorUpdates.peek(); 771 } 772 } 773 messagesLock.writeLock().lock(); 774 try { 775 for (MessageContext messageContext : orderedUpdates) { 776 if (!messages.addMessageLast(messageContext.message)) { 777 // cursor suppressed a duplicate 778 messageContext.duplicate = true; 779 } 780 if (messageContext.onCompletion != null) { 781 messageContext.onCompletion.run(); 782 } 783 } 784 } finally { 785 messagesLock.writeLock().unlock(); 786 } 787 } finally { 788 sendLock.unlock(); 789 } 790 for (MessageContext messageContext : orderedUpdates) { 791 if (!messageContext.duplicate) { 792 messageSent(messageContext.context, messageContext.message); 793 } 794 } 795 orderedUpdates.clear(); 796 } 797 798 final class CursorAddSync extends Synchronization { 799 800 private final MessageContext messageContext; 801 802 CursorAddSync(MessageContext messageContext) { 803 this.messageContext = messageContext; 804 this.messageContext.message.incrementReferenceCount(); 805 } 806 807 @Override 808 public void afterCommit() throws Exception { 809 if (store != null && messageContext.message.isPersistent()) { 810 doPendingCursorAdditions(); 811 } else { 812 cursorAdd(messageContext.message); 813 messageSent(messageContext.context, messageContext.message); 814 } 815 messageContext.message.decrementReferenceCount(); 816 } 817 818 @Override 819 public void afterRollback() throws Exception { 820 messageContext.message.decrementReferenceCount(); 821 } 822 } 823 824 void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message) throws IOException, 825 Exception { 826 final ConnectionContext context = producerExchange.getConnectionContext(); 827 ListenableFuture<Object> result = null; 828 829 producerExchange.incrementSend(); 830 do { 831 checkUsage(context, producerExchange, message); 832 sendLock.lockInterruptibly(); 833 try { 834 message.getMessageId().setBrokerSequenceId(getDestinationSequenceId()); 835 if (store != null && message.isPersistent()) { 836 message.getMessageId().setFutureOrSequenceLong(null); 837 try { 838 //AMQ-6133 - don't store async if using persistJMSRedelivered 839 //This flag causes a sync update later on dispatch which can cause a race 840 //condition if the original add is processed after the update, which can cause 841 //a duplicate message to be stored 842 if (messages.isCacheEnabled() && !isPersistJMSRedelivered()) { 843 result = store.asyncAddQueueMessage(context, message, isOptimizeStorage()); 844 result.addListener(new PendingMarshalUsageTracker(message)); 845 } else { 846 store.addMessage(context, message); 847 } 848 if (isReduceMemoryFootprint()) { 849 message.clearMarshalledState(); 850 } 851 } catch (Exception e) { 852 // we may have a store in inconsistent state, so reset the cursor 853 // before restarting normal broker operations 854 resetNeeded = true; 855 throw e; 856 } 857 } 858 if(tryOrderedCursorAdd(message, context)) { 859 break; 860 } 861 } finally { 862 sendLock.unlock(); 863 } 864 } while (started.get()); 865 866 if (store == null || (!context.isInTransaction() && !message.isPersistent())) { 867 messageSent(context, message); 868 } 869 if (result != null && message.isResponseRequired() && !result.isCancelled()) { 870 try { 871 result.get(); 872 } catch (CancellationException e) { 873 // ignore - the task has been cancelled if the message 874 // has already been deleted 875 } 876 } 877 } 878 879 private boolean tryOrderedCursorAdd(Message message, ConnectionContext context) throws Exception { 880 boolean result = true; 881 882 if (context.isInTransaction()) { 883 context.getTransaction().addSynchronization(new CursorAddSync(new MessageContext(context, message, null))); 884 } else if (store != null && message.isPersistent()) { 885 doPendingCursorAdditions(); 886 } else { 887 // no ordering issue with non persistent messages 888 result = tryCursorAdd(message); 889 } 890 891 return result; 892 } 893 894 private void checkUsage(ConnectionContext context,ProducerBrokerExchange producerBrokerExchange, Message message) throws ResourceAllocationException, IOException, InterruptedException { 895 if (message.isPersistent()) { 896 if (store != null && systemUsage.getStoreUsage().isFull(getStoreUsageHighWaterMark())) { 897 final String logMessage = "Persistent store is Full, " + getStoreUsageHighWaterMark() + "% of " 898 + systemUsage.getStoreUsage().getLimit() + ". Stopping producer (" 899 + message.getProducerId() + ") to prevent flooding " 900 + getActiveMQDestination().getQualifiedName() + "." 901 + " See http://activemq.apache.org/producer-flow-control.html for more info"; 902 903 waitForSpace(context, producerBrokerExchange, systemUsage.getStoreUsage(), getStoreUsageHighWaterMark(), logMessage); 904 } 905 } else if (messages.getSystemUsage() != null && systemUsage.getTempUsage().isFull()) { 906 final String logMessage = "Temp Store is Full (" 907 + systemUsage.getTempUsage().getPercentUsage() + "% of " + systemUsage.getTempUsage().getLimit() 908 +"). Stopping producer (" + message.getProducerId() 909 + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." 910 + " See http://activemq.apache.org/producer-flow-control.html for more info"; 911 912 waitForSpace(context, producerBrokerExchange, messages.getSystemUsage().getTempUsage(), logMessage); 913 } 914 } 915 916 private void expireMessages() { 917 LOG.debug("{} expiring messages ..", getActiveMQDestination().getQualifiedName()); 918 919 // just track the insertion count 920 List<Message> browsedMessages = new InsertionCountList<Message>(); 921 doBrowse(browsedMessages, this.getMaxExpirePageSize()); 922 asyncWakeup(); 923 LOG.debug("{} expiring messages done.", getActiveMQDestination().getQualifiedName()); 924 } 925 926 @Override 927 public void gc() { 928 } 929 930 @Override 931 public void acknowledge(ConnectionContext context, Subscription sub, MessageAck ack, MessageReference node) 932 throws IOException { 933 messageConsumed(context, node); 934 if (store != null && node.isPersistent()) { 935 store.removeAsyncMessage(context, convertToNonRangedAck(ack, node)); 936 } 937 } 938 939 Message loadMessage(MessageId messageId) throws IOException { 940 Message msg = null; 941 if (store != null) { // can be null for a temp q 942 msg = store.getMessage(messageId); 943 if (msg != null) { 944 msg.setRegionDestination(this); 945 } 946 } 947 return msg; 948 } 949 950 public long getPendingMessageSize() { 951 messagesLock.readLock().lock(); 952 try{ 953 return messages.messageSize(); 954 } finally { 955 messagesLock.readLock().unlock(); 956 } 957 } 958 959 public long getPendingMessageCount() { 960 return this.destinationStatistics.getMessages().getCount(); 961 } 962 963 @Override 964 public String toString() { 965 return destination.getQualifiedName() + ", subscriptions=" + consumers.size() 966 + ", memory=" + memoryUsage.getPercentUsage() + "%, size=" + destinationStatistics.getMessages().getCount() + ", pending=" 967 + indexOrderedCursorUpdates.size(); 968 } 969 970 @Override 971 public void start() throws Exception { 972 if (started.compareAndSet(false, true)) { 973 if (memoryUsage != null) { 974 memoryUsage.start(); 975 } 976 if (systemUsage.getStoreUsage() != null) { 977 systemUsage.getStoreUsage().start(); 978 } 979 systemUsage.getMemoryUsage().addUsageListener(this); 980 messages.start(); 981 if (getExpireMessagesPeriod() > 0) { 982 scheduler.executePeriodically(expireMessagesTask, getExpireMessagesPeriod()); 983 } 984 doPageIn(false); 985 } 986 } 987 988 @Override 989 public void stop() throws Exception { 990 if (started.compareAndSet(true, false)) { 991 if (taskRunner != null) { 992 taskRunner.shutdown(); 993 } 994 if (this.executor != null) { 995 ThreadPoolUtils.shutdownNow(executor); 996 executor = null; 997 } 998 999 scheduler.cancel(expireMessagesTask); 1000 1001 if (flowControlTimeoutTask.isAlive()) { 1002 flowControlTimeoutTask.interrupt(); 1003 } 1004 1005 if (messages != null) { 1006 messages.stop(); 1007 } 1008 1009 for (MessageReference messageReference : pagedInMessages.values()) { 1010 messageReference.decrementReferenceCount(); 1011 } 1012 pagedInMessages.clear(); 1013 1014 systemUsage.getMemoryUsage().removeUsageListener(this); 1015 if (memoryUsage != null) { 1016 memoryUsage.stop(); 1017 } 1018 if (store != null) { 1019 store.stop(); 1020 } 1021 } 1022 } 1023 1024 // Properties 1025 // ------------------------------------------------------------------------- 1026 @Override 1027 public ActiveMQDestination getActiveMQDestination() { 1028 return destination; 1029 } 1030 1031 public MessageGroupMap getMessageGroupOwners() { 1032 if (messageGroupOwners == null) { 1033 messageGroupOwners = getMessageGroupMapFactory().createMessageGroupMap(); 1034 messageGroupOwners.setDestination(this); 1035 } 1036 return messageGroupOwners; 1037 } 1038 1039 public DispatchPolicy getDispatchPolicy() { 1040 return dispatchPolicy; 1041 } 1042 1043 public void setDispatchPolicy(DispatchPolicy dispatchPolicy) { 1044 this.dispatchPolicy = dispatchPolicy; 1045 } 1046 1047 public MessageGroupMapFactory getMessageGroupMapFactory() { 1048 return messageGroupMapFactory; 1049 } 1050 1051 public void setMessageGroupMapFactory(MessageGroupMapFactory messageGroupMapFactory) { 1052 this.messageGroupMapFactory = messageGroupMapFactory; 1053 } 1054 1055 public PendingMessageCursor getMessages() { 1056 return this.messages; 1057 } 1058 1059 public void setMessages(PendingMessageCursor messages) { 1060 this.messages = messages; 1061 } 1062 1063 public boolean isUseConsumerPriority() { 1064 return useConsumerPriority; 1065 } 1066 1067 public void setUseConsumerPriority(boolean useConsumerPriority) { 1068 this.useConsumerPriority = useConsumerPriority; 1069 } 1070 1071 public boolean isStrictOrderDispatch() { 1072 return strictOrderDispatch; 1073 } 1074 1075 public void setStrictOrderDispatch(boolean strictOrderDispatch) { 1076 this.strictOrderDispatch = strictOrderDispatch; 1077 } 1078 1079 public boolean isOptimizedDispatch() { 1080 return optimizedDispatch; 1081 } 1082 1083 public void setOptimizedDispatch(boolean optimizedDispatch) { 1084 this.optimizedDispatch = optimizedDispatch; 1085 } 1086 1087 public int getTimeBeforeDispatchStarts() { 1088 return timeBeforeDispatchStarts; 1089 } 1090 1091 public void setTimeBeforeDispatchStarts(int timeBeforeDispatchStarts) { 1092 this.timeBeforeDispatchStarts = timeBeforeDispatchStarts; 1093 } 1094 1095 public int getConsumersBeforeDispatchStarts() { 1096 return consumersBeforeDispatchStarts; 1097 } 1098 1099 public void setConsumersBeforeDispatchStarts(int consumersBeforeDispatchStarts) { 1100 this.consumersBeforeDispatchStarts = consumersBeforeDispatchStarts; 1101 } 1102 1103 public void setAllConsumersExclusiveByDefault(boolean allConsumersExclusiveByDefault) { 1104 this.allConsumersExclusiveByDefault = allConsumersExclusiveByDefault; 1105 } 1106 1107 public boolean isAllConsumersExclusiveByDefault() { 1108 return allConsumersExclusiveByDefault; 1109 } 1110 1111 public boolean isResetNeeded() { 1112 return resetNeeded; 1113 } 1114 1115 // Implementation methods 1116 // ------------------------------------------------------------------------- 1117 private QueueMessageReference createMessageReference(Message message) { 1118 QueueMessageReference result = new IndirectMessageReference(message); 1119 return result; 1120 } 1121 1122 @Override 1123 public Message[] browse() { 1124 List<Message> browseList = new ArrayList<Message>(); 1125 doBrowse(browseList, getMaxBrowsePageSize()); 1126 return browseList.toArray(new Message[browseList.size()]); 1127 } 1128 1129 public void doBrowse(List<Message> browseList, int max) { 1130 final ConnectionContext connectionContext = createConnectionContext(); 1131 try { 1132 int maxPageInAttempts = 1; 1133 if (max > 0) { 1134 messagesLock.readLock().lock(); 1135 try { 1136 maxPageInAttempts += (messages.size() / max); 1137 } finally { 1138 messagesLock.readLock().unlock(); 1139 } 1140 while (shouldPageInMoreForBrowse(max) && maxPageInAttempts-- > 0) { 1141 pageInMessages(!memoryUsage.isFull(110), max); 1142 } 1143 } 1144 doBrowseList(browseList, max, dispatchPendingList, pagedInPendingDispatchLock, connectionContext, "redeliveredWaitingDispatch+pagedInPendingDispatch"); 1145 doBrowseList(browseList, max, pagedInMessages, pagedInMessagesLock, connectionContext, "pagedInMessages"); 1146 1147 // we need a store iterator to walk messages on disk, independent of the cursor which is tracking 1148 // the next message batch 1149 } catch (BrokerStoppedException ignored) { 1150 } catch (Exception e) { 1151 LOG.error("Problem retrieving message for browse", e); 1152 } 1153 } 1154 1155 protected void doBrowseList(List<Message> browseList, int max, PendingList list, ReentrantReadWriteLock lock, ConnectionContext connectionContext, String name) throws Exception { 1156 List<MessageReference> toExpire = new ArrayList<MessageReference>(); 1157 lock.readLock().lock(); 1158 try { 1159 addAll(list.values(), browseList, max, toExpire); 1160 } finally { 1161 lock.readLock().unlock(); 1162 } 1163 for (MessageReference ref : toExpire) { 1164 if (broker.isExpired(ref)) { 1165 LOG.debug("expiring from {}: {}", name, ref); 1166 messageExpired(connectionContext, ref); 1167 } else { 1168 lock.writeLock().lock(); 1169 try { 1170 list.remove(ref); 1171 } finally { 1172 lock.writeLock().unlock(); 1173 } 1174 ref.decrementReferenceCount(); 1175 } 1176 } 1177 } 1178 1179 private boolean shouldPageInMoreForBrowse(int max) { 1180 int alreadyPagedIn = 0; 1181 pagedInMessagesLock.readLock().lock(); 1182 try { 1183 alreadyPagedIn = pagedInMessages.size(); 1184 } finally { 1185 pagedInMessagesLock.readLock().unlock(); 1186 } 1187 int messagesInQueue = alreadyPagedIn; 1188 messagesLock.readLock().lock(); 1189 try { 1190 messagesInQueue += messages.size(); 1191 } finally { 1192 messagesLock.readLock().unlock(); 1193 } 1194 1195 LOG.trace("max {}, alreadyPagedIn {}, messagesCount {}, memoryUsage {}%", new Object[]{max, alreadyPagedIn, messagesInQueue, memoryUsage.getPercentUsage()}); 1196 return (alreadyPagedIn < max) 1197 && (alreadyPagedIn < messagesInQueue) 1198 && messages.hasSpace(); 1199 } 1200 1201 private void addAll(Collection<? extends MessageReference> refs, List<Message> l, int max, 1202 List<MessageReference> toExpire) throws Exception { 1203 for (Iterator<? extends MessageReference> i = refs.iterator(); i.hasNext() && l.size() < max;) { 1204 QueueMessageReference ref = (QueueMessageReference) i.next(); 1205 if (ref.isExpired() && (ref.getLockOwner() == null)) { 1206 toExpire.add(ref); 1207 } else if (l.contains(ref.getMessage()) == false) { 1208 l.add(ref.getMessage()); 1209 } 1210 } 1211 } 1212 1213 public QueueMessageReference getMessage(String id) { 1214 MessageId msgId = new MessageId(id); 1215 pagedInMessagesLock.readLock().lock(); 1216 try { 1217 QueueMessageReference ref = (QueueMessageReference)this.pagedInMessages.get(msgId); 1218 if (ref != null) { 1219 return ref; 1220 } 1221 } finally { 1222 pagedInMessagesLock.readLock().unlock(); 1223 } 1224 messagesLock.writeLock().lock(); 1225 try{ 1226 try { 1227 messages.reset(); 1228 while (messages.hasNext()) { 1229 MessageReference mr = messages.next(); 1230 QueueMessageReference qmr = createMessageReference(mr.getMessage()); 1231 qmr.decrementReferenceCount(); 1232 messages.rollback(qmr.getMessageId()); 1233 if (msgId.equals(qmr.getMessageId())) { 1234 return qmr; 1235 } 1236 } 1237 } finally { 1238 messages.release(); 1239 } 1240 }finally { 1241 messagesLock.writeLock().unlock(); 1242 } 1243 return null; 1244 } 1245 1246 public void purge() throws Exception { 1247 ConnectionContext c = createConnectionContext(); 1248 List<MessageReference> list = null; 1249 long originalMessageCount = this.destinationStatistics.getMessages().getCount(); 1250 do { 1251 doPageIn(true, false, getMaxPageSize()); // signal no expiry processing needed. 1252 pagedInMessagesLock.readLock().lock(); 1253 try { 1254 list = new ArrayList<MessageReference>(pagedInMessages.values()); 1255 }finally { 1256 pagedInMessagesLock.readLock().unlock(); 1257 } 1258 1259 for (MessageReference ref : list) { 1260 try { 1261 QueueMessageReference r = (QueueMessageReference) ref; 1262 removeMessage(c, r); 1263 } catch (IOException e) { 1264 } 1265 } 1266 // don't spin/hang if stats are out and there is nothing left in the 1267 // store 1268 } while (!list.isEmpty() && this.destinationStatistics.getMessages().getCount() > 0); 1269 1270 if (this.destinationStatistics.getMessages().getCount() > 0) { 1271 LOG.warn("{} after purge of {} messages, message count stats report: {}", getActiveMQDestination().getQualifiedName(), originalMessageCount, this.destinationStatistics.getMessages().getCount()); 1272 } 1273 gc(); 1274 this.destinationStatistics.getMessages().setCount(0); 1275 getMessages().clear(); 1276 } 1277 1278 @Override 1279 public void clearPendingMessages() { 1280 messagesLock.writeLock().lock(); 1281 try { 1282 if (resetNeeded) { 1283 messages.gc(); 1284 messages.reset(); 1285 resetNeeded = false; 1286 } else { 1287 messages.rebase(); 1288 } 1289 asyncWakeup(); 1290 } finally { 1291 messagesLock.writeLock().unlock(); 1292 } 1293 } 1294 1295 /** 1296 * Removes the message matching the given messageId 1297 */ 1298 public boolean removeMessage(String messageId) throws Exception { 1299 return removeMatchingMessages(createMessageIdFilter(messageId), 1) > 0; 1300 } 1301 1302 /** 1303 * Removes the messages matching the given selector 1304 * 1305 * @return the number of messages removed 1306 */ 1307 public int removeMatchingMessages(String selector) throws Exception { 1308 return removeMatchingMessages(selector, -1); 1309 } 1310 1311 /** 1312 * Removes the messages matching the given selector up to the maximum number 1313 * of matched messages 1314 * 1315 * @return the number of messages removed 1316 */ 1317 public int removeMatchingMessages(String selector, int maximumMessages) throws Exception { 1318 return removeMatchingMessages(createSelectorFilter(selector), maximumMessages); 1319 } 1320 1321 /** 1322 * Removes the messages matching the given filter up to the maximum number 1323 * of matched messages 1324 * 1325 * @return the number of messages removed 1326 */ 1327 public int removeMatchingMessages(MessageReferenceFilter filter, int maximumMessages) throws Exception { 1328 int movedCounter = 0; 1329 Set<MessageReference> set = new LinkedHashSet<MessageReference>(); 1330 ConnectionContext context = createConnectionContext(); 1331 do { 1332 doPageIn(true); 1333 pagedInMessagesLock.readLock().lock(); 1334 try { 1335 set.addAll(pagedInMessages.values()); 1336 } finally { 1337 pagedInMessagesLock.readLock().unlock(); 1338 } 1339 List<MessageReference> list = new ArrayList<MessageReference>(set); 1340 for (MessageReference ref : list) { 1341 IndirectMessageReference r = (IndirectMessageReference) ref; 1342 if (filter.evaluate(context, r)) { 1343 1344 removeMessage(context, r); 1345 set.remove(r); 1346 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 1347 return movedCounter; 1348 } 1349 } 1350 } 1351 } while (set.size() < this.destinationStatistics.getMessages().getCount()); 1352 return movedCounter; 1353 } 1354 1355 /** 1356 * Copies the message matching the given messageId 1357 */ 1358 public boolean copyMessageTo(ConnectionContext context, String messageId, ActiveMQDestination dest) 1359 throws Exception { 1360 return copyMatchingMessages(context, createMessageIdFilter(messageId), dest, 1) > 0; 1361 } 1362 1363 /** 1364 * Copies the messages matching the given selector 1365 * 1366 * @return the number of messages copied 1367 */ 1368 public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest) 1369 throws Exception { 1370 return copyMatchingMessagesTo(context, selector, dest, -1); 1371 } 1372 1373 /** 1374 * Copies the messages matching the given selector up to the maximum number 1375 * of matched messages 1376 * 1377 * @return the number of messages copied 1378 */ 1379 public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest, 1380 int maximumMessages) throws Exception { 1381 return copyMatchingMessages(context, createSelectorFilter(selector), dest, maximumMessages); 1382 } 1383 1384 /** 1385 * Copies the messages matching the given filter up to the maximum number of 1386 * matched messages 1387 * 1388 * @return the number of messages copied 1389 */ 1390 public int copyMatchingMessages(ConnectionContext context, MessageReferenceFilter filter, ActiveMQDestination dest, 1391 int maximumMessages) throws Exception { 1392 int movedCounter = 0; 1393 int count = 0; 1394 Set<MessageReference> set = new LinkedHashSet<MessageReference>(); 1395 do { 1396 int oldMaxSize = getMaxPageSize(); 1397 setMaxPageSize((int) this.destinationStatistics.getMessages().getCount()); 1398 doPageIn(true); 1399 setMaxPageSize(oldMaxSize); 1400 pagedInMessagesLock.readLock().lock(); 1401 try { 1402 set.addAll(pagedInMessages.values()); 1403 } finally { 1404 pagedInMessagesLock.readLock().unlock(); 1405 } 1406 List<MessageReference> list = new ArrayList<MessageReference>(set); 1407 for (MessageReference ref : list) { 1408 IndirectMessageReference r = (IndirectMessageReference) ref; 1409 if (filter.evaluate(context, r)) { 1410 1411 r.incrementReferenceCount(); 1412 try { 1413 Message m = r.getMessage(); 1414 BrokerSupport.resend(context, m, dest); 1415 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 1416 return movedCounter; 1417 } 1418 } finally { 1419 r.decrementReferenceCount(); 1420 } 1421 } 1422 count++; 1423 } 1424 } while (count < this.destinationStatistics.getMessages().getCount()); 1425 return movedCounter; 1426 } 1427 1428 /** 1429 * Move a message 1430 * 1431 * @param context 1432 * connection context 1433 * @param m 1434 * QueueMessageReference 1435 * @param dest 1436 * ActiveMQDestination 1437 * @throws Exception 1438 */ 1439 public boolean moveMessageTo(ConnectionContext context, QueueMessageReference m, ActiveMQDestination dest) throws Exception { 1440 BrokerSupport.resend(context, m.getMessage(), dest); 1441 removeMessage(context, m); 1442 messagesLock.writeLock().lock(); 1443 try { 1444 messages.rollback(m.getMessageId()); 1445 if (isDLQ()) { 1446 DeadLetterStrategy stratagy = getDeadLetterStrategy(); 1447 stratagy.rollback(m.getMessage()); 1448 } 1449 } finally { 1450 messagesLock.writeLock().unlock(); 1451 } 1452 return true; 1453 } 1454 1455 /** 1456 * Moves the message matching the given messageId 1457 */ 1458 public boolean moveMessageTo(ConnectionContext context, String messageId, ActiveMQDestination dest) 1459 throws Exception { 1460 return moveMatchingMessagesTo(context, createMessageIdFilter(messageId), dest, 1) > 0; 1461 } 1462 1463 /** 1464 * Moves the messages matching the given selector 1465 * 1466 * @return the number of messages removed 1467 */ 1468 public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest) 1469 throws Exception { 1470 return moveMatchingMessagesTo(context, selector, dest, Integer.MAX_VALUE); 1471 } 1472 1473 /** 1474 * Moves the messages matching the given selector up to the maximum number 1475 * of matched messages 1476 */ 1477 public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest, 1478 int maximumMessages) throws Exception { 1479 return moveMatchingMessagesTo(context, createSelectorFilter(selector), dest, maximumMessages); 1480 } 1481 1482 /** 1483 * Moves the messages matching the given filter up to the maximum number of 1484 * matched messages 1485 */ 1486 public int moveMatchingMessagesTo(ConnectionContext context, MessageReferenceFilter filter, 1487 ActiveMQDestination dest, int maximumMessages) throws Exception { 1488 int movedCounter = 0; 1489 Set<MessageReference> set = new LinkedHashSet<MessageReference>(); 1490 do { 1491 doPageIn(true); 1492 pagedInMessagesLock.readLock().lock(); 1493 try { 1494 set.addAll(pagedInMessages.values()); 1495 } finally { 1496 pagedInMessagesLock.readLock().unlock(); 1497 } 1498 List<MessageReference> list = new ArrayList<MessageReference>(set); 1499 for (MessageReference ref : list) { 1500 if (filter.evaluate(context, ref)) { 1501 // We should only move messages that can be locked. 1502 moveMessageTo(context, (QueueMessageReference)ref, dest); 1503 set.remove(ref); 1504 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 1505 return movedCounter; 1506 } 1507 } 1508 } 1509 } while (set.size() < this.destinationStatistics.getMessages().getCount() && set.size() < maximumMessages); 1510 return movedCounter; 1511 } 1512 1513 public int retryMessages(ConnectionContext context, int maximumMessages) throws Exception { 1514 if (!isDLQ()) { 1515 throw new Exception("Retry of message is only possible on Dead Letter Queues!"); 1516 } 1517 int restoredCounter = 0; 1518 Set<MessageReference> set = new LinkedHashSet<MessageReference>(); 1519 do { 1520 doPageIn(true); 1521 pagedInMessagesLock.readLock().lock(); 1522 try { 1523 set.addAll(pagedInMessages.values()); 1524 } finally { 1525 pagedInMessagesLock.readLock().unlock(); 1526 } 1527 List<MessageReference> list = new ArrayList<MessageReference>(set); 1528 for (MessageReference ref : list) { 1529 if (ref.getMessage().getOriginalDestination() != null) { 1530 1531 moveMessageTo(context, (QueueMessageReference)ref, ref.getMessage().getOriginalDestination()); 1532 set.remove(ref); 1533 if (++restoredCounter >= maximumMessages && maximumMessages > 0) { 1534 return restoredCounter; 1535 } 1536 } 1537 } 1538 } while (set.size() < this.destinationStatistics.getMessages().getCount() && set.size() < maximumMessages); 1539 return restoredCounter; 1540 } 1541 1542 /** 1543 * @return true if we would like to iterate again 1544 * @see org.apache.activemq.thread.Task#iterate() 1545 */ 1546 @Override 1547 public boolean iterate() { 1548 MDC.put("activemq.destination", getName()); 1549 boolean pageInMoreMessages = false; 1550 synchronized (iteratingMutex) { 1551 1552 // If optimize dispatch is on or this is a slave this method could be called recursively 1553 // we set this state value to short-circuit wakeup in those cases to avoid that as it 1554 // could lead to errors. 1555 iterationRunning = true; 1556 1557 // do early to allow dispatch of these waiting messages 1558 synchronized (messagesWaitingForSpace) { 1559 Iterator<Runnable> it = messagesWaitingForSpace.values().iterator(); 1560 while (it.hasNext()) { 1561 if (!memoryUsage.isFull()) { 1562 Runnable op = it.next(); 1563 it.remove(); 1564 op.run(); 1565 } else { 1566 registerCallbackForNotFullNotification(); 1567 break; 1568 } 1569 } 1570 } 1571 1572 if (firstConsumer) { 1573 firstConsumer = false; 1574 try { 1575 if (consumersBeforeDispatchStarts > 0) { 1576 int timeout = 1000; // wait one second by default if 1577 // consumer count isn't reached 1578 if (timeBeforeDispatchStarts > 0) { 1579 timeout = timeBeforeDispatchStarts; 1580 } 1581 if (consumersBeforeStartsLatch.await(timeout, TimeUnit.MILLISECONDS)) { 1582 LOG.debug("{} consumers subscribed. Starting dispatch.", consumers.size()); 1583 } else { 1584 LOG.debug("{} ms elapsed and {} consumers subscribed. Starting dispatch.", timeout, consumers.size()); 1585 } 1586 } 1587 if (timeBeforeDispatchStarts > 0 && consumersBeforeDispatchStarts <= 0) { 1588 iteratingMutex.wait(timeBeforeDispatchStarts); 1589 LOG.debug("{} ms elapsed. Starting dispatch.", timeBeforeDispatchStarts); 1590 } 1591 } catch (Exception e) { 1592 LOG.error(e.toString()); 1593 } 1594 } 1595 1596 messagesLock.readLock().lock(); 1597 try{ 1598 pageInMoreMessages |= !messages.isEmpty(); 1599 } finally { 1600 messagesLock.readLock().unlock(); 1601 } 1602 1603 pagedInPendingDispatchLock.readLock().lock(); 1604 try { 1605 pageInMoreMessages |= !dispatchPendingList.isEmpty(); 1606 } finally { 1607 pagedInPendingDispatchLock.readLock().unlock(); 1608 } 1609 1610 // Perhaps we should page always into the pagedInPendingDispatch 1611 // list if 1612 // !messages.isEmpty(), and then if 1613 // !pagedInPendingDispatch.isEmpty() 1614 // then we do a dispatch. 1615 boolean hasBrowsers = browserDispatches.size() > 0; 1616 1617 if (pageInMoreMessages || hasBrowsers || !dispatchPendingList.hasRedeliveries()) { 1618 try { 1619 pageInMessages(hasBrowsers && getMaxBrowsePageSize() > 0, getMaxPageSize()); 1620 } catch (Throwable e) { 1621 LOG.error("Failed to page in more queue messages ", e); 1622 } 1623 } 1624 1625 if (hasBrowsers) { 1626 ArrayList<MessageReference> alreadyDispatchedMessages = null; 1627 pagedInMessagesLock.readLock().lock(); 1628 try{ 1629 alreadyDispatchedMessages = new ArrayList<MessageReference>(pagedInMessages.values()); 1630 }finally { 1631 pagedInMessagesLock.readLock().unlock(); 1632 } 1633 1634 Iterator<BrowserDispatch> browsers = browserDispatches.iterator(); 1635 while (browsers.hasNext()) { 1636 BrowserDispatch browserDispatch = browsers.next(); 1637 try { 1638 MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext(); 1639 msgContext.setDestination(destination); 1640 1641 QueueBrowserSubscription browser = browserDispatch.getBrowser(); 1642 1643 LOG.debug("dispatch to browser: {}, already dispatched/paged count: {}", browser, alreadyDispatchedMessages.size()); 1644 boolean added = false; 1645 for (MessageReference node : alreadyDispatchedMessages) { 1646 if (!((QueueMessageReference)node).isAcked() && !browser.isDuplicate(node.getMessageId()) && !browser.atMax()) { 1647 msgContext.setMessageReference(node); 1648 if (browser.matches(node, msgContext)) { 1649 browser.add(node); 1650 added = true; 1651 } 1652 } 1653 } 1654 // are we done browsing? no new messages paged 1655 if (!added || browser.atMax()) { 1656 browser.decrementQueueRef(); 1657 browserDispatches.remove(browserDispatch); 1658 } 1659 } catch (Exception e) { 1660 LOG.warn("exception on dispatch to browser: {}", browserDispatch.getBrowser(), e); 1661 } 1662 } 1663 } 1664 1665 if (pendingWakeups.get() > 0) { 1666 pendingWakeups.decrementAndGet(); 1667 } 1668 MDC.remove("activemq.destination"); 1669 iterationRunning = false; 1670 1671 return pendingWakeups.get() > 0; 1672 } 1673 } 1674 1675 public void pauseDispatch() { 1676 dispatchSelector.pause(); 1677 } 1678 1679 public void resumeDispatch() { 1680 dispatchSelector.resume(); 1681 wakeup(); 1682 } 1683 1684 public boolean isDispatchPaused() { 1685 return dispatchSelector.isPaused(); 1686 } 1687 1688 protected MessageReferenceFilter createMessageIdFilter(final String messageId) { 1689 return new MessageReferenceFilter() { 1690 @Override 1691 public boolean evaluate(ConnectionContext context, MessageReference r) { 1692 return messageId.equals(r.getMessageId().toString()); 1693 } 1694 1695 @Override 1696 public String toString() { 1697 return "MessageIdFilter: " + messageId; 1698 } 1699 }; 1700 } 1701 1702 protected MessageReferenceFilter createSelectorFilter(String selector) throws InvalidSelectorException { 1703 1704 if (selector == null || selector.isEmpty()) { 1705 return new MessageReferenceFilter() { 1706 1707 @Override 1708 public boolean evaluate(ConnectionContext context, MessageReference messageReference) throws JMSException { 1709 return true; 1710 } 1711 }; 1712 } 1713 1714 final BooleanExpression selectorExpression = SelectorParser.parse(selector); 1715 1716 return new MessageReferenceFilter() { 1717 @Override 1718 public boolean evaluate(ConnectionContext context, MessageReference r) throws JMSException { 1719 MessageEvaluationContext messageEvaluationContext = context.getMessageEvaluationContext(); 1720 1721 messageEvaluationContext.setMessageReference(r); 1722 if (messageEvaluationContext.getDestination() == null) { 1723 messageEvaluationContext.setDestination(getActiveMQDestination()); 1724 } 1725 1726 return selectorExpression.matches(messageEvaluationContext); 1727 } 1728 }; 1729 } 1730 1731 protected void removeMessage(ConnectionContext c, QueueMessageReference r) throws IOException { 1732 removeMessage(c, null, r); 1733 pagedInPendingDispatchLock.writeLock().lock(); 1734 try { 1735 dispatchPendingList.remove(r); 1736 } finally { 1737 pagedInPendingDispatchLock.writeLock().unlock(); 1738 } 1739 } 1740 1741 protected void removeMessage(ConnectionContext c, Subscription subs, QueueMessageReference r) throws IOException { 1742 MessageAck ack = new MessageAck(); 1743 ack.setAckType(MessageAck.STANDARD_ACK_TYPE); 1744 ack.setDestination(destination); 1745 ack.setMessageID(r.getMessageId()); 1746 removeMessage(c, subs, r, ack); 1747 } 1748 1749 protected void removeMessage(ConnectionContext context, Subscription sub, final QueueMessageReference reference, 1750 MessageAck ack) throws IOException { 1751 LOG.trace("ack of {} with {}", reference.getMessageId(), ack); 1752 // This sends the ack the the journal.. 1753 if (!ack.isInTransaction()) { 1754 acknowledge(context, sub, ack, reference); 1755 getDestinationStatistics().getDequeues().increment(); 1756 dropMessage(reference); 1757 } else { 1758 try { 1759 acknowledge(context, sub, ack, reference); 1760 } finally { 1761 context.getTransaction().addSynchronization(new Synchronization() { 1762 1763 @Override 1764 public void afterCommit() throws Exception { 1765 getDestinationStatistics().getDequeues().increment(); 1766 dropMessage(reference); 1767 wakeup(); 1768 } 1769 1770 @Override 1771 public void afterRollback() throws Exception { 1772 reference.setAcked(false); 1773 wakeup(); 1774 } 1775 }); 1776 } 1777 } 1778 if (ack.isPoisonAck() || (sub != null && sub.getConsumerInfo().isNetworkSubscription())) { 1779 // message gone to DLQ, is ok to allow redelivery 1780 messagesLock.writeLock().lock(); 1781 try { 1782 messages.rollback(reference.getMessageId()); 1783 } finally { 1784 messagesLock.writeLock().unlock(); 1785 } 1786 if (sub != null && sub.getConsumerInfo().isNetworkSubscription()) { 1787 getDestinationStatistics().getForwards().increment(); 1788 } 1789 } 1790 // after successful store update 1791 reference.setAcked(true); 1792 } 1793 1794 private void dropMessage(QueueMessageReference reference) { 1795 if (!reference.isDropped()) { 1796 reference.drop(); 1797 destinationStatistics.getMessages().decrement(); 1798 pagedInMessagesLock.writeLock().lock(); 1799 try { 1800 pagedInMessages.remove(reference); 1801 } finally { 1802 pagedInMessagesLock.writeLock().unlock(); 1803 } 1804 } 1805 } 1806 1807 public void messageExpired(ConnectionContext context, MessageReference reference) { 1808 messageExpired(context, null, reference); 1809 } 1810 1811 @Override 1812 public void messageExpired(ConnectionContext context, Subscription subs, MessageReference reference) { 1813 LOG.debug("message expired: {}", reference); 1814 broker.messageExpired(context, reference, subs); 1815 destinationStatistics.getExpired().increment(); 1816 try { 1817 removeMessage(context, subs, (QueueMessageReference) reference); 1818 messagesLock.writeLock().lock(); 1819 try { 1820 messages.rollback(reference.getMessageId()); 1821 } finally { 1822 messagesLock.writeLock().unlock(); 1823 } 1824 } catch (IOException e) { 1825 LOG.error("Failed to remove expired Message from the store ", e); 1826 } 1827 } 1828 1829 private final boolean cursorAdd(final Message msg) throws Exception { 1830 messagesLock.writeLock().lock(); 1831 try { 1832 return messages.addMessageLast(msg); 1833 } finally { 1834 messagesLock.writeLock().unlock(); 1835 } 1836 } 1837 1838 private final boolean tryCursorAdd(final Message msg) throws Exception { 1839 messagesLock.writeLock().lock(); 1840 try { 1841 return messages.tryAddMessageLast(msg, 50); 1842 } finally { 1843 messagesLock.writeLock().unlock(); 1844 } 1845 } 1846 1847 final void messageSent(final ConnectionContext context, final Message msg) throws Exception { 1848 destinationStatistics.getEnqueues().increment(); 1849 destinationStatistics.getMessages().increment(); 1850 destinationStatistics.getMessageSize().addSize(msg.getSize()); 1851 messageDelivered(context, msg); 1852 consumersLock.readLock().lock(); 1853 try { 1854 if (consumers.isEmpty()) { 1855 onMessageWithNoConsumers(context, msg); 1856 } 1857 }finally { 1858 consumersLock.readLock().unlock(); 1859 } 1860 LOG.debug("{} Message {} sent to {}", new Object[]{ broker.getBrokerName(), msg.getMessageId(), this.destination }); 1861 wakeup(); 1862 } 1863 1864 @Override 1865 public void wakeup() { 1866 if (optimizedDispatch && !iterationRunning) { 1867 iterate(); 1868 pendingWakeups.incrementAndGet(); 1869 } else { 1870 asyncWakeup(); 1871 } 1872 } 1873 1874 private void asyncWakeup() { 1875 try { 1876 pendingWakeups.incrementAndGet(); 1877 this.taskRunner.wakeup(); 1878 } catch (InterruptedException e) { 1879 LOG.warn("Async task runner failed to wakeup ", e); 1880 } 1881 } 1882 1883 private void doPageIn(boolean force) throws Exception { 1884 doPageIn(force, true, getMaxPageSize()); 1885 } 1886 1887 private void doPageIn(boolean force, boolean processExpired, int maxPageSize) throws Exception { 1888 PendingList newlyPaged = doPageInForDispatch(force, processExpired, maxPageSize); 1889 pagedInPendingDispatchLock.writeLock().lock(); 1890 try { 1891 if (dispatchPendingList.isEmpty()) { 1892 dispatchPendingList.addAll(newlyPaged); 1893 1894 } else { 1895 for (MessageReference qmr : newlyPaged) { 1896 if (!dispatchPendingList.contains(qmr)) { 1897 dispatchPendingList.addMessageLast(qmr); 1898 } 1899 } 1900 } 1901 } finally { 1902 pagedInPendingDispatchLock.writeLock().unlock(); 1903 } 1904 } 1905 1906 private PendingList doPageInForDispatch(boolean force, boolean processExpired, int maxPageSize) throws Exception { 1907 List<QueueMessageReference> result = null; 1908 PendingList resultList = null; 1909 1910 int toPageIn = Math.min(maxPageSize, messages.size()); 1911 int pagedInPendingSize = 0; 1912 pagedInPendingDispatchLock.readLock().lock(); 1913 try { 1914 pagedInPendingSize = dispatchPendingList.size(); 1915 } finally { 1916 pagedInPendingDispatchLock.readLock().unlock(); 1917 } 1918 if (isLazyDispatch() && !force) { 1919 // Only page in the minimum number of messages which can be 1920 // dispatched immediately. 1921 toPageIn = Math.min(getConsumerMessageCountBeforeFull(), toPageIn); 1922 } 1923 1924 if (LOG.isDebugEnabled()) { 1925 LOG.debug("{} toPageIn: {}, force:{}, Inflight: {}, pagedInMessages.size {}, pagedInPendingDispatch.size {}, enqueueCount: {}, dequeueCount: {}, memUsage:{}, maxPageSize:{}", 1926 new Object[]{ 1927 this, 1928 toPageIn, 1929 force, 1930 destinationStatistics.getInflight().getCount(), 1931 pagedInMessages.size(), 1932 pagedInPendingSize, 1933 destinationStatistics.getEnqueues().getCount(), 1934 destinationStatistics.getDequeues().getCount(), 1935 getMemoryUsage().getUsage(), 1936 maxPageSize 1937 }); 1938 } 1939 1940 if (toPageIn > 0 && (force || (haveRealConsumer() && pagedInPendingSize < maxPageSize))) { 1941 int count = 0; 1942 result = new ArrayList<QueueMessageReference>(toPageIn); 1943 messagesLock.writeLock().lock(); 1944 try { 1945 try { 1946 messages.setMaxBatchSize(toPageIn); 1947 messages.reset(); 1948 while (count < toPageIn && messages.hasNext()) { 1949 MessageReference node = messages.next(); 1950 messages.remove(); 1951 1952 QueueMessageReference ref = createMessageReference(node.getMessage()); 1953 if (processExpired && ref.isExpired()) { 1954 if (broker.isExpired(ref)) { 1955 messageExpired(createConnectionContext(), ref); 1956 } else { 1957 ref.decrementReferenceCount(); 1958 } 1959 } else { 1960 result.add(ref); 1961 count++; 1962 } 1963 } 1964 } finally { 1965 messages.release(); 1966 } 1967 } finally { 1968 messagesLock.writeLock().unlock(); 1969 } 1970 // Only add new messages, not already pagedIn to avoid multiple 1971 // dispatch attempts 1972 pagedInMessagesLock.writeLock().lock(); 1973 try { 1974 if(isPrioritizedMessages()) { 1975 resultList = new PrioritizedPendingList(); 1976 } else { 1977 resultList = new OrderedPendingList(); 1978 } 1979 for (QueueMessageReference ref : result) { 1980 if (!pagedInMessages.contains(ref)) { 1981 pagedInMessages.addMessageLast(ref); 1982 resultList.addMessageLast(ref); 1983 } else { 1984 ref.decrementReferenceCount(); 1985 // store should have trapped duplicate in it's index, also cursor audit 1986 // we need to remove the duplicate from the store in the knowledge that the original message may be inflight 1987 // note: jdbc store will not trap unacked messages as a duplicate b/c it gives each message a unique sequence id 1988 LOG.warn("{}, duplicate message {} paged in, is cursor audit disabled? Removing from store and redirecting to dlq", this, ref.getMessage()); 1989 if (store != null) { 1990 ConnectionContext connectionContext = createConnectionContext(); 1991 store.removeMessage(connectionContext, new MessageAck(ref.getMessage(), MessageAck.POSION_ACK_TYPE, 1)); 1992 broker.getRoot().sendToDeadLetterQueue(connectionContext, ref.getMessage(), null, new Throwable("duplicate paged in from store for " + destination)); 1993 } 1994 } 1995 } 1996 } finally { 1997 pagedInMessagesLock.writeLock().unlock(); 1998 } 1999 } else { 2000 // Avoid return null list, if condition is not validated 2001 resultList = new OrderedPendingList(); 2002 } 2003 2004 return resultList; 2005 } 2006 2007 private final boolean haveRealConsumer() { 2008 return consumers.size() - browserDispatches.size() > 0; 2009 } 2010 2011 private void doDispatch(PendingList list) throws Exception { 2012 boolean doWakeUp = false; 2013 2014 pagedInPendingDispatchLock.writeLock().lock(); 2015 try { 2016 if (isPrioritizedMessages() && !dispatchPendingList.isEmpty() && list != null && !list.isEmpty()) { 2017 // merge all to select priority order 2018 for (MessageReference qmr : list) { 2019 if (!dispatchPendingList.contains(qmr)) { 2020 dispatchPendingList.addMessageLast(qmr); 2021 } 2022 } 2023 list = null; 2024 } 2025 2026 doActualDispatch(dispatchPendingList); 2027 // and now see if we can dispatch the new stuff.. and append to the pending 2028 // list anything that does not actually get dispatched. 2029 if (list != null && !list.isEmpty()) { 2030 if (dispatchPendingList.isEmpty()) { 2031 dispatchPendingList.addAll(doActualDispatch(list)); 2032 } else { 2033 for (MessageReference qmr : list) { 2034 if (!dispatchPendingList.contains(qmr)) { 2035 dispatchPendingList.addMessageLast(qmr); 2036 } 2037 } 2038 doWakeUp = true; 2039 } 2040 } 2041 } finally { 2042 pagedInPendingDispatchLock.writeLock().unlock(); 2043 } 2044 2045 if (doWakeUp) { 2046 // avoid lock order contention 2047 asyncWakeup(); 2048 } 2049 } 2050 2051 /** 2052 * @return list of messages that could get dispatched to consumers if they 2053 * were not full. 2054 */ 2055 private PendingList doActualDispatch(PendingList list) throws Exception { 2056 List<Subscription> consumers; 2057 consumersLock.readLock().lock(); 2058 2059 try { 2060 if (this.consumers.isEmpty()) { 2061 // slave dispatch happens in processDispatchNotification 2062 return list; 2063 } 2064 consumers = new ArrayList<Subscription>(this.consumers); 2065 } finally { 2066 consumersLock.readLock().unlock(); 2067 } 2068 2069 Set<Subscription> fullConsumers = new HashSet<Subscription>(this.consumers.size()); 2070 2071 for (Iterator<MessageReference> iterator = list.iterator(); iterator.hasNext();) { 2072 2073 MessageReference node = iterator.next(); 2074 Subscription target = null; 2075 for (Subscription s : consumers) { 2076 if (s instanceof QueueBrowserSubscription) { 2077 continue; 2078 } 2079 if (!fullConsumers.contains(s)) { 2080 if (!s.isFull()) { 2081 if (dispatchSelector.canSelect(s, node) && assignMessageGroup(s, (QueueMessageReference)node) && !((QueueMessageReference) node).isAcked() ) { 2082 // Dispatch it. 2083 s.add(node); 2084 LOG.trace("assigned {} to consumer {}", node.getMessageId(), s.getConsumerInfo().getConsumerId()); 2085 iterator.remove(); 2086 target = s; 2087 break; 2088 } 2089 } else { 2090 // no further dispatch of list to a full consumer to 2091 // avoid out of order message receipt 2092 fullConsumers.add(s); 2093 LOG.trace("Subscription full {}", s); 2094 } 2095 } 2096 } 2097 2098 if (target == null && node.isDropped()) { 2099 iterator.remove(); 2100 } 2101 2102 // return if there are no consumers or all consumers are full 2103 if (target == null && consumers.size() == fullConsumers.size()) { 2104 return list; 2105 } 2106 2107 // If it got dispatched, rotate the consumer list to get round robin 2108 // distribution. 2109 if (target != null && !strictOrderDispatch && consumers.size() > 1 2110 && !dispatchSelector.isExclusiveConsumer(target)) { 2111 consumersLock.writeLock().lock(); 2112 try { 2113 if (removeFromConsumerList(target)) { 2114 addToConsumerList(target); 2115 consumers = new ArrayList<Subscription>(this.consumers); 2116 } 2117 } finally { 2118 consumersLock.writeLock().unlock(); 2119 } 2120 } 2121 } 2122 2123 return list; 2124 } 2125 2126 protected boolean assignMessageGroup(Subscription subscription, QueueMessageReference node) throws Exception { 2127 boolean result = true; 2128 // Keep message groups together. 2129 String groupId = node.getGroupID(); 2130 int sequence = node.getGroupSequence(); 2131 if (groupId != null) { 2132 2133 MessageGroupMap messageGroupOwners = getMessageGroupOwners(); 2134 // If we can own the first, then no-one else should own the 2135 // rest. 2136 if (sequence == 1) { 2137 assignGroup(subscription, messageGroupOwners, node, groupId); 2138 } else { 2139 2140 // Make sure that the previous owner is still valid, we may 2141 // need to become the new owner. 2142 ConsumerId groupOwner; 2143 2144 groupOwner = messageGroupOwners.get(groupId); 2145 if (groupOwner == null) { 2146 assignGroup(subscription, messageGroupOwners, node, groupId); 2147 } else { 2148 if (groupOwner.equals(subscription.getConsumerInfo().getConsumerId())) { 2149 // A group sequence < 1 is an end of group signal. 2150 if (sequence < 0) { 2151 messageGroupOwners.removeGroup(groupId); 2152 subscription.getConsumerInfo().decrementAssignedGroupCount(destination); 2153 } 2154 } else { 2155 result = false; 2156 } 2157 } 2158 } 2159 } 2160 2161 return result; 2162 } 2163 2164 protected void assignGroup(Subscription subs, MessageGroupMap messageGroupOwners, MessageReference n, String groupId) throws IOException { 2165 messageGroupOwners.put(groupId, subs.getConsumerInfo().getConsumerId()); 2166 Message message = n.getMessage(); 2167 message.setJMSXGroupFirstForConsumer(true); 2168 subs.getConsumerInfo().incrementAssignedGroupCount(destination); 2169 } 2170 2171 protected void pageInMessages(boolean force, int maxPageSize) throws Exception { 2172 doDispatch(doPageInForDispatch(force, true, maxPageSize)); 2173 } 2174 2175 private void addToConsumerList(Subscription sub) { 2176 if (useConsumerPriority) { 2177 consumers.add(sub); 2178 Collections.sort(consumers, orderedCompare); 2179 } else { 2180 consumers.add(sub); 2181 } 2182 } 2183 2184 private boolean removeFromConsumerList(Subscription sub) { 2185 return consumers.remove(sub); 2186 } 2187 2188 private int getConsumerMessageCountBeforeFull() throws Exception { 2189 int total = 0; 2190 consumersLock.readLock().lock(); 2191 try { 2192 for (Subscription s : consumers) { 2193 if (s.isBrowser()) { 2194 continue; 2195 } 2196 int countBeforeFull = s.countBeforeFull(); 2197 total += countBeforeFull; 2198 } 2199 } finally { 2200 consumersLock.readLock().unlock(); 2201 } 2202 return total; 2203 } 2204 2205 /* 2206 * In slave mode, dispatch is ignored till we get this notification as the 2207 * dispatch process is non deterministic between master and slave. On a 2208 * notification, the actual dispatch to the subscription (as chosen by the 2209 * master) is completed. (non-Javadoc) 2210 * @see 2211 * org.apache.activemq.broker.region.BaseDestination#processDispatchNotification 2212 * (org.apache.activemq.command.MessageDispatchNotification) 2213 */ 2214 @Override 2215 public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception { 2216 // do dispatch 2217 Subscription sub = getMatchingSubscription(messageDispatchNotification); 2218 if (sub != null) { 2219 MessageReference message = getMatchingMessage(messageDispatchNotification); 2220 sub.add(message); 2221 sub.processMessageDispatchNotification(messageDispatchNotification); 2222 } 2223 } 2224 2225 private QueueMessageReference getMatchingMessage(MessageDispatchNotification messageDispatchNotification) 2226 throws Exception { 2227 QueueMessageReference message = null; 2228 MessageId messageId = messageDispatchNotification.getMessageId(); 2229 2230 pagedInPendingDispatchLock.writeLock().lock(); 2231 try { 2232 for (MessageReference ref : dispatchPendingList) { 2233 if (messageId.equals(ref.getMessageId())) { 2234 message = (QueueMessageReference)ref; 2235 dispatchPendingList.remove(ref); 2236 break; 2237 } 2238 } 2239 } finally { 2240 pagedInPendingDispatchLock.writeLock().unlock(); 2241 } 2242 2243 if (message == null) { 2244 pagedInMessagesLock.readLock().lock(); 2245 try { 2246 message = (QueueMessageReference)pagedInMessages.get(messageId); 2247 } finally { 2248 pagedInMessagesLock.readLock().unlock(); 2249 } 2250 } 2251 2252 if (message == null) { 2253 messagesLock.writeLock().lock(); 2254 try { 2255 try { 2256 messages.setMaxBatchSize(getMaxPageSize()); 2257 messages.reset(); 2258 while (messages.hasNext()) { 2259 MessageReference node = messages.next(); 2260 messages.remove(); 2261 if (messageId.equals(node.getMessageId())) { 2262 message = this.createMessageReference(node.getMessage()); 2263 break; 2264 } 2265 } 2266 } finally { 2267 messages.release(); 2268 } 2269 } finally { 2270 messagesLock.writeLock().unlock(); 2271 } 2272 } 2273 2274 if (message == null) { 2275 Message msg = loadMessage(messageId); 2276 if (msg != null) { 2277 message = this.createMessageReference(msg); 2278 } 2279 } 2280 2281 if (message == null) { 2282 throw new JMSException("Slave broker out of sync with master - Message: " 2283 + messageDispatchNotification.getMessageId() + " on " 2284 + messageDispatchNotification.getDestination() + " does not exist among pending(" 2285 + dispatchPendingList.size() + ") for subscription: " 2286 + messageDispatchNotification.getConsumerId()); 2287 } 2288 return message; 2289 } 2290 2291 /** 2292 * Find a consumer that matches the id in the message dispatch notification 2293 * 2294 * @param messageDispatchNotification 2295 * @return sub or null if the subscription has been removed before dispatch 2296 * @throws JMSException 2297 */ 2298 private Subscription getMatchingSubscription(MessageDispatchNotification messageDispatchNotification) 2299 throws JMSException { 2300 Subscription sub = null; 2301 consumersLock.readLock().lock(); 2302 try { 2303 for (Subscription s : consumers) { 2304 if (messageDispatchNotification.getConsumerId().equals(s.getConsumerInfo().getConsumerId())) { 2305 sub = s; 2306 break; 2307 } 2308 } 2309 } finally { 2310 consumersLock.readLock().unlock(); 2311 } 2312 return sub; 2313 } 2314 2315 @Override 2316 public void onUsageChanged(@SuppressWarnings("rawtypes") Usage usage, int oldPercentUsage, int newPercentUsage) { 2317 if (oldPercentUsage > newPercentUsage) { 2318 asyncWakeup(); 2319 } 2320 } 2321 2322 @Override 2323 protected Logger getLog() { 2324 return LOG; 2325 } 2326 2327 protected boolean isOptimizeStorage(){ 2328 boolean result = false; 2329 if (isDoOptimzeMessageStorage()){ 2330 consumersLock.readLock().lock(); 2331 try{ 2332 if (consumers.isEmpty()==false){ 2333 result = true; 2334 for (Subscription s : consumers) { 2335 if (s.getPrefetchSize()==0){ 2336 result = false; 2337 break; 2338 } 2339 if (s.isSlowConsumer()){ 2340 result = false; 2341 break; 2342 } 2343 if (s.getInFlightUsage() > getOptimizeMessageStoreInFlightLimit()){ 2344 result = false; 2345 break; 2346 } 2347 } 2348 } 2349 } finally { 2350 consumersLock.readLock().unlock(); 2351 } 2352 } 2353 return result; 2354 } 2355}