Rabbit MQ Guide
Messaging using RabbitMQ
Definition
- Open-source multi-protocol supported messaged broker
- Written in Erlang language
- Protocols supported : AMQP(Started from here), MQTT, and STOMP
Points to Note
- Ready messages : Waiting to be consumed by Consumer
- Load Balancing mechanism : Round robin Scheduling
- Message types supported : String, JSON, XML, Serializable Object, etc.
- Ways of Purging Messages
- UI : Purge Message
- CMD : rabbitmqctl.bat purge_queue <Queue-name>
- There is no max Queue length but is configurable. Refer Queue Length Page
- Maximum message size is 512MiB from Version 3.8 onwards. Refer Github Page for entire properties.
Best Practices
Refer below links for the same.
Downloads & Setup
- RabbitMQ Download
- ERLang OPT Download
- Set environment variables (ERLANG_HOME : C/PrgFiles/erlang/, Path(Append at end) : %ERLANG_HOME%\bin)
- Enable RabbitMQ Management Plugin from CommandLine (rabbitmq_server-3.8.8\sbin>rabbitmq-plugins.bat enable rabbitmq_management)
- Start the Server : Double click rabbitmq-server.bat(Takes some time to start server)
- Visit the Localhost Installation of RabbitMQ at http://localhost:15672 (Default Credentials : guest/guest)
Message Flow
- Producer Application writes to Exchange
- Exchange has Conditions of Message Routing
- Exchange routes messages to appropriate Queues
- Consumer Applications listen to the appropriate Quees
Exchange Types(Based on Binding with Queue)
- Direct - queue.key = exchange.binding.key
- Fanout - Exchange publishes to all bound queues(No Key requirement)
- Topic - queue.key=Partial key of Exchange.binding. Eg.(Exchange Keys - pen.pencil.eraser)=Queues(.pen.,.pencil.,#.eraser)[*-Exactly one word, #->Any number of words]
- Headers - Headers of Exchange = Header of Queue(any/all)
Dependencies
- RabbitMQ Java client library : Allows Java applications to interface with RabbitMQ.
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
- RabbitMQ Dependency for AMQP
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
</dependency>
Publishing Messages to Queue
public class Publisher {
public static void main(String[] args) {
ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// Default Exchange type is Direct
String exchangeType = "";
String message = "Hello Anupama!!!";
// channel.basicPublish(<exchange-name>, <routing-key>,<basic-prop(Eg.Headers)>, <message-body>)
switch (exchangeType) {
case "topic":
channel.basicPublish("Topic-Exchange", "pen.pencil.eraser", null, message.getBytes());
break;
case "fanout":
channel.basicPublish("Fanout-Exchange", "", null, message.getBytes());
break;
case "header":
Map<String, Object> headersMap = new HashMap<String, Object>();
headersMap.put("item1", "pen");
headersMap.put("item2", "pencil");
headersMap.put("item2", "eraser");
BasicProperties br = new BasicProperties();
br = br.builder().headers(headersMap).build();
channel.basicPublish("Headers-Exchange", "", br, message.getBytes());
break;
default:
channel.basicPublish("Direct-Exchange", "pen", null, message.getBytes());
break;
}
channel.close();
connection.close();
}
}
Consuming Messages from Queue
public class Consumer {
public static void main(String[] args) {
ConnectionFactory factory = new ConnectionFactory();
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String (delivery.getBody());
};
//channel.basicConsume(<queue-name>, <autoAck-flag>, <deliverCallback> , <cancelCallback>)
channel.basicConsume("Queue-1", true, deliverCallback, consumerTag -> {});
}
}
Spring Boot Properties
- spring.rabbitmq.host=localhost
- spring.rabbitmq.port=5672
Publishing Serializable Custom Object to Queue from Controller
@RestController
@RequestMapping("/publish")
public class RabbitMQPublishingController {
@Autowired
RabbitTemplate rabbitTemplate;
@Autowired
Product product;
@GetMapping("/product/{name}")
public String publishProduct(@PathVariable("name") String name) {
product.setProductName(name);
//Simple Message Converter used here - Converts String,ByteArray & Serializable object only
rabbitTemplate.convertAndSend(<exchange-name>,<routing-key>, product);
return "Message Sent Successfully";
}
}
Consuming Serializable Custom Object from Queue from Service Layer
@Service
public class RabbitMQConsumerService {
@RabbitListener(queues = "Pens")
public void getMessage(Product product) {
//Business Logic using Product Object
}
}
Similarly, serialize the messages as per requirement and publish it. And then de-serialize at Consumer end accordingly. Will come up with a detailed separate post on Serialization concepts. That's it for now. Enjoy learning!! :pray: