Setting Message Priority Levels
Message priority levels can be used to instruct the JMS provider to deliver urgent messages first. The message’s priority is contained in the JMSPriority header. You can set the priority level in either of two ways:- You can use the
setPriority()method of theMessageProducerinterface to set the priority level for all messages sent by that producer. For example, the following call sets a priority level of '7' for a producer:
producer.setPriority(7);
- You can use the long form of the
send()or thepublish()method to set the priority level for a specific message. The third argument sets the priority level. For example, the following send call sets the priority level for message to '3':
producer.send(message, DeliveryMode.NON_PERSISTENT, 3, 0);
Setting the priority directly on the JMS Message using theThe ten levels of priority range from 0 (lowest) to 9 (highest). If you do not specify a priority level, the default level is 4.setJMSPriority()method of theMessageinterface does not work as in that case the priority of the producer is taken!
A JMS provider tries to deliver higher-priority messages before lower-priority ones but depending on the setup and configuration it will not always deliver messages in exact order of priority. This last point is an import one to note when designing your application.For example on ActiveMQ, once you hit a situation where consumers are slow, or producers are just significantly faster, you'll observe that the cache will fill up (possibly with lower priority messages) while higher priority messages get stuck on disk and are not available until they're paged in. In this case, you can make a decision to tradeoff optimized message dispatching for priority enforcement. You can disable the cache, message expiration check, and lower you consumer prefetch to 1 to ensure getting the high priority messages from the store ahead of lower priority messages. However this sort of tradeoff can have significant performance implications, so always test your scenarios thoroughly.
ActiveMQ Priority Example
Let's illustrate the above by creating a simple producer with two differentsend() methods. The first method will send a message with the default priority level and the second method will accept an additional parameter specifying the priority to be set on the message. The example uses Maven and assumes a default ActiveMQ message broker is up and running.Tools used:
- ActiveMQ 5.10
- Maven 3
First let's look at the below Maven POM file which contains the needed dependencies for Logback, JUnit and ActiveMQ.
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>info.source4code</groupId>
<artifactId>jms-activemq-priority</artifactId>
<version>1.0</version>
<packaging>jar</packaging>
<name>JMS - Priority using ActiveMQ</name>
<url>http://www.source4code.info/2014/01/jms-priority-using-activemq.html</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>1.6</java.version>
<slf4j.version>1.7.7</slf4j.version>
<ch.qos.logback.version>1.1.2</ch.qos.logback.version>
<junit.version>4.12-beta-1</junit.version>
<activemq.version>5.10.0</activemq.version>
<maven-compiler-plugin.version>3.1</maven-compiler-plugin.version>
</properties>
<dependencies>
<!-- Logging -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>${slf4j.version}</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>${ch.qos.logback.version}</version>
</dependency>
<!-- JUnit -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>${junit.version}</version>
</dependency>
<!-- ActiveMQ -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>${activemq.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>${maven-compiler-plugin.version}</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
Next is the
Producer class which contains the two send() methods: one which applies default priority and one which applies a custom priority. The class also contains two methods for opening/closing a connection to the message broker as well as a method for creating the message producer. package info.source4code.jms.priority;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class Producer {
private static final Logger LOGGER = LoggerFactory
.getLogger(Producer.class);
private Connection connection;
private Session session;
private MessageProducer producer;
public void openConnection() throws JMSException {
// Create a new connection factory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_BROKER_URL);
connection = connectionFactory.createConnection();
}
public void closeConnection() throws JMSException {
connection.close();
}
public void createProducer(String queue) throws JMSException {
// Create a session for sending messages
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// Create the destination to which messages will be sent
Destination destination = session.createQueue(queue);
// Create a MessageProducer for sending the messages
producer = session.createProducer(destination);
}
public void send(String text) throws JMSException {
TextMessage message = session.createTextMessage(text);
producer.send(message);
LOGGER.info("{} sent with default priority(=4)", text);
}
public void send(String text, int priority) throws JMSException {
TextMessage message = session.createTextMessage(text);
// Note: setting the priority directly on the JMS Message does not work
// as in that case the priority of the producer is taken
producer.send(message, DeliveryMode.PERSISTENT, priority, 0);
LOGGER.info("{} sent with priority={}", text, priority);
}
}
For receiving the messages a
Consumer class with again two methods for opening/closing a connection to the message broker is shown below. In addition a method to create a message consumer for a specific queue and a method to receive a single message are available on the class. package info.source4code.jms.priority;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class Consumer {
private static final Logger LOGGER = LoggerFactory
.getLogger(Consumer.class);
private Connection connection;
private Session session;
private MessageConsumer consumer;
public void openConnection() throws JMSException {
// Create a new connection factory
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_BROKER_URL);
connection = connectionFactory.createConnection();
}
public void closeConnection() throws JMSException {
connection.close();
}
public void createConsumer(String queue) throws JMSException {
// Create a session for receiving messages
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// Create the destination from which messages will be received
Destination destination = session.createQueue(queue);
// Create a MessageConsumer for receiving messages
consumer = session.createConsumer(destination);
// Start the connection in order to receive messages
connection.start();
}
public String receive(int timeout) throws JMSException {
// Read a message from the destination
Message message = consumer.receive(timeout);
// Cast the message to the correct type
TextMessage input = (TextMessage) message;
// Retrieve the message content
String text = input.getText();
LOGGER.info("{} received", text);
return text;
}
}
Last, a JUnit test class in which a first
testSend() test case will send three messages with default priority to a 'priority.q' queue and then verifies if the messages are read in first in, first out (FIFO) order. And then a second testSendWithPriority() test case which will send three messages with custom priorities where the last message gets the highest priority and then verifies if the message are read in last in, first out (LIFO) order. package info.source4code.jms.priority;
import static org.junit.Assert.assertEquals;
import javax.jms.JMSException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
public class ProducerTest {
private static Consumer consumer;
private static Producer producer;
private static String QUEUE = "priority.q";
@BeforeClass
public static void setUpBeforeClass() throws JMSException {
producer = new Producer();
producer.openConnection();
producer.createProducer(QUEUE);
consumer = new Consumer();
consumer.openConnection();
consumer.createConsumer(QUEUE);
}
@AfterClass
public static void tearDownAfterClass() throws JMSException {
producer.closeConnection();
consumer.closeConnection();
}
@Test
public void testSend() throws JMSException, InterruptedException {
producer.send("message1");
producer.send("message2");
producer.send("message3");
Thread.sleep(1000);
// Messages should be received FIFO as priority is the same for all
assertEquals("message1", consumer.receive(5000));
assertEquals("message2", consumer.receive(5000));
assertEquals("message3", consumer.receive(5000));
}
@Test
public void testSendWithPriority() throws JMSException,
InterruptedException {
producer.send("message1", 1);
producer.send("message2", 2);
producer.send("message3", 3);
Thread.sleep(1000);
// Messages should be received LIFO as priority=1 is lower than
// priority=3
assertEquals("message3", consumer.receive(5000));
assertEquals("message2", consumer.receive(5000));
assertEquals("message1", consumer.receive(5000));
}
}
Make sure a default ActiveMQ message broker is up and running, open a command prompt and execute following Maven command:
mvn test
This will trigger Maven to run the above test case and will result in the following log statements. Even though 'message1' was sent first in both test cases, in the first test it is received first whereas in the second test it is received last because of the different assigned priority.
23:24:30.126 [main] INFO i.source4code.jms.priority.Producer -
message1 sent with default priority(=4)
23:24:30.152 [main] INFO i.source4code.jms.priority.Producer -
message2 sent with default priority(=4)
23:24:30.175 [main] INFO i.source4code.jms.priority.Producer -
message3 sent with default priority(=4)
23:24:31.177 [main] INFO i.source4code.jms.priority.Consumer -
message1 received
23:24:31.177 [main] INFO i.source4code.jms.priority.Consumer -
message2 received
23:24:31.178 [main] INFO i.source4code.jms.priority.Consumer -
message3 received
23:24:31.203 [main] INFO i.source4code.jms.priority.Producer -
message1 sent with priority=1
23:24:31.226 [main] INFO i.source4code.jms.priority.Producer -
message2 sent with priority=2
23:24:31.253 [main] INFO i.source4code.jms.priority.Producer -
message3 sent with priority=3
23:24:32.253 [main] INFO i.source4code.jms.priority.Consumer -
message3 received
23:24:32.254 [main] INFO i.source4code.jms.priority.Consumer -
message2 received
23:24:32.254 [main] INFO i.source4code.jms.priority.Consumer -
message1 received
By browsing the 'priority.q' using the ActiveMQ console we can verify the
JMSPriority header set on the messages sent by the above test cases. The testSend() test case will create three messages a shown below, each with priority set to 4.
The second
testSendWithPriority() test case results in three messages, each with a different priority ranging from 1 till 3.
If you would like to run the above code sample you can download the full source code and their corresponding JUnit test cases here.
This concludes the priority using ActiveMQ example. If you found this post helpful or have any questions or remarks, please leave a comment.
