【NestJs】记录使用redis订阅自动取消订单

Hello，大家好，我是致力于NestJs后端开发的开发人员大山，目前在用NestJs为后端开发一套外贸商城系统，今天来记录下如何使用redis订阅自动取消订单

前言：在业务实现的过程中，自动取消订单这个业务不仅可以redis订阅取消订单，也可以使用redis延迟队列、定时任务等等解决方案去完成这个业务需求。

使用redis订阅自动取消订单

我们在使用各大购物平台的时候，前台有一些是等待付款的订单，超过10分钟或者30分钟便会自动取消订单，从而释放商品库存加入数据库订单表里，前后台订单状态成为取消支付/取消订单，后台再修改订单数据统计。

那么接下来就用redis订阅来实现自动取消订单吧

1、在项目配置文件里加上过期时间和开启redis消息订阅配置

在redis的配置文件里，找到notify-keyspace-events，没有则加上，我这里是宝塔的redis，修改完毕重载redis配置，再重启redis即可

notify-keyspace-events: Ex

关于notify-keyspace-events的触发条件命令如下（可组合），我项目里使用Ex即可。

K：键空间通知
E：键事件通知
g：一般命令通知
$：字符串命令通知
l：列表命令通知
s：集合命令通知
h：哈希命令通知
z：有序集合命令通知
x：过期事件通知
e：驱逐事件通知
A：参数的通用配置

那么配置好相关的，就可以写业务代码了，我这里是生成订单加入数据库后再存入redis，你也可以生成临时订单键存入redis，等过期的时候再加入数据库减去库存

2、业务代码

/** 创建用户订单 */
async createUserOrder(token: string, createOrderDto: CreateClienOrderDto) : Promise<ResultData> {
    // 一堆比对商品金额的业务代码，这里就不展示了。。。。
    const result = await this.manager.transaction(async (transactionalEntityManager) => {
      return await transactionalEntityManager.save<OrderEntity>(createOrderDto)
    })
    
    // 插入订单商品关联表
    if (productList.length > 0) {
      await this.orderProductService.createOrUpdateOrderProduct({ orderId: result.id, productList })
    }
    
    // 在redis创建订单数据，并在30分钟后过期，ClientRedisKeyPrefix.USER_INFO_ORDER是自定义redis名和key
    const orderRedisKey = getClientRedisKey(ClientRedisKeyPrefix.USER_INFO_ORDER, result.id);
    const orderData = instanceToPlain(result)
    // 存入redis
    await this.redisService.hmset(
      orderRedisKey,
      orderData,
      ms(this.config.get<string>('jwt.orderExpiresin')) / 1000,
    )
    
    // redis的订单数据过期后，自动取消订单
    const handleKeyExpiration = async (key: string) => {
      const orderId = key.split(':').pop();
      console.log(orderId, '自动取消的订单id');
      // 过期回调事件，取消订单
      await this.updateOrderStatus(orderId, 'cancel_payment')
    };
    // 订阅消息
    await this.redisService.listenForKeyExpiration(ClientRedisKeyPrefix.USER_INFO_ORDER, handleKeyExpiration)
    // ResultData 是个人封装的返回结果，你可以自己定义。
    return ResultData.ok();
}

/** 更新订单状态 */
  async updateOrderStatus(orderId: string, orderStatus:string): Promise<boolean> {
    /** 查询订单是否存在 */
    const existing = await this.orderRepo.findOne({ where: { id: orderId } })
    if (!existing) ResultData.fail(AppHttpCode.USER_NOT_FOUND, 'The current order does not exist')

    /** 更新订单状态 */
    const { affected } = await this.manager.transaction(async (transactionalEntityManager) => {
      return await transactionalEntityManager.update<OrderEntity>(OrderEntity, orderId, { id: orderId, orderStatus })
    })
    if (!affected) return false
    return true
  }

3、方法和配置

getClientRedisKey，在一个方法文件里去写getClientRedisKey方法

export function getClientRedisKey(moduleKeyPrefix: ClientRedisKeyPrefix, id: string | number): string {
  return `${moduleKeyPrefix}${id}`
}

ClientRedisKeyPrefix，同样，在一个配置文件里去定义好

export enum ClientRedisKeyPrefix {
  USER_INFO = 'client:user:info:', //客户端用户登陆后用户信息的键名前缀
  USER_INFO_EMAIL = 'client:user:info:email:', //客户端用户注册/修改密码发送邮箱验证码的键名前缀
  USER_INFO_CART = 'client:user:info:cart:', //客户端用户登陆后购物车数量的键名前缀
  USER_INFO_ORDER = 'client:user:info:order:'， //客户端用户登陆后创建的键名前缀
}

