RabbitMQ Publisher Subscribe Example in Java on Ubuntu

To install RabbitMQ on Ubuntu is as simple as executing:

sudo apt-get install rabbitmq-server

Next we need to download the java client libraries as they are not included in the ubuntu package. You can find the libraries at http://www.rabbitmq.com/java-client.html Unzip the archive and include the jars in your classpath

First we need a way of creating connections to RabbitMQ and also a method to setup exchanges, queues and routing keys: import com.rabbitmq.client.*; import java.io.IOException; import java.util.logging.Level; import java.util.logging.Logger; public class RabbitMQFactory { private static int DEFAULT_PORT = 5672; // The default port RabbitMQ runs on public static final String EXCHANGE = "testExchange"; public static final String QUEUE = "testQueue"; public static final String ROUTINGKEY = "testRoutingKey"; public static Channel getDefaultChannel() throws IOException { ConnectionFactory factory = new ConnectionFactory(); factory.setPassword("guest"); factory.setPassword("guest"); factory.setVirtualHost("/"); factory.setHost("localhost"); factory.setPort(DEFAULT_PORT); return factory.newConnection().createChannel(); } public static void createQueue(String exchange, String queue, String key) { try { Channel channel = getDefaultChannel(); channel.exchangeDeclare(exchange, "direct", true); channel.queueDeclare(queue, true, false, false, null); channel.queueBind(queue, exchange, key); } catch (IOException ex) { Logger.getLogger(RabbitMQFactory.class.getName()).log(Level.SEVERE, null, ex); } } // Create the topics public static void main(String[] args) { createQueue(EXCHANGE, QUEUE, ROUTINGKEY); } } Run the main method to setup our test queue.

Then create the publisher: import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import java.io.IOException; import java.util.Date; import java.util.Random; public class Producer { public static void main(String[] args) throws IOException, InterruptedException { Random random = new Random(); Channel channel = RabbitMQFactory.getDefaultChannel(); while(true) { String msgText = "Current time is: " + new Date().toString(); System.out.println("Adding msg: " + msgText); channel.basicPublish(RabbitMQFactory.EXCHANGE, RabbitMQFactory.ROUTINGKEY, null, msgText.getBytes()); Thread.sleep(random.nextInt(5) * 1000); } } }

Then the consumer import com.rabbitmq.client.Channel; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.client.QueueingConsumer.Delivery; import java.io.IOException; import java.util.Date; import java.util.Random; public class Consumer { public static void main(String[] args) throws IOException, InterruptedException { Random random = new Random(); Channel channel = RabbitMQFactory.getDefaultChannel(); QueueingConsumer cons = new QueueingConsumer(channel); channel.basicConsume(RabbitMQFactory.QUEUE, false, cons); while(true) { Delivery delivery = cons.nextDelivery(); System.out.println("Got Message: " + new String(delivery.getBody())); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }

The execute the main method on the Publisher and Consumer and watch the messages flow.