皇冠体育寻求亚洲战略合作伙伴,皇冠代理招募中,皇冠平台开放会员注册、充值、提现、电脑版下载、APP下载。

首页科技正文

RabbitMQ-延迟行列

admin2021-11-2061

usdt支付接口

菜宝钱包(www.caibao.it)是使用TRC-20协议的Usdt第三方支付平台,Usdt收款平台、Usdt自动充提平台、usdt跑分平台。免费提供入金通道、Usdt钱包支付接口、Usdt自动充值接口、Usdt无需实名寄售回收。菜宝Usdt钱包一键生成Usdt钱包、一键调用API接口、一键无实名出售Usdt。

1. 简介

我们在上一篇博文中遗留了一个小问题,就是虽然TTL + DLX能实现延迟行列的功效,然则有两个问题。

首先营业场景为:好比海底捞预约,每小我私人预约的时间段纷歧致,有个可能一个小时后,有的可能三个小时等,当快到预约时间点需要给用户举行短信通知。

  1. 通过给Queue设置过时时间的方式不现实,由于很有可能每条纪录的过时时间都纷歧样,不能能设置那么多的Queue
  2. 直接给Message设置过时时间,这种方式也欠好,由于这种方式是当该新闻在行列头部时(消费时),才会单独判断这一新闻是否过时。例:现在有两条新闻,第一条新闻过时时间为30s,而第二条新闻过时时间为15s,当过了15秒后,第二条新闻不会立刻过时,而是要品级一条新闻被消费后,第二条新闻被消费时,才会判断是否过时,也就是等到第二条新闻投往DLX已经已往45s了。

这也就抛出了本章主题:延迟行列

RabbitMQ默认没有提供延迟行列功效,而是要通过插件提供的x-delayed-message(延迟交流机)来实现。

延迟行列:用户可以使用该类型声明一个交流,x-delayed-message然后使用自界说标头公布新闻,x-delay以毫秒为单元示意新闻的延迟时间。新闻将在x-delay毫秒后通报到响应的行列。

2. 安装插件

官方插件地址:https://www.rabbitmq.com/community-plugins.html

找到插件rabbitmq_delayed_message_exchange,进入GitHub下载内陆RabbitMQ对应的插件版本(下载.ez文件)。

我这里下载的是3.8.9版本,如图:

下载到内陆后将文件放置RabbitMQ的plugins目录。

我这里内陆是使用docker-compose安装的服务,imagerabbitmq:3.8.3-management(虽然版本没对起来,然则测试能用,然则使用3.9的版本会报错,插件安装失败)安装的服务,操作步骤如下:

新2代理网址

新2代理网址(www.22223388.com)实时更新发布最新最快最有效的新2网址和新2最新网址,包括新2手机网址,新2备用网址,皇冠最新网址,新2足球网址,新2网址大全。

  1. 将下载好的文件放置RabbitMQ插件目录

    rabbitmq:容器服务名

    $ docker cp /Users/ludangxin/Downloads/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez rabbitmq:/opt/rabbitmq/plugins/
    
  2. 进入容器

    $ docker exec -it rabbitmq /bin/bash
    
  3. 查看现有的插件列表

    $ rabbitmq-plugins list
    # 输出部门内容如下 [E*] = 明确启用; e = 隐式启用
    [  ] rabbitmq_amqp1_0                  3.8.3
    [  ] rabbitmq_auth_backend_cache       3.8.3
    [  ] rabbitmq_auth_backend_http        3.8.3
    [  ] rabbitmq_auth_backend_ldap        3.8.3
    [  ] rabbitmq_auth_backend_oauth2      3.8.3
    [  ] rabbitmq_auth_mechani *** _ssl       3.8.3
    [  ] rabbitmq_consistent_hash_exchange 3.8.3
    [  ] rabbitmq_event_exchange           3.8.3
    [  ] rabbitmq_federation               3.8.3
    [  ] rabbitmq_federation_management    3.8.3
    [  ] rabbitmq_jms_topic_exchange       3.8.3
    [E*] rabbitmq_management               3.8.3
    [e*] rabbitmq_management_agent         3.8.3
    [  ] rabbitmq_mqtt                     3.8.3
    
  4. 启用插件

    $ rabbitmq-plugins enable rabbitmq_delayed_message_exchange
    
  5. 再次查看安装列表就有了rabbitmq_delayed_message_exchange

安装完毕后上岸RabbitMQ控制台查看,会发现多了个x-delayed-message类型的Exchange。

3. 实现延迟行列

3.1 引入所需依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</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.amqp</groupId>
    <artifactId>spring-rabbit-test</artifactId>
    <scope>test</scope>
</dependency>

