SpringBoot使用RabbitMq实现队列和延时队列（）

　　本篇文章为你整理了SpringBoot使用RabbitMq实现队列和延时队列（）的详细内容，包含有 SpringBoot使用RabbitMq实现队列和延时队列，希望能帮助你了解 SpringBoot使用RabbitMq实现队列和延时队列。

　　闲来无事看了看RabbitMq的队列，总结了一些队列的实现方法，当然，免不了各种看别人的博客哈哈哈

　　其中延时队列有两种方式，一种是使用TTl+死信队列实现，一种是直接用RabbitMq的官方插件

　　第一种写起来比较麻烦，这里就只说第二种了

　　接下来从头开始说吧

　　1.首先是插件下载

　　打开链接https://github.com/rabbitmq/rabbitmq-routing-node-stamp/releases

　　然后下载图中的文件，放到你的安装目录的plugins文件夹下

　　然后在安装目录sbin下执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange即可

　　运行完直接关掉黑窗口就行

　　2.导入依赖

 dependency

　　 groupId org.springframework.amqp /groupId

　　 artifactId spring-rabbit /artifactId

　　 /dependency

　　3.在yml中配置

spring:

　　 rabbitmq:

　　 host: localhost

　　 port: 5672

　　 username: guest

　　 password: guest

　　 virtual-host: /

　　 # 消息确认（ACK）

　　 publisher-confirm-type: correlated #确认消息已发送到交换机(Exchange)

　　 publisher-returns: true #确认消息已发送到队列(Queue)

　　 listener:

　　 simple:

　　 #自动签收auto 手动 manual

　　 acknowledge-mode: auto

　　 retry:

　　 enabled: true #开启重试

　　 max-attempts: 3 #最大重试次数，默认3次，达到次数后，会进行消息移除。若绑定了死信队列，则会放入死信队列中

　　 initial-interval: 2000ms #重试间隔时间

　　4.在项目下创建一个rabbitMq目录，后续将他的配置文件放入此目录

　　创建接口文件

public interface RabbitConsts {

　　 * 普通队列

　　 String QUEUE_NAME = "gzl.queue"; //队列名称

　　 String EXCHANGE_NAME = "gzl.exchange"; //交换器名称

　　 String ROUTING_KEY = "gzl.outing"; //路由键

　　 * 延迟队列

　　 String DELAY_QUEUE = "delay.queue";

　　 * 延迟队列交换器

　　 String DELAY_MODE_QUEUE = "delay.mode";

　　}

　　创建MessageStruct消息体（传数据用）

@Data

　　@Builder

　　@NoArgsConstructor

　　@AllArgsConstructor

　　public class MessageStruct implements Serializable {

　　 private static final long serialVersionUID = 392365881428311040L;

　　 private String message;

　　}

　　创建RabbitMqConfig工具类

@Slf4j

　　@Configuration

　　public class RabbitMqConfig {

　　 @Bean

　　 public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {

　　 connectionFactory.setPublisherConfirms(true);

　　 connectionFactory.setPublisherReturns(true);

　　 RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);

　　 rabbitTemplate.setMandatory(true);

　　 rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause));

　　 rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) - log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message));

　　 return rabbitTemplate;

　　 * 队列

　　 @Bean

　　 public Queue queue()

　　 * 创建队列，参数说明：

　　 * String name：队列名称。

　　 * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。

　　 * 持久化的队列会存盘，在服务器重启的时候不会丢失相关信息。

　　 * boolean exclusive：设置是否排他，默认也是 false。为 true 则设置队列为排他。

　　 * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，

　　 * 当没有生产者或者消费者使用此队列，该队列会自动删除。

　　 * Map String, Object arguments：设置队列的其他一些参数。

　　 return new Queue(RabbitConsts.QUEUE_NAME, true, false, false, null);

　　 * Direct交换器

　　 @Bean

　　 public DirectExchange exchange()

　　 * 创建交换器，参数说明：

　　 * String name：交换器名称

　　 * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。

　　 * 持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息。

　　 * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，

　　 return new DirectExchange(RabbitConsts.EXCHANGE_NAME, true, false);

　　 * 绑定

　　 @Bean

　　 Binding binding(DirectExchange exchange, Queue queue)

　　 //将队列和交换机绑定, 并设置用于匹配键：routingKey

　　 return BindingBuilder.bind(queue).to(exchange).with(RabbitConsts.ROUTING_KEY);

　　
args.put("x-delayed-type", "direct");

　　 return new CustomExchange(RabbitConsts.DELAY_MODE_QUEUE, "x-delayed-message", true, false, args);

　　 * 延迟队列绑定自定义交换器

　　 * @param delayQueue 队列

　　 * @param delayExchange 延迟交换器

　　 @Bean

　　 public Binding delayBinding(Queue delayQueue, CustomExchange delayExchange) {

　　 return BindingBuilder.bind(delayQueue).to(delayExchange).with(RabbitConsts.DELAY_QUEUE).noargs();

　　//===================延迟队列插件实现end====================/

　　}

　　然后自己写个接口调用生产和消费就行了

@Autowired

　　private RabbitTemplate rabbitTemplate;

　　生产者发送消息

　　1.普通队列发送（无延时）

rabbitTemplate.convertAndSend(RabbitConsts.EXCHANGE_NAME,

　　 RabbitConsts.ROUTING_KEY, new MessageStruct("非延时！"));

　　2.延时队列发送

rabbitTemplate.convertAndSend(RabbitConsts.DELAY_MODE_QUEUE,

　　 RabbitConsts.DELAY_QUEUE, new MessageStruct("延时6s, " +

　　 new Date()), message - {

　　 message.getMessageProperties().setHeader("x-delay", 6000);

　　 return message;

　　);

　　消费者接收消息（无延时）

//开启队列获取

　　 @RabbitListener(queues = RabbitConsts.QUEUE_NAME)

　　 @RabbitHandler

　　 public void getInfo(MessageStruct messageStruct,Message message, Channel channel) throws IOException {

　　 long deliveryTag = message.getMessageProperties().getDeliveryTag();

　　 String msg = JSONObject.toJSONString(messageStruct.getMessage());

　　 Date date = new Date();

　　 SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

　　 System.out.println("时间："+df.format(date));

　　 System.out.println("接收到消息:" + msg);

　　// 消息确认如果yml中的配置为 acknowledge-mode: manual则需要手动确认消息，否则会造成消息堆积

　　// channel.basicNack(deliveryTag,false,false);

　　 }

　　消费者延时接收消息

//开启延时队列获取

　　 @RabbitListener(queues = RabbitConsts.DELAY_QUEUE)

　　 @RabbitHandler

　　 public void setListY(MessageStruct messageStruct,Message message, Channel channel) throws IOException {

　　 long deliveryTag = message.getMessageProperties().getDeliveryTag();

　　 String msg = JSONObject.toJSONString(messageStruct.getMessage());

　　 Date date = new Date();

　　 SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

　　 System.out.println("时间："+df.format(date));

　　 System.out.println("延时接收到消息:" + msg);

　　// 消息确认如果yml中的配置为 acknowledge-mode: manual则需要手动确认消息，否则会造成消息堆积

　　// channel.basicNack(deliveryTag,false,false);

　　 }

　　最后控制台输出

　　以上就是SpringBoot使用RabbitMq实现队列和延时队列（）的详细内容，想要了解更多 SpringBoot使用RabbitMq实现队列和延时队列的内容，请持续关注盛行IT软件开发工作室。

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。

相关文章阅读