Comment SpringBoot intègre Redisson pour implémenter la file d'attente différée-javaDidacticiel-php.cn

Comment SpringBoot intègre Redisson pour implémenter la file d'attente différée

王林

Libérer： 2023-05-30 14:40:14

avant

1718 Les gens l'ont consulté

Scénarios d'utilisation

1. La commande a été passée avec succès et aucun paiement n'a été effectué dans les 30 minutes. Délai de paiement & #xff0c; La commande sera automatiquement annulée

2. Signature de la commande & #xff0c; Aucune évaluation ne sera effectuée dans les 7 jours suivant la signature. Commande expirée et non évaluée ; Le système affiche par défaut des avis positifs

3. Commande passée avec succès ; Le commerçant n'a pas accepté la commande pendant 5 minutes ， Commande annulée

4. La livraison a expiré ，Push. Rappel SMS

… …

Pour les scénarios avec de longs délais et de faibles performances en temps réel, nous pouvons utiliser la planification des tâches pour traiter les interrogations à intervalles réguliers. Par exemple, xxl-job

Aujourd'hui, nous utilisons une méthode relativement simple et légère : utiliser la file d'attente différée de Redis pour le traitement. Bien sûr, il existe de meilleures solutions : vous pouvez choisir la solution optimale en fonction de la sélection technologique et du système commercial de l'entreprise. Par exemple, en utilisant le middleware de messages Kafka et la file d'attente de retard de RabbitMQ

Je ne discuterai pas de son principe d'implémentation pour l'instant. Passons directement au code réel pour implémenter la file d'attente de retard basée sur Redis

1. Introduisons la dépendance Redisson

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.10.5</version>
</dependency>

Copier après la connexion

. 2. Configuration de Nacos Connexion Redis

spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password: 123456
    database: 12
    timeout: 3000

Copier après la connexion

3. Créer une configuration RedissonConfig

/**
 * Created by LPB on 2020/04/20.
 */
@Configuration
public class RedissonConfig {

	@Value("${spring.redis.host}")
	private String host;
	@Value("${spring.redis.port}")
	private int port;
	@Value("${spring.redis.database}")
	private int database;
	@Value("${spring.redis.password}")
	private String password;

	@Bean
	public RedissonClient redissonClient() {
		Config config = new Config();
		config.useSingleServer()
			.setAddress("redis://" + host + ":" + port)
			.setDatabase(database)
			.setPassword(password);
		return Redisson.create(config);
	}

}

Copier après la connexion

4. Encapsuler la classe d'outil de file d'attente de retard

/**
 * redis延迟队列工具
 * Created by LPB on 2021/04/20.
 */
@Slf4j
@Component
public class RedisDelayQueueUtil {

    @Autowired
	private RedissonClient redissonClient;

    /**
     * 添加延迟队列
     * @param value 队列值
     * @param delay 延迟时间
     * @param timeUnit 时间单位
     * @param queueCode 队列键
     * @param <T>
     */
    public <T> void addDelayQueue(T value, long delay, TimeUnit timeUnit, String queueCode){
        try {
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            delayedQueue.offer(value, delay, timeUnit);
			log.info("(添加延时队列成功) 队列键&#xff1a;{}&#xff0c;队列值&#xff1a;{}&#xff0c;延迟时间&#xff1a;{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
        } catch (Exception e) {
            log.error("(添加延时队列失败) {}", e.getMessage());
            throw new RuntimeException("(添加延时队列失败)");
        }
    }

	/**
	 * 获取延迟队列
	 * @param queueCode
	 * @param <T>
	 * @return
	 * @throws InterruptedException
	 */
    public <T> T getDelayQueue(String queueCode) throws InterruptedException {
        RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        T value  = (T) blockingDeque.take();
        return value;
	}
}

Copier après la connexion

5. Créer une énumération d'entreprise de file d'attente de retard

/**
 * 延迟队列业务枚举
 * Created by LPB on 2021/04/20.
 */
@Getter
@NoArgsConstructor
@AllArgsConstructor
public enum RedisDelayQueueEnum {

	ORDER_PAYMENT_TIMEOUT("ORDER_PAYMENT_TIMEOUT","订单支付超时&#xff0c;自动取消订单", "orderPaymentTimeout"),
	ORDER_TIMEOUT_NOT_EVALUATED("ORDER_TIMEOUT_NOT_EVALUATED", "订单超时未评价&#xff0c;系统默认好评", "orderTimeoutNotEvaluated");

	/**
	 * 延迟队列 Redis Key
	 */
	private String code;

	/**
	 * 中文描述
	 */
	private String name;

	/**
	 * 延迟队列具体业务实现的 Bean
	 * 可通过 Spring 的上下文获取
	 */
	private String beanId;

}

Copier après la connexion

6. Énumérez les beans définis et implémentez l'exécuteur de file d'attente de retard

/**
 * 延迟队列执行器
 * Created by LPB on 2021/04/20.
 */
public interface RedisDelayQueueHandle<T> {

	void execute(T t);

}

Copier après la connexion

/**
 * 订单支付超时处理类
 * Created by LPB on 2021/04/20.
 */
@Component
@Slf4j
public class OrderPaymentTimeout implements RedisDelayQueueHandle<Map> {
	@Override
	public void execute(Map map) {
		log.info("(收到订单支付超时延迟消息) {}", map);
		// TODO 订单支付超时&#xff0c;自动取消订单处理业务...

	}
}

Copier après la connexion

/**
 * 订单超时未评价处理类
 * Created by LPB on 2021/04/20.
 */
@Component
@Slf4j
public class OrderTimeoutNotEvaluated implements RedisDelayQueueHandle<Map> {
	@Override
	public void execute(Map map) {
		log.info("(收到订单超时未评价延迟消息) {}", map);
		// TODO 订单超时未评价&#xff0c;系统默认好评处理业务...

	}
}

Copier après la connexion

Les étapes ci-dessus，Le code principal de la file d'attente de retard Redis a été terminé，Ce qui suit Écrivons une interface de test ， utilisez PostMan pour la simuler et la tester

9. Créez une interface de test ， simulez l'ajout d'une file d'attente de retard

/**
 * 启动延迟队列
 * Created by LPB on 2021/04/20.
 */
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {

	@Autowired
	private RedisDelayQueueUtil redisDelayQueueUtil;

	@Override
	public void run(String... args) {
		new Thread(() -> {
			while (true){
				try {
					RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
					for (RedisDelayQueueEnum queueEnum : queueEnums) {
						Object value = redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());
						if (value != null) {
							RedisDelayQueueHandle redisDelayQueueHandle = SpringUtil.getBean(queueEnum.getBeanId());
							redisDelayQueueHandle.execute(value);
						}
					}
				} catch (InterruptedException e) {
					log.error("(Redis延迟队列异常中断) {}", e.getMessage());
				}
			}
		}).start();
		log.info("(Redis延迟队列启动成功)");
	}
}

Copier après la connexion