14. Springboot integrates RabbitMQ

Table of contents

1 Introduction

What is RabbitMQ?
Install RabbitMQ
Springboot integrates RabbitMQ

4.1. Add dependencies

4.2. Add configuration

4.3. Add controller as producer

4.4. Set producer message confirmation CallBack

4.5. Add Consumer as a consumer

4.6. Start the program and access

1 Introduction

Message Queue (MQ for short) is an asynchronous messaging middleware that decouples communication between applications. Applications can send messages to queues without knowing who will receive them. Receiving applications can retrieve messages from the queue without knowing who sent them. Message queue is an important middleware that helps asynchronous, reliable, and scalable communication between applications. Common message queue middleware include ActiveMQ, RabbitMQ, Kafka… Today we will introduce RabbitMQ.

2. What is RabbitMQ?

RabbitMQ is an open source message queue server that implements the AMQP (Advanced Message Queuing Protocol) standard. AMQP is an application layer protocol designed for message-oriented middleware. Clients and message middleware based on this protocol can transmit messages and are not restricted by products, development languages, etc.

AMQP: Advanced Message Queue, Advanced Message Queuing Protocol. It is an open standard for application layer protocols and is designed for message-oriented middleware. Clients and message middleware based on this protocol can transmit messages and are not restricted by products, development languages, etc.

Key features of RabbitMQ include:

High performance: RabbitMQ is capable of handling large volumes of messages and provides low-latency performance.
Reliability: RabbitMQ provides persistent message storage to ensure that messages will not be lost.
Scalability: RabbitMQ can easily scale to meet growing demand.
Flexibility: RabbitMQ supports multiple programming languages and clients and provides rich functionality and configuration options.

Common application scenarios of RabbitMQ include:

Distributed systems: RabbitMQ can be used for asynchronous communication in distributed systems.
Asynchronous processing: RabbitMQ can be used to process tasks asynchronously to improve system performance and efficiency.
Message queue: RabbitMQ can be used to implement message queues, such as task queues, publish/subscribe queues, etc.
Message notifications: RabbitMQ can be used to send message notifications, such as emails or SMS messages.

3. Install RabbitMQ

RabbitMQ is an open source implementation of AMQP developed in Erlang language. Therefore, you need to install the Erlang environment before installing RabbitMQ.

Erlang download address: Downloads – Erlang/OTP

RabbitMQ download address: Installing RabbitMQ | RabbitMQ

Install Erlang first, and then install RabbitMQ. The installation project is relatively simple, and the next step is a no-brainer.

After installing RabbitMQ, open the cmd window and enter the sbin of the RabbitMQ installation directory. My directory is:

D:\RabbitMQ Server\rabbitmq_server-3.13.0\sbin

Then enter the following command to install the plug-in:

rabbitmq-plugins enable rabbitmq_management

The following prompt indicates that the installation was successful.

To verify whether RabbitMQ is installed successfully, enter the following command:

rabbitmqctl status

At this time, you can see the RabbitMQ management page by directly accessing http://127.0.0.1:15672. The default port of RabbitMQ is 15672, and the default management page account and password are guest.

After logging in, you can see an initial management interface:

4. Springboot integrates RabbitMQ

4.1. Add dependencies

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-parent</artifactId>
       <version>3.2.3</version>
       <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>springboot-rabbitmq</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-rabbitmq</name>
    <description>springboot-rabbitmq</description>
    <properties>
       <java.version>17</java.version>
    </properties>
    <dependencies>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter</artifactId>
       </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-devtools</artifactId>
          <scope>runtime</scope>
          <optional>true</optional>
       </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-configuration-processor</artifactId>
          <optional>true</optional>
       </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-test</artifactId>
          <scope>test</scope>
       </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-web</artifactId>
       </dependency>
       <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-amqp</artifactId>
       </dependency>
       <dependency>
          <groupId>cn.hutool</groupId>
          <artifactId>hutool-all</artifactId>
          <version>5.8.24</version>
       </dependency>
    </dependencies>

    <build>
       <plugins>
          <plugin>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-maven-plugin</artifactId>
          </plugin>
       </plugins>
    </build>