3.2 application.yaml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    # rabbit 默认的虚拟主机
    virtual-host: /
    # rabbit 用户名密码
    username: admin
    password: admin123

3.3 RabbitConfig

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.amqp.core.QueueBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

/**
 * 延迟行列设置
 *
 * @author ludangxin
 * @date 2021/9/16
 */
@Configuration
public class RabbitDelayedConfig {
    public static final String QUEUE_NAME_DELAYED = "DELAY.QUEUE";
    public static final String EXCHANGE_NAME_DELAYED = "DELAY.EXCHANGE";
    public static final String ROUTING_KEY_DELAYED = "DELAY.#";

    @Bean(QUEUE_NAME_DELAYED)
    public Queue queue() {
       return QueueBuilder.durable(QUEUE_NAME_DELAYED).build();
    }

    @Bean(EXCHANGE_NAME_DELAYED)
    public CustomExchange exchange() {
       Map<String, Object> arguments = new HashMap<>(1);
       // 在这里声明一个主题类型的延迟行列,固然其他类型的也可以。
       arguments.put("x-delayed-type", "topic");
       return new CustomExchange(EXCHANGE_NAME_DELAYED, "x-delayed-message", true, false, arguments);
    }

    @Bean
    public Binding bindingNotify(@Qualifier(QUEUE_NAME_DELAYED) Queue queue, @Qualifier(EXCHANGE_NAME_DELAYED) CustomExchange customExchange) {
       return BindingBuilder.bind(queue).to(customExchange).with(ROUTING_KEY_DELAYED).noargs();
    }
}

3.4 Producer

import com.ldx.rabbitmq.config.RabbitDelayedConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * 延迟新闻生产者
 *
 * @author ludangxin
 * @date 2021/9/9
 */
@Component
public class DelayProducer {

   @Autowired
   private RabbitTemplate rabbitTemplate;

   public void sendDelayedMsg(String msg, Integer delay) {
      MessageProperties mp = new MessageProperties();
      // 设置过时时间
      mp.setDelay(delay);
      Message message = new Message(msg.getBytes(), mp);
      rabbitTemplate.convertAndSend(RabbitDelayedConfig.EXCHANGE_NAME_DELAYED, "DELAY.MSG", message);
   }
}

3.5 Consumer

import com.ldx.rabbitmq.config.RabbitDelayedConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * 延迟新闻消费者
 *
 * @author ludangxin
 * @date 2021/9/9
 */
@Slf4j
@Component
public class DelayConsumer {

    @RabbitListener(queues = {RabbitDelayedConfig.QUEUE_NAME_DELAYED})
    public void delayQueue(Message message){
        log.info(new String(message.getBody()) + ",竣事时间为:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
    }

}

3.6 测试代码

@Autowired
private DelayProducer delayProducer;

@Test
@SneakyThrows
public void sendDelayedMsg() {
   for(int i = 16; i >= 10; i --) {
      String msg = "我将在" + i + "s后过时,最先时间为:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
      delayProducer.sendDelayedMsg(msg,i * 1000);
   }
   // 使历程壅闭,利便Consumer监听输出Message
   System.in.read();
}

3.7 启动测试

启动测试代码,输出内容如下:

从日志内容可以看出,新闻存活了30s,相符预期。

2021-09-16 23:40:10.806  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在10s后过时,最先时间为:2021-09-16 23:40:00,竣事时间为:2021-09-16 23:40:10
2021-09-16 23:40:11.792  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在11s后过时,最先时间为:2021-09-16 23:40:00,竣事时间为:2021-09-16 23:40:11
2021-09-16 23:40:12.791  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在12s后过时,最先时间为:2021-09-16 23:40:00,竣事时间为:2021-09-16 23:40:12
2021-09-16 23:40:13.791  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在13s后过时,最先时间为:2021-09-16 23:40:00,竣事时间为:2021-09-16 23:40:13
2021-09-16 23:40:14.788  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在14s后过时,最先时间为:2021-09-16 23:40:00,竣事时间为:2021-09-16 23:40:14
2021-09-16 23:40:15.785  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在15s后过时,最先时间为:2021-09-16 23:40:00,竣事时间为:2021-09-16 23:40:15
2021-09-16 23:40:16.785  INFO 7883 --- [ntContainer#0-1] c.ldx.rabbitmq.consumer.Delay2Consumer   : 我将在16s后过时,最先时间为:2021-09-16 23:40:00,竣事时间为:2021-09-16 23:40:16

澳洲幸运5开户www.a55555.net)是澳洲幸运5彩票官方网站,开放澳洲幸运5彩票会员开户、澳洲幸运5彩票代理开户、澳洲幸运5彩票线上投注、澳洲幸运5实时开奖等服务的平台。

网友评论

1条评论