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