4、接下来再去写redis的消息订阅

redisService.listenForKeyExpiration的方法

import { InjectRedis } from '@liaoliaots/nestjs-redis'
import { Injectable } from '@nestjs/common'
import Redis from 'ioredis'

@Injectable()
export class RedisService {
  constructor(@InjectRedis() private readonly client: Redis) {}

  getClient(): Redis {
    return this.client
  }

  /**
   * 监听键过期事件
   * @param callbackKey 回调key
   * @param callback 键过期时执行的回调函数
   */
  listenForKeyExpiration(callbackKey: string, callback: (key: string) => void): void {
    const pubsub = this.client.duplicate();
    console.log(pubsub, 'redis请求订阅')
    pubsub.psubscribe('__keyevent@0__:expired', (err) => {
      console.log(err, 'redis订阅情况，err返回null是订阅成功')
      if (err) throw err;
    });
    pubsub.on('pmessage', (pattern, channel, message) => {
      const key = message.toString();
      // 只有需要返回redis的key才会回调
      if (key.includes(callbackKey)) {
        callback(key);
      }
    });
    // 添加错误处理
    pubsub.on('error', (err) => {
      console.error('Error in pubsub:', err);
    });
    // 添加结束处理
    pubsub.on('end', () => {
      console.log('Pubsub ended');
    });
  }
}

接下来我们来测试下，先去下个单，然后在redis查看，存在就是对的。

等待redis时间过期后，由订阅回调事件触发，再去更改订单状态。

下次记录下创建ip黑名单和用ip黑名单缓存去屏蔽ip，本来想今天一篇文章写完，但是我的懒病又犯了。。。下次写吧

由于本人外贸商城系统不是开源的，有需要的朋友可联系我购买，含有大量的NestJs相关业务代码，便宜实惠。

本人开源的NestJs项目，欢迎各位朋友点个start！谢谢！

git链接：gitee.com/wx375149069…

2024年12月13日修改redis代码如下：

import { InjectRedis } from '@liaoliaots/nestjs-redis'
import { Injectable } from '@nestjs/common'
import Redis from 'ioredis'
@Injectable() export class RedisService { 
    constructor(@InjectRedis() private readonly client: Redis){}
    
    getClient(): Redis { return this.client }
    
    /** * 监听键过期事件 * 
    @param callbackKey 回调key * 
    @param callback 键过期时执行的回调函数 
    */
    listenForKeyExpiration(callbackKey: string, callback: (key: string) => void): void {
    const pubsub = this.client.duplicate();
    const subscribePattern = '__keyevent@0__:expired'
    let reconnectTimeout: NodeJS.Timeout | null = null;
    const reconnectAndSubscribe = () => {
      if (reconnectTimeout) {
        clearTimeout(reconnectTimeout);
        reconnectTimeout = null;
      }
      // console.log(pubsub, 'redis请求订阅')
      pubsub.psubscribe(subscribePattern, (err) => {
        console.log(`redis订阅情况：${err ? err : '成功'}`)
        if (err) throw err;
      });
      pubsub.on('pmessage', (pattern, channel, message) => {
        console.log(message, '监听到过期事件')
        const key = message.toString();
        // 只有需要返回redis的key才会回调
        if (key.includes(callbackKey)) {
          callback(key);
        }
      });
      // 添加错误处理
      pubsub.on('error', (err) => {
        console.error('Redis 订阅错误信息:', err);
        reconnectTimeout = setTimeout(reconnectAndSubscribe, 5000); // 5秒后重试
      });
      // 添加结束处理
      pubsub.on('end', (ee) => {
        console.log('Redis 订阅结束信息：', ee);
        reconnectTimeout = setTimeout(reconnectAndSubscribe, 5000); // 5秒后重试
      });
    }
    reconnectAndSubscribe()

    // 监听客户端的连接断开事件
    this.client.on('error', (err) => {
      console.error('Redis client error:', err);
      reconnectAndSubscribe();
    });

    this.client.on('end', () => {
      console.log('Redis client connection ended');
      reconnectAndSubscribe();
    });

    this.client.on('reconnecting', () => {
      console.log('Redis client is reconnecting');
      reconnectAndSubscribe();
    });

    this.client.on('ready', () => {
      console.log('Redis client is ready');
    });
  }