You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2011/05/20 17:16:51 UTC
svn commit: r1125427 - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/broker/region/Queue.java
test/java/org/apache/activemq/usecases/LargeQueueSparseDeleteTest.java
Author: tabish
Date: Fri May 20 15:16:51 2011
New Revision: 1125427
URL: http://svn.apache.org/viewvc?rev=1125427&view=rev
Log:
Fix for: https://issues.apache.org/jira/browse/AMQ-3312
Replace usage of CopyOnWriteArraySet in the MBean methods that don't require a thread safe collection.
Added:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/LargeQueueSparseDeleteTest.java (with props)
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java?rev=1125427&r1=1125426&r2=1125427&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java Fri May 20 15:16:51 2011
@@ -30,7 +30,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CancellationException;
-import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
@@ -78,8 +77,8 @@ import org.slf4j.MDC;
/**
* The Queue is a List of MessageEntry objects that are dispatched to matching
* subscriptions.
- *
- *
+ *
+ *
*/
public class Queue extends BaseDestination implements Task, UsageListener {
protected static final Logger LOG = LoggerFactory.getLogger(Queue.class);
@@ -113,7 +112,7 @@ public class Queue extends BaseDestinati
private CountDownLatch consumersBeforeStartsLatch;
private final AtomicLong pendingWakeups = new AtomicLong();
private boolean allConsumersExclusiveByDefault = false;
-
+
private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() {
public void run() {
asyncWakeup();
@@ -369,7 +368,7 @@ public class Queue extends BaseDestinati
try {
sub.add(context, this);
-
+
// needs to be synchronized - so no contention with dispatching
// consumersLock.
consumersLock.writeLock().lock();
@@ -461,7 +460,7 @@ public class Queue extends BaseDestinati
} else if (isAllConsumersExclusiveByDefault()) {
Subscription exclusiveConsumer = null;
for (Subscription s : consumers) {
- if (exclusiveConsumer == null
+ if (exclusiveConsumer == null
|| s.getConsumerInfo().getPriority() > exclusiveConsumer
.getConsumerInfo().getPriority()) {
exclusiveConsumer = s;
@@ -670,11 +669,11 @@ public class Queue extends BaseDestinati
Exception {
final ConnectionContext context = producerExchange.getConnectionContext();
Future<Object> result = null;
-
+
checkUsage(context, message);
sendLock.lockInterruptibly();
try {
- if (store != null && message.isPersistent()) {
+ if (store != null && message.isPersistent()) {
message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
if (messages.isCacheEnabled()) {
result = store.asyncAddQueueMessage(context, message);
@@ -690,7 +689,7 @@ public class Queue extends BaseDestinati
// a big TX does not blow up
// our memory. This increment is decremented once the tx finishes..
message.incrementReferenceCount();
-
+
context.getTransaction().addSynchronization(new Synchronization() {
@Override
public void afterCommit() throws Exception {
@@ -749,11 +748,11 @@ public class Queue extends BaseDestinati
}
} else if (messages.getSystemUsage() != null && systemUsage.getTempUsage().isFull()) {
final String logMessage = "Usage Manager Temp Store is Full ("
- + systemUsage.getTempUsage().getPercentUsage() + "% of " + systemUsage.getTempUsage().getLimit()
+ + systemUsage.getTempUsage().getPercentUsage() + "% of " + systemUsage.getTempUsage().getLimit()
+"). Stopping producer (" + message.getProducerId()
+ ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "."
+ " See http://activemq.apache.org/producer-flow-control.html for more info";
-
+
waitForSpace(context, messages.getSystemUsage().getTempUsage(), logMessage);
}
}
@@ -1154,7 +1153,7 @@ public class Queue extends BaseDestinati
/**
* Removes the messages matching the given selector
- *
+ *
* @return the number of messages removed
*/
public int removeMatchingMessages(String selector) throws Exception {
@@ -1164,7 +1163,7 @@ public class Queue extends BaseDestinati
/**
* Removes the messages matching the given selector up to the maximum number
* of matched messages
- *
+ *
* @return the number of messages removed
*/
public int removeMatchingMessages(String selector, int maximumMessages) throws Exception {
@@ -1174,12 +1173,12 @@ public class Queue extends BaseDestinati
/**
* Removes the messages matching the given filter up to the maximum number
* of matched messages
- *
+ *
* @return the number of messages removed
*/
public int removeMatchingMessages(MessageReferenceFilter filter, int maximumMessages) throws Exception {
int movedCounter = 0;
- Set<MessageReference> set = new CopyOnWriteArraySet<MessageReference>();
+ Set<MessageReference> set = new HashSet<MessageReference>();
ConnectionContext context = createConnectionContext();
do {
doPageIn(true);
@@ -1215,7 +1214,7 @@ public class Queue extends BaseDestinati
/**
* Copies the messages matching the given selector
- *
+ *
* @return the number of messages copied
*/
public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest)
@@ -1226,7 +1225,7 @@ public class Queue extends BaseDestinati
/**
* Copies the messages matching the given selector up to the maximum number
* of matched messages
- *
+ *
* @return the number of messages copied
*/
public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest,
@@ -1237,14 +1236,14 @@ public class Queue extends BaseDestinati
/**
* Copies the messages matching the given filter up to the maximum number of
* matched messages
- *
+ *
* @return the number of messages copied
*/
public int copyMatchingMessages(ConnectionContext context, MessageReferenceFilter filter, ActiveMQDestination dest,
int maximumMessages) throws Exception {
int movedCounter = 0;
int count = 0;
- Set<MessageReference> set = new CopyOnWriteArraySet<MessageReference>();
+ Set<MessageReference> set = new HashSet<MessageReference>();
do {
int oldMaxSize = getMaxPageSize();
setMaxPageSize((int) this.destinationStatistics.getMessages().getCount());
@@ -1311,7 +1310,7 @@ public class Queue extends BaseDestinati
/**
* Moves the messages matching the given selector
- *
+ *
* @return the number of messages removed
*/
public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest)
@@ -1335,7 +1334,7 @@ public class Queue extends BaseDestinati
public int moveMatchingMessagesTo(ConnectionContext context, MessageReferenceFilter filter,
ActiveMQDestination dest, int maximumMessages) throws Exception {
int movedCounter = 0;
- Set<QueueMessageReference> set = new CopyOnWriteArraySet<QueueMessageReference>();
+ Set<QueueMessageReference> set = new HashSet<QueueMessageReference>();
do {
doPageIn(true);
pagedInMessagesLock.readLock().lock();
@@ -1632,8 +1631,8 @@ public class Queue extends BaseDestinati
messagesLock.writeLock().unlock();
}
}
-
- final void messageSent(final ConnectionContext context, final Message msg) throws Exception {
+
+ final void messageSent(final ConnectionContext context, final Message msg) throws Exception {
destinationStatistics.getEnqueues().increment();
destinationStatistics.getMessages().increment();
messageDelivered(context, msg);
@@ -1751,7 +1750,7 @@ public class Queue extends BaseDestinati
try {
resultList = new ArrayList<QueueMessageReference>(result.size());
for (QueueMessageReference ref : result) {
- if (!pagedInMessages.containsKey(ref.getMessageId())) {
+ if (!pagedInMessages.containsKey(ref.getMessageId())) {
pagedInMessages.put(ref.getMessageId(), ref);
resultList.add(ref);
} else {
@@ -2065,7 +2064,7 @@ public class Queue extends BaseDestinati
/**
* Find a consumer that matches the id in the message dispatch notification
- *
+ *
* @param messageDispatchNotification
* @return sub or null if the subscription has been removed before dispatch
* @throws JMSException
Added: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/LargeQueueSparseDeleteTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/LargeQueueSparseDeleteTest.java?rev=1125427&view=auto
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/LargeQueueSparseDeleteTest.java (added)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/LargeQueueSparseDeleteTest.java Fri May 20 15:16:51 2011
@@ -0,0 +1,200 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.usecases;
+
+import javax.jms.Connection;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import org.apache.activemq.EmbeddedBrokerTestSupport;
+import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
+import org.apache.activemq.broker.region.Queue;
+import org.apache.activemq.broker.ConnectionContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.junit.Assert;
+
+/**
+ * This unit test creates a fixed size queue and moves the last message in the
+ * queue to another queue. The test is used to very the performance of
+ * {@link org.apache.activemq.broker.region.Queue#moveMatchingMessagesTo(org.apache.activemq.broker.ConnectionContext, String, org.apache.activemq.command.ActiveMQDestination)}.
+ */
+public class LargeQueueSparseDeleteTest extends EmbeddedBrokerTestSupport {
+ private static final Logger LOG = LoggerFactory.getLogger(LargeQueueSparseDeleteTest.class);
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void setUp() throws Exception {
+ super.useTopic = false;
+ super.setUp();
+ }
+
+ /**
+ * The test queue is filled with QUEUE_SIZE test messages, each with a
+ * numeric id property beginning at 0. Once the queue is filled, the last
+ * message (id = QUEUE_SIZE-1) is moved to another queue. The test succeeds
+ * if the move completes within TEST_TIMEOUT milliseconds.
+ *
+ * @throws Exception
+ */
+ public void testMoveMessages() throws Exception {
+ final int QUEUE_SIZE = 30000;
+ final String MOVE_TO_DESTINATION_NAME = getDestinationString()
+ + ".dest";
+ final long TEST_TIMEOUT = 6000;
+
+ // Populate a test queue with uniquely-identifiable messages.
+ Connection conn = createConnection();
+ try {
+ conn.start();
+ Session session = conn.createSession(true,
+ Session.SESSION_TRANSACTED);
+ MessageProducer producer = session.createProducer(destination);
+ for (int i = 0; i < QUEUE_SIZE; i++) {
+ Message message = session.createMessage();
+ message.setIntProperty("id", i);
+ producer.send(message);
+ }
+ session.commit();
+ } finally {
+ conn.close();
+ }
+
+ // Access the implementation of the test queue and move the last message
+ // to another queue. Verify that the move occurred within the limits of
+ // the test.
+ Queue queue = (Queue) broker.getRegionBroker().getDestinationMap().get(
+ destination);
+
+ ConnectionContext context = new ConnectionContext(
+ new NonCachedMessageEvaluationContext());
+ context.setBroker(broker.getBroker());
+ context.getMessageEvaluationContext().setDestination(destination);
+
+ long startTimeMillis = System.currentTimeMillis();
+ Assert.assertEquals(1, queue
+ .moveMatchingMessagesTo(context, "id=" + (QUEUE_SIZE - 1),
+ createDestination(MOVE_TO_DESTINATION_NAME)));
+
+ long durationMillis = System.currentTimeMillis() - startTimeMillis;
+
+ LOG.info("It took " + durationMillis
+ + "ms to move the last message from a queue a " + QUEUE_SIZE
+ + " messages.");
+
+ Assert.assertTrue("Moving the message took too long: " + durationMillis
+ + "ms", durationMillis < TEST_TIMEOUT);
+ }
+
+ public void testCopyMessages() throws Exception {
+ final int QUEUE_SIZE = 30000;
+ final String MOVE_TO_DESTINATION_NAME = getDestinationString()
+ + ".dest";
+ final long TEST_TIMEOUT = 6000;
+
+ // Populate a test queue with uniquely-identifiable messages.
+ Connection conn = createConnection();
+ try {
+ conn.start();
+ Session session = conn.createSession(true,
+ Session.SESSION_TRANSACTED);
+ MessageProducer producer = session.createProducer(destination);
+ for (int i = 0; i < QUEUE_SIZE; i++) {
+ Message message = session.createMessage();
+ message.setIntProperty("id", i);
+ producer.send(message);
+ }
+ session.commit();
+ } finally {
+ conn.close();
+ }
+
+ // Access the implementation of the test queue and move the last message
+ // to another queue. Verify that the move occurred within the limits of
+ // the test.
+ Queue queue = (Queue) broker.getRegionBroker().getDestinationMap().get(
+ destination);
+
+ ConnectionContext context = new ConnectionContext(
+ new NonCachedMessageEvaluationContext());
+ context.setBroker(broker.getBroker());
+ context.getMessageEvaluationContext().setDestination(destination);
+
+ long startTimeMillis = System.currentTimeMillis();
+ Assert.assertEquals(1,
+ queue.copyMatchingMessagesTo(context, "id=" + (QUEUE_SIZE - 1), createDestination(MOVE_TO_DESTINATION_NAME)));
+
+ long durationMillis = System.currentTimeMillis() - startTimeMillis;
+
+ LOG.info("It took " + durationMillis
+ + "ms to copy the last message from a queue a " + QUEUE_SIZE
+ + " messages.");
+
+ Assert.assertTrue("Copying the message took too long: " + durationMillis
+ + "ms", durationMillis < TEST_TIMEOUT);
+ }
+
+ public void testRemoveMessages() throws Exception {
+ final int QUEUE_SIZE = 30000;
+ final long TEST_TIMEOUT = 6000;
+
+ // Populate a test queue with uniquely-identifiable messages.
+ Connection conn = createConnection();
+ try {
+ conn.start();
+ Session session = conn.createSession(true,
+ Session.SESSION_TRANSACTED);
+ MessageProducer producer = session.createProducer(destination);
+ for (int i = 0; i < QUEUE_SIZE; i++) {
+ Message message = session.createMessage();
+ message.setIntProperty("id", i);
+ producer.send(message);
+ }
+ session.commit();
+ } finally {
+ conn.close();
+ }
+
+ // Access the implementation of the test queue and move the last message
+ // to another queue. Verify that the move occurred within the limits of
+ // the test.
+ Queue queue = (Queue) broker.getRegionBroker().getDestinationMap().get(
+ destination);
+
+ ConnectionContext context = new ConnectionContext(
+ new NonCachedMessageEvaluationContext());
+ context.setBroker(broker.getBroker());
+ context.getMessageEvaluationContext().setDestination(destination);
+
+ long startTimeMillis = System.currentTimeMillis();
+ Assert.assertEquals(1,
+ queue.removeMatchingMessages("id=" + (QUEUE_SIZE - 1)));
+
+ long durationMillis = System.currentTimeMillis() - startTimeMillis;
+
+ LOG.info("It took " + durationMillis
+ + "ms to remove the last message from a queue a " + QUEUE_SIZE
+ + " messages.");
+
+ Assert.assertTrue("Removing the message took too long: " + durationMillis
+ + "ms", durationMillis < TEST_TIMEOUT);
+ }
+}
Propchange: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/LargeQueueSparseDeleteTest.java
------------------------------------------------------------------------------
svn:eol-style = native