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