当前位置：首页 > news >正文

SpringBoot 整合RabbitMQ 之延迟队列实验

news 来源：原创 2024/5/20 5:37:14

在Spring Boot中整合RabbitMQ并实现延迟队列的功能，可以按照以下步骤进行：

添加依赖：在pom.xml文件中添加RabbitMQ和Spring AMQP相关的依赖。

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置RabbitMQ连接信息：在application.properties或application.yml文件中配置RabbitMQ的连接信息，包括host、port、username、password等。

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

创建消息发送者：创建一个消息发送者类，用于发送消息到RabbitMQ。

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component
public class MessageSender {@Autowiredprivate AmqpTemplate amqpTemplate;public void sendMessage(String message, long delayTime) {amqpTemplate.convertAndSend("exchangeName", "routingKey", message, message -> {message.getMessageProperties().setDelay((int) delayTime);return message;});}
}

创建消息接收者：创建一个消息接收者类，用于监听RabbitMQ中的消息。

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class MessageReceiver {@RabbitListener(queues = "queueName")public void receiveMessage(String message) {System.out.println("Received message: " + message);}
}

创建延迟队列配置类：创建一个延迟队列的配置类，用于声明交换机、队列和绑定关系。

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class DelayedQueueConfig {@Beanpublic Queue queue() {return new Queue("queueName");}@Beanpublic CustomExchange exchange() {Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");return new CustomExchange("exchangeName", "x-delayed-message", true, false, args);}@Beanpublic Binding binding(Queue queue, CustomExchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("routingKey").noargs();}
}