</project>

4.2. Add configuration

# rabbitmqConnection configuration information
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

# Ensure messages are returned if not received by the queue
spring.rabbitmq.publisher-returns=true
# After the message is successfully published to the exchange, the callback method will be triggered.
spring.rabbitmq.publisher-confirm-type=correlated

4.3. Add controller as producer

Create a new controller for sending messages.

package com.example.springbootrabbitmq.controller;

import com.example.springbootrabbitmq.config.MqProducerCallBack;
import jakarta.annotation.Resource;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("push/message")
public class PushMessageController {

    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private MqProducerCallBack mqProducerCallBack;


    @GetMapping("test")
    public String sendMessage() {
        // correlationData：There is only one inside the object id Attributes，Used to indicate the uniqueness of the current message。
        CorrelationData correlationData = new CorrelationData("id_" + System.currentTimeMillis());
        // Message confirmation and return callback
        rabbitTemplate.setConfirmCallback(mqProducerCallBack);
        rabbitTemplate.setReturnsCallback(mqProducerCallBack);
        // message sending
        rabbitTemplate.convertAndSend("my-queue", "hello world", message -> {
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message;
        }, correlationData);
        return "publisher success...";
    }
}

4.4. Set producer message confirmation CallBack

package com.example.springbootrabbitmq.config;

import cn.hutool.json.JSONUtil;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

@Component
public class MqProducerCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    /**
     * correlationData：There is only one inside the object id Attributes，Used to indicate the uniqueness of the current message。
     * ack：Message delivered tobroker status，truesuccess，falsefail。
     * cause：Reason for delivery failure。
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            System.err.println("informationID=" + correlationData.getId() + "Delivery failed，Reason for failure：" + cause);
        } else {
            System.out.println("Message delivery received confirmation，correlationData=" + correlationData.getId());
        }
    }

    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        System.out.println("Return message result：" + JSONUtil.toJsonStr(returnedMessage));
    }

}

4.5. Add Consumer as a consumer

package com.example.springbootrabbitmq.consumer;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
public class PushMessageConsumer {


    /**
     * basicAck：Indicates successful confirmation，After using this receipt method，The message will berabbitmq broker delete。
     * void basicAck(long deliveryTag, boolean multiple)
     * deliveryTag：Represents the message delivery sequence number，Every time a message is consumed or the message is re-delivered,，deliveryTagwill increase。In manual message confirmation mode，We can specifydeliveryTagnews proceedingack、nack、rejectWait for operations。
     * multiple：Whether to confirm in batches，The value is true will be a one-time ackAll messages less than the current deliveryTag news。
     * */
    @RabbitListener(queuesToDeclare = @Queue(value = "my-queue"))
    @RabbitHandler
    public void consume(String msg, Channel channel, Message message) throws IOException {
        try {
            System.out.println("Consumer receives message：" + msg);

            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("deliveryTag：" + message.getMessageProperties().getDeliveryTag());
            System.out.println("redelivered：" + message.getMessageProperties().getRedelivered());
        } catch (Exception e) {
            if (message.getMessageProperties().getRedelivered()) {
                System.err.println("The message has been repeated and failed to be processed,Decline to receive again！");
                /**
                 * reject message，requeue=false Indicates that he will not rejoin the team，If a dead letter queue is configured, enter the dead letter queue.
                 * basicReject：reject message，andbasicNackThe difference is that batch operations cannot be performed，Other usages are very similar。
                 * deliveryTag：Represents the message delivery sequence number。
                 * requeue：The value is true The message will be requeued。
                 */
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                System.out.println("The message is about to be returned to the queue again for processing！");
                /**
                 * requeueWhether to return to the queue，trueRe-enlist
                 * deliveryTag：Represents the message delivery sequence number。
                 * multiple：Whether to confirm in batches。
                 * requeue：The value is true The message will be requeued。
                 */
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
            }
        }
    }

}