{"id":44550,"date":"2017-01-17T14:07:29","date_gmt":"2017-01-17T08:37:29","guid":{"rendered":"http:\/\/www.tothenew.com\/blog\/?p=44550"},"modified":"2017-01-17T15:20:25","modified_gmt":"2017-01-17T09:50:25","slug":"how-to-integrate-rabbitmq-using-spring","status":"publish","type":"post","link":"https:\/\/www.tothenew.com\/blog\/how-to-integrate-rabbitmq-using-spring\/","title":{"rendered":"How To Integrate RabbitMQ using Spring?"},"content":{"rendered":"

In this blog, we will see two different implementations of RabbitMQ, but before going to the implementation part let’s take a brief intro of some prerequisites.<\/span><\/p>\n

JMS <\/b><\/span><\/p>\n

The Java Message Service (JMS) is a Message Oriented Middleware Java API that supports the formal communication between software components. It allows applications to create, send, receive, and read messages in such a way that communication is loosely coupled, reliable and asynchronous. <\/span><\/p>\n

AMQP <\/b><\/span><\/p>\n

The Advanced Message Queuing Protocol (AMQP) is an open standard for message-oriented middlewares. AMQP creates full functional interoperability between conforming clients and messaging middleware servers (also called “brokers”). <\/span><\/p>\n

What is RabbitMQ? <\/b><\/span><\/p>\n

RabbitMQ is an open source message-oriented middleware (also called “brokers”) that implements the Advanced Message Queuing Protocol (AMQP). <\/span><\/p>\n

Benefits of using RabbitMQ<\/span><\/p>\n

    \n
  1. Robust messaging for applications <\/span><\/li>\n
  2. Runs on all major operating systems<\/li>\n
  3. Supports a huge number of developer platforms <\/span><\/li>\n
  4. Open source and commercially supported<\/span><\/li>\n
  5. Easy to use<\/li>\n<\/ol>\n

    Installation Steps<\/b><\/span><\/p>\n

      \n
    1. You can use below link to install RabbitMQ Server in your computer.<\/span><\/li>\n<\/ol>\n

      https:\/\/www.rabbitmq.com\/install-debian.html .<\/span><\/p>\n

      Now use below command to enable RabbitMQ Managment Plugin. <\/span><\/p>\n

      rabbitmq-plugins enable rabbitmq_management<\/span><\/em><\/p>\n

      2. Check\u00a0<\/span>localhost:15672<\/span><\/em><\/p>\n

      Default user name=guest<\/em><\/p>\n

      Default password=guest<\/em><\/span><\/p>\n

      For complete reference Please Visit https:\/\/www.rabbitmq.com\/management.html.<\/span><\/p>\n

      Now you’re ready to use RabbitMQ.<\/span><\/p>\n

      RabbitMQ Use Case<\/span><\/strong><\/p>\n

      Let’s take a scenario in which we have to perform multiple different tasks by events we pass in the queue. We have Producer (who sends a message) and Consumer (who receives a message).<\/span><\/p>\n

      Consumer passes a message to the EventHandler Class. It is explained more clearly in the following steps. <\/span><\/p>\n

      Here we will see two different implementations in Spring. First JMS Base integration and other Spring particular AMQP base integration.<\/span><\/p>\n

      JMS Base Integration:<\/strong>
      \n<\/span><\/p>\n

      Step 1.1: Adding Library Dependency <\/b><\/span><\/p>\n

      To get started with RabbitMQ Java AMQP, it is recommended to use a dependency management system. <\/span><\/p>\n

      The maven dependency is given below: <\/span><\/p>\n

      [code language=”xml”]
      \n<dependency>
      \n\t<groupId>com.rabbitmq<\/groupId>
      \n\t<artifactId>amqp-client<\/artifactId>
      \n\t<version>4.0.0<\/version>
      \n<\/dependency>
      \n[\/code]<\/p>\n

      Alternatively, if you’re using Gradle: <\/span><\/p>\n

      [code language=”xml”]
      \ndependencies {
      \n compile ‘com.rabbitmq:amqp-client:4.0.0’
      \n}
      \n[\/code]<\/p>\n

      Step 1.2: Create Connector Class<\/b><\/span><\/p>\n

      Now, we need to create an abstract Connector class. Basic implementation of RabbitMQ is same for Producer as well as Consumer. We can generalize our code in the connector class and Producer\/Consumer will have to inherit this class.<\/span><\/p>\n

      [code language=”java”]
      \npublic abstract class Connector {
      \n\tprotected Channel myChannel;
      \n\tprotected Connection connection;
      \n\tprotected String queueName;<\/p>\n

      \tpublic Connector(String queueName) throws IOException,TimeoutException {
      \n\t\tthis.queueName=queueName;
      \n\t\tConnectionFactory connectionFactory = new ConnectionFactory();
      \n\t\t\/\/ Hostname of your rabbitmq server
      \n\t\tconnectionFactory.setHost("localhost");
      \n\t\t\/\/ getting a connection
      \n\t\tconnection = connectionFactory.newConnection();<\/p>\n

      \t\t\/*this will create a new channel, using an internally allocated channel number or we can say it will simply declare a queue for this channel. If queue does not exist.*\/
      \n\t\tmyChannel= connection.createChannel();<\/p>\n

      \t\tmyChannel.queueDeclare(queueName, false, false, false, null);
      \n\t}<\/p>\n

      \tpublic void close() throws IOException, TimeoutException {
      \n\t\tthis.myChannel.close();
      \n\t\tthis.connection.close();
      \n\t}
      \n}
      \n[\/code]<\/p>\n

      Step 1.3: Create Producer<\/b><\/span><\/p>\n

      A program that sends messages is a Producer.<\/span><\/p>\n

      It has to inherit connector class so that producer is able to communicate with the queue.<\/span><\/p>\n

      sendMessage (Serializable object)<\/b> method will send the serializable data to the queue. <\/span><\/p>\n

      [code language=”java”]
      \npublic class Producer extends Connector {
      \n public Producer(String queueName) throws IOException, TimeoutException{<\/p>\n

      super(queueName);
      \n }<\/p>\n

      public void sendMessage(Serializable object) throws IOException {
      \n channel.basicPublish("",queueName, null, SerializationUtils.serialize(object));
      \n }
      \n}
      \n[\/code]<\/p>\n

      \u00a0Step 1.4: Create Consumer<\/b><\/span><\/p>\n

      A consumer is a program that mostly waits to receive messages. I<\/span>t has to inherit connector class so that Consumer is able to communicate with the queue.<\/p>\n

      Now the consumer will receive the serialized data (message) so we have to deserialize this data before use it.<\/span><\/p>\n

      public void handleDelivery(String consumerTag, Envelope env, BasicProperties props, byte[] body).<\/b> <\/span><\/p>\n

      This method will receive the message and body is the serialized data which we will deserialize before using it.<\/span><\/p>\n

      [code language=”java”]
      \npublic class QueueConsumer extends Connector implements Runnable, Consumer { <\/p>\n

      public QueueConsumer(String queueName) throws IOException, TimeoutException {
      \n super(queueName);
      \n } <\/p>\n

      public void run() {
      \n try {
      \n \/\/ start consuming messages. Auto acknowledge messages.
      \n channel.basicConsume(queueName, true, this);
      \n } catch (IOException e) {
      \n e.printStackTrace();
      \n }
      \n }
      \n public void handleConsumeOk(String info) {
      \n \/\/Called when the consumer is registered. <\/p>\n

      } <\/p>\n

      \/\/ Called when new message is available. <\/p>\n

      public void handleDelivery(String info, Envelope env, BasicProperties props, byte[] body)
      \n throws IOException {
      \n Map map = (HashMap) SerializationUtils.deserialize(body);
      \n EventHandler.handler((Events) map.get("event"), map.get("message"));
      \n } <\/p>\n

      public void handleCancel(String info) {
      \n \/\/Called when the consumer is canceled..
      \n } <\/p>\n

      public void handleCancelOk(String info) {
      \n } <\/p>\n

      public void handleRecoverOk(String info) {
      \n } <\/p>\n

      public void handleShutdownSignal(String info, ShutdownSignalException exception) {
      \n }
      \n}
      \n[\/code]<\/p>\n

      Step 1.5: Create enum Events<\/b><\/span><\/p>\n

      These are the events which will be used to decide which task to perform.<\/span><\/p>\n

      [code language=”java”]
      \nenum Events {
      \nEvent1,Event2;<\/p>\n

      }
      \n[\/code]<\/p>\n

      Step 1.6: Create class EventHandler<\/b><\/span><\/p>\n

      This class will handle all the events you want to trigger whenever consumer consumes a message.<\/span><\/p>\n

      [code language=”java”]
      \npublic class EventHandler {
      \n static void handler(Events event, Object message) {
      \n switch (event) {
      \n case Event1:
      \n System.out.println("publish 1 event" + message);
      \n break;
      \n case Event2:
      \n System.out.println("publish event 2" + message);
      \n break;
      \n }
      \n }
      \n}<\/p>\n

      [\/code]<\/p>\n

      Step 1.7: Create a constant Class<\/b><\/span><\/p>\n

      This class will have the queue name.<\/span><\/p>\n

      [code language=”java”]
      \npublic final class Constants {
      \n public static final String queueName = "queueName"; <\/p>\n

      }
      \n[\/code]<\/p>\n

      Step 1.8: Create Service Class EventPublisherService<\/b><\/span><\/p>\n

      When the EventPublisherService bean is created at that moment Consumer, producer instant will be created and the consumer thread will be started. <\/span><\/p>\n

      public void publishEvent(Events event, Object messages) <\/b>method will create a message and that will be sent to the queue.<\/span><\/p>\n

      [code language=”java”]
      \n@Service
      \npublic class EventPublisherService {
      \n final private Producer producer; <\/p>\n

      EventPublisherService() throws IOException, TimeoutException {
      \n QueueConsumer consumer = new QueueConsumer(Constants.queueName);
      \n Thread consumerThread = new Thread(consumer);
      \n consumerThread.start();
      \n producer = new Producer(Constants.queueName);
      \n } <\/p>\n

      public void publishEvent(Events event, Object messages) throws IOException, TimeoutException { <\/p>\n

      HashMap<String, Object> message = new HashMap<String, Object>();
      \n message.put("event", event);
      \n message.put("message", messages);
      \n producer.sendMessage(message);
      \n }
      \n}
      \n[\/code]<\/p>\n

      Step 1.9: Create Controller <\/b><\/span><\/p>\n

      We just have to pass Event Type and message to publishEvent (Events, message) <\/b>method of EventPublisherService<\/b> and rest is handled by it.<\/span><\/p>\n

      [code language=”java”]
      \n@Controller
      \n@RequestMapping("\/")
      \npublic class UserController {
      \n @Autowired
      \n EventPublisherService eventPublisherService;
      \n @RequestMapping("\/rabbitMq")
      \n public ModelAndView welcome() throws IOException, TimeoutException {
      \n Integer message1 = 11111;
      \n eventPublisherService.publishEvent(Events.Event1, message1);
      \n String message2 = "Implementing RabbitMQ";
      \n eventPublisherService.publishEvent(Events.Event2, message2);
      \n return new ModelAndView("index", "message", "Implemented");
      \n }
      \n}<\/p>\n

      [\/code]<\/p>\n

      Spring Specific Implementation<\/b><\/span><\/p>\n

      There will be no Producer and consumer classes in this implementation.\u00a0<\/span>We only need to use below command\u00a0to send message.<\/p>\n

      template.convertAndSend(Constants.queueName,message);<\/span><\/p>\n

      And @RabbitListener(queues = \u201cqueueName\u201d) Annotation to receive a message. It is explained more clearly in the following steps:<\/span><\/p>\n

      Step 2.1: Adding Library Dependency <\/b><\/span><\/p>\n

      [code language=”xml”]
      \n<dependency>
      \n <groupId>org.springframework.amqp<\/groupId>
      \n <artifactId>spring-rabbit<\/artifactId>
      \n <version>1.4.5.RELEASE<\/version>
      \n<\/dependency>
      \n[\/code]<\/p>\n

      alternatively for Gradle<\/p>\n

      [code language=”java”]
      \ncompile group: ‘org.springframework.amqp’, name: ‘spring-rabbit’, version: ‘1.4.5.RELEASE’
      \n[\/code]<\/p>\n

      Step 2.2: Create a configuration class named RabbitMqConfiguration<\/b><\/span><\/p>\n

      Here we will create all the mandatory beans which are required By RabbitMQ.<\/span><\/p>\n

      [code language=”java”]
      \n@Configuration
      \npublic class RabbitMqConfiguration {
      \n @Bean
      \n public ConnectionFactory connectionFactory() {
      \n CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
      \n return connectionFactory;
      \n }<\/p>\n

      @Bean
      \n public AmqpAdmin amqpAdmin() {
      \n return new RabbitAdmin(connectionFactory());
      \n }<\/p>\n

      @Bean
      \n public RabbitTemplate rabbitTemplate() {
      \n return new RabbitTemplate(connectionFactory());
      \n }<\/p>\n

      @Bean
      \n public Queue myQueue() {
      \n return new Queue(Constants.queueName);
      \n }
      \n @Bean(name="rabbitListenerContainerFactory")
      \n public SimpleRabbitListenerContainerFactory rabbitListenerContainerlistenerFactory(){
      \n SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
      \n factory.setConnectionFactory(connectionFactory());
      \n return factory;
      \n }<\/p>\n

      }
      \n[\/code]<\/p>\n

      Step 2.3: Repeat Steps<\/b><\/span><\/p>\n

      The Events enum<\/span>, <\/span>EventHandler class, Constants class and <\/span><\/span>controller <\/span><\/span>Class<\/span><\/span>\u00a0<\/b>are same as created in Step 1.5,1.6,1.7 and 1.9.<\/span><\/p>\n

      Step 2.4: Create EventPublisherService<\/b><\/span><\/p>\n

      EventPublisherService will create a message and then send it to the queue using template.convertAndSend(Constants.queueName,message); <\/span><\/p>\n

      this service Act like a producer.<\/span><\/p>\n

      [code language=”java”]
      \n@Service
      \npublic class EventPublisherService {
      \n @Autowired
      \n AmqpTemplate template;
      \n public void publishEvent(Events event, Object messages) throws IOException, TimeoutException {
      \n Map<String, Object> message = new HashMap<String, Object>();
      \n message.put("event", event);
      \n message.put("message", messages);
      \n template.convertAndSend(Constants.queueName,message);
      \n }<\/p>\n

      }
      \n[\/code]<\/p>\n

      Step 2.5 Create Listener Class or consumer named RabbitMqMessageListener <\/b><\/span><\/p>\n

      This is the Listener class or Consumer whenever Producer sends a message to the queue this class will receive it and the only method having annotation @RabbitListener(queues = Constants.queueName) will receive the messages.<\/span><\/p>\n

      [code language=”java”]
      \n@EnableRabbit
      \n@Component
      \npublic class RabbitMqMessageListener {
      \n @RabbitListener(queues = Constants.queueName)
      \n public void processQueue(Map<String, Object> message) {
      \n EventHandler.handler((Events) message.get("event"), message.get("message"));
      \n }
      \n[\/code]<\/p>\n

      Step <\/b>2<\/b>.<\/b>6<\/b>: <\/b>This is an additional step which depends on the use case. In the preceding step, I have used view(index.jsp) in the controller. So now you have to create the view(index.jsp) in webapp\/WEB-INF\/views. <\/em><\/span><\/p>\n

      Finally use the command: mvn jetty:run<\/em> in your root directory of a project to run the project.<\/span><\/p>\n

      Hope this will Help. Thank You.<\/span><\/p>\n

      You can find the source code at following git repo’s <\/span><\/p>\n

      basic RabbitMq JMS integration<\/span><\/p>\n

      https:\/\/github.com\/amitraturi36\/RabbitMqSpringIntegration.git<\/span><\/a><\/p>\n

      spring specific AMQP Integration<\/span><\/p>\n

      https:\/\/github.com\/amitraturi36\/SpringRabbitMQIntegrationExample.git<\/span><\/a><\/p>\n

      Also See:<\/p>\n

      Sending Scheduled\/delayed messages with RabbitMQ through java client<\/a><\/p>\n

      Few Simple steps for integrating Rabbit MQ with Grails\u00a0<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"

      In this blog, we will see two different implementations of RabbitMQ, but before going to the implementation part let’s take a brief intro of some prerequisites. JMS The Java Message Service (JMS) is a Message Oriented Middleware Java API that supports the formal communication between software components. It allows applications to create, send, receive, and […]<\/p>\n","protected":false},"author":1032,"featured_media":0,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"iawp_total_views":15},"categories":[7,446,1],"tags":[1077,4383,4382],"aioseo_notices":[],"_links":{"self":[{"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/posts\/44550"}],"collection":[{"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/users\/1032"}],"replies":[{"embeddable":true,"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/comments?post=44550"}],"version-history":[{"count":0,"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/posts\/44550\/revisions"}],"wp:attachment":[{"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/media?parent=44550"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/categories?post=44550"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.tothenew.com\/blog\/wp-json\/wp\/v2\/tags?post=44550"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}