Hi,<br>I'm developing a stress test to compare and measure the performance between RabbitMQ and ActiveMQ. Also the objective was to compare the standard AMQP with the JMS specifications. There is something that I would like to share with the community since I think in could be a possible bug or a memory leak problem. The stress test is very easy; there are no exponential ore base variables, just a linear test case. I'm going to describe the scenario. <br>
The scenario is composed by two JAVA clients: a publisher and a subscriber, and one instance of RabbitMQ (V. 2.2.0) running over a virtual machine 2GB RAM. The clients, both publisher and subscriber are placed in two different physical machines. The messages to be sent by the publisher and received by the subscriber are described within a cluster. A cluster specifies the size of the body message (in byte) and the total number of messages to send and receive for that dimension. Ex: C1 = (512::10000). The publisher uses a list of clusters, and for each one of them it starts a new loop in which messages are sent. Es: C1 = (512::500000); C2 = (1024::100000); C3 = (2048::10000). <br>
At begin and at the end of each cluster, the publisher sends a special message used to inform the subscriber that a new cluster is started or is just finished. Also a special message is sent to the subscriber when all the clusters have been elaborated. The subscriber does nothing more than receiving messages and calculates the elapsed time (in seconds) between the start and the end message for each cluster. The Publisher uses a "direct" exchange (non-durable, non auto-delete) to publish the messages while the subscriber creates and binds a new queue over the same "direct" exchange.<br>
<br>This is the code used to initialize the publisher:<br><br>ConnectionFactory factory = new ConnectionFactory();<br>factory.setUsername("guest");<br>factory.setPassword("guest");<br>factory.setVirtualHost("/");<br>
factory.setHost("RabbitMQ_Srv_01");<br>factory.setPort(5672);<br>conn = factory.newConnection();<br>channel = conn.createChannel();<br>boolean durable = false;<br>boolean autoDelete = false;<br>channel.exchangeDeclare("linearTest", "direct", durable, autoDelete, null);<br>
...<br>//Define some properties<br>AMQP.BasicProperties props = MessageProperties.PERSISTENT_TEXT_PLAIN;<br>Map<String, Object> headers = new HashMap<String, Object>();<br>headers.put("TestCommand", CURRENT_COMMAND);<br>
...<br>props.setHeaders(headers);<br>//Basic publish<br>channel.basicPublish("linearExch", "linearKey", props, CLUSTER[i].messageBodyBytes);<br>...<br><br>This is the code used to initialize the subscriber:<br>
<br>ConnectionFactory factory = new ConnectionFactory();<br>factory.setUsername("guest");<br>factory.setPassword("guest");<br>factory.setVirtualHost("/");<br>factory.setHost("RabbitMQ_Srv_01");<br>
factory.setPort(5672);<br>Connection conn = factory.newConnection();<br>Channel channel = conn.createChannel();<br>boolean durable = false;<br>boolean autoDelete = false;<br>boolean exclusive = false;<br>channel.exchangeDeclare("linearExch", "direct", durable, autoDelete, null);<br>
channel.queueDeclare("linearQueue", durable, exclusive, autoDelete, null);<br>channel.queueBind("linearQueue", "linearExch", "linearKey", null);<br>QueueingConsumer consumer = new QueueingConsumer(channel);<br>
boolean autoAck = false;<br>channel.basicConsume("linearQueue", autoAck, consumer);<br> <br>while (true) {<br>try {<br>QueueingConsumer.Delivery delivery = consumer.nextDelivery();<br>onMessage(delivery);<br>channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);<br>
//If shutdown from the publisher is received<br>if (shutdown) {<br>channel.close();<br>conn.close();<br>break;<br>}<br>}<br>catch (InterruptedException ie) {<br>continue;<br>}<br>}<br><br>This is how I've set up the test case:<br>
<br>1. Started the RabbitMQ broker by using the .bat "rubbitmq-server.bat" (NO extra configuration used).<br>2. Open the task manager to view the RabbitMQ (.erl process) memory usage.<br>3. Started the subscriber.<br>
4. Started the publisher.<br>5. Defined Clusters As: {C1(512:10000); C2(512:10000); C3(512:10000); C4(512:100000); C5(512:100000); C6(512:100000); C7(1024:10000); C8(1024:10000); C9(1024:10000); C10(1024:100000); C11(1024:100000); C12(1024:100000); C13(2048:10000); C14(2048:10000); C15(2048:10000); C16(2048:100000); C17(2048:100000); C18(2048:100000) }<br>
<br>These are some results and notifications:<br><br>. I've started a first test case, I've noticed that RabbitMQ uses lots of memory during the test. At the first step, where only the broker was running, just 19 KB of memory were used. At the end of the process the memory usage increases up to more than 850 KB. After stopping the publisher and the subscriber, the memory usage was very slowly decreasing from 850-800KB down to 50-20 KB. <br>
<br>. I've started a second test case but this time using the publisher and the subscriber over the same physical machine. After a few minutes the broker crashed down showing the following message: "eheap_alloc: Cannot allocate 373662860 bytes of memory (of type "heap")". I've repeated the same test more and more but every time the broker crashed down.<br>
<br>. I’ve started a third test case, similar to the first one (publisher and subscriber placed in two different machines) but this time using the following cluster list: {C1(512:10000); C2(512:10000); C3(512:10000); C4(512:100000); C5(512:100000); C6(512:100000); C7(512:500000); C8(512:500000); C9(512:500000)}. At the time of elaborating the last element (C9) the broker crashed down showing the same message as the one in the previous test. <br>
<br>I think that the reason for the second test case can be found in the network usage, which causes the broker to store in the memory a lots of message because publisher and subscriber are placed in the same machine and, it’s clearly that the subscriber leads substantially the usage and the performance of the process (maybe because of the acknowledgements). But I don’t have a good idea for the third test case, which causes the same error.<br>
<br>Now I’m facing with the fact that, comparing to ActiveMQ, the second test and the third test are successfully done, in fact, even if the subscriber and the publisher are placed in the same machine (the ActiveMQ broker runs in the same virtual machine as the one used for the RabbitMQ) the test case comes successfully to the end, and the broke doesn’t crash down. However, the first test case (described in the first point) seems to be more efficient, in terms of seconds, by using RabbitMQ. <br>
<br>Here some questions:<br><br>. In general, what do you think about the test? <br>. Am I adopting a very strange test case which is not well handled by RabbitMQ? <br>. Is there any connection with the solved bug described in the RabbitMQ 2.2 release notes (“fix memory leak when long-running channels consume and cancel on many queues”)?<br>
. How does RabbitMQ handle message flooding? As stated in ActiveMQ official page, the flow control, in the current version means that: “if the broker detects that the memory limit for the destination, or the temp- or file-store limits for the broker, have been exceeded, then the flow of messages can be slowed down. The producer will be either blocked until resources are available or will receive a JMSException”.<br>
<br>Thanks for any response. <br>