NestJS小28-NestJS的互斥实现

by 雪隐 from https://juejin.cn/user/1433418895994094
本文欢迎分享与聚合，全文转载就不必了，尊重版权，圈子就这么大，若急用可联系授权

原文链接

在您正在读这篇文章的时候，我假设您已经有了多线程和异步进程的知识。我将不对这些进行详细的解释。网上有很多资料，我相信会有比我自己更好的解释。

今天，我准备向您们介绍利用async-mutex库在NestJS中使用互斥锁。去看一下这个库因为它会给您一个关于互斥的很好的解释。下面是它的概述。

术语“互斥”通常指的是用于同步运行在不同线程上的并发进程的数据结构。

也许您会想这东西有什么用。我过去常常也有同样的想法，但是马上，我找到了他的使用方式。在我的情况中，我需要它解决竞态条件问题。什么是竞态条件？他是各种各样的行动来实现“检查和执行”。检查值和根据这个值来行动。它看起来平常且无害的，但是请您想象一下2个线程或者异步进程同时发生。哪一个将会先被执行？例如，

在任何软件或框架中，都有可能发生竞态条件，只要涉及到并发或异步操作，并且这些操作之间存在某种依赖关系或有可能导致共享资源的不一致状态。NestJS 是一个用于构建服务器端应用程序的框架，它底层使用 Express 或 Fastify，并且支持异步操作，所以在 NestJS 中确实可能会出现竞态条件。

在NestJS中，竞态条件可能出现在以下几个地方：

1. 并发数据库访问：当两个或更多的请求同时对数据库进行写入操作时，可能会导致数据不一致。
2. 中间件并发执行：如果两个中间件或多个中间件访问和修改同一个资源，可能会出现问题。
3. 服务之间的竞态：在 NestJS 中，可以将功能拆分为多个服务。如果多个服务都访问和修改共享资源，也可能出现竞态条件。
4. 异步操作的依赖关系：NestJS 支持 Promise 和 Observable，如果您在不适当的地方或不正确的顺序中解析或订阅它们，可能会遇到问题。

总的来说，NestJS 本身并不特别容易产生竞态条件，但由于其支持并发和异步操作，所以开发者在使用时需要注意避免这种情况。

一个电子钱包叫”A“，里面有30个硬币。一个电子钱包叫”B“,里面有5个硬币。“A”想要转给“B”20枚硬币并完成转账。在后台，请求像往常一样进来，我们的代码会检查他是否有足够的硬币，也就是检查的部分，从“A” 的钱包中扣除这些硬币，加到“B”的钱包，然后更新余额，也就是行动部分。非常简单。但在过程中，由于某种原因，例如双击或恶意攻击，如同时发出大量请求或进入过多的请求，这可能会导致问题。

问题：2个线程或者异步进程同时进行（在几毫秒之内）

现象：当2个进程在同一时间内发生（几毫秒之内）。他们在同一时间执行验证，在这2个进程中，“A”都有30个硬币。最后，本来由于没有足够的硬币，第二个进程应该被取消掉，但是他们都进行了取出，添加，更新余额的操作。

结果：“B”收到了40个硬币，“A”只取出了20个硬币。

我们应该如何阻止它呢？那么，就需要利用互斥锁。互斥帮助异步的进程，确认某些类似于上面例子的情况通过锁定第一个进程并在处理完成时释放锁，然后处理第二个进程。

说明和解释已经足够了，让我们在NestJS中来实现他吧。

npm install async-mutex  
  
nest g module MutexModule  
nest g service MutexService

互斥模块Module

import { Global, Module } from '@nestjs/common';  
import { MutexService } from './mutex.service';  
  
@Global() // to be able to use from every module  
@Module({  
  providers: [MutexService],  
  exports: [MutexService],  
})  
export class MutexModule {}

互斥服务Service

import { Injectable } from '@nestjs/common';
import { Mutex } from 'async-mutex';

@Injectable()
export class MutexService {
  private mutex = new Mutex();

  async runLocked<T>(callback: () => Promise<T>): Promise<T> {
    // acquire lock
    const release = await this.mutex.acquire();
    try {
      return await callback();
    } catch (e) {
      // handle exception
    } finally {
      // release the lock in the end
      release();
    }
  }
}

使用互斥服务

//...  
private mutexService: MutexService, // Inject the service  
//...  
await this.mutexService.runLocked(async () => {  
  //... check and act  
});  
//...

就是这样，你都准备好了。这是我在代码中使用的方法，它按预期工作得很好。这是一种不好的做法吗？我不确定。如果你知道更好的方法或建议，请在评论中分享，让大家都看到。

这篇好文章是转载于：学新通技术网