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