MySQL数据库: 高并发电商场景下的架构设计与优化

// app.module.ts
import { module } from '@nestjs/common';
import { typeormmodule } from '@nestjs/typeorm';
import { configmodule, configservice } from '@nestjs/config';
 
@module({
  imports: [
    typeormmodule.forrootasync({
      imports: [configmodule],
      usefactory: (config: configservice) => ({
        type: 'mysql',
        host: config.get('db_host'),
        port: config.get('db_port'),
        username: config.get('db_user'),
        password: config.get('db_pass'),
        database: config.get('db_name'),
        entities: [__dirname + '//*.entity{.ts,.js}'],
        synchronize: false,
        logging: ['error'], // 避免日志影响性能
        extra: {
          connectionlimit: 100,
          queuelimit: 10,
          connecttimeout: 10000,
          acquiretimeout: 15000,
          timeout: 10000,
        },
        replication: {
          master: {
            host: config.get('db_master_host'),
            port: config.get('db_master_port'),
            username: config.get('db_master_user'),
            password: config.get('db_master_pass'),
          },
          slaves: [
            { host: 'slave1.example.com', port: 3306 },
            { host: 'slave2.example.com', port: 3306 },
            // ... 其他 slave
          ],
        },
      }),
      inject: [configservice],
    }),
  ],
})
export class appmodule {}

// redis-lock.service.ts
import { injectable } from '@nestjs/common';
import { redis } from 'ioredis';
 
@injectable()
export class redislockservice {
  private readonly lock_script = `
    return redis.call('set', keys[1], argv[1], 'nx', 'px', argv[2])
  `;
  private readonly unlock_script = `
    if redis.call("get",keys[1]) == argv[1] then
        return redis.call("del",keys[1])
    else
        return 0 
    end
  `;
 
  constructor(private readonly redis: redis) {}
 
  async acquirelock(
    key: string,
    requestid: string,
    expirems: number,
  ): promise<boolean> {
    const result = await this.redis.eval(
      this.lock_script,
      1,
      key,
      requestid,
      expirems,
    );
    return result === 'ok';
  }
 
  async releaselock(key: string, requestid: string): promise<boolean> {
    const result = await this.redis.eval(this.unlock_script, 1, key, requestid);
    return result === 1;
  }
}

alter table product add column version int default 0;
update product 
set stock = stock - 1, version = version + 1 
where id = ? and version = ?

// product.service.ts
@injectable()
export class productservice {
  @injectrepository(product)
  private repo: repository<product>;
 
  async decreasestock(productid: number, expectedversion: number): promise<boolean> {
    const result = await this.repo
      .createquerybuilder()
      .update(product)
      .set({ stock: () => 'stock - 1', version: () => 'version + 1' })
      .where('id = :id and version = :version', { id: productid, version: expectedversion })
      .execute();
 
    return result.affected > 0;
  }
}

// order.processor.ts
@processor('order_queue')
export class orderprocessor {
  @process('create_order')
  async handleordercreation(job: job) {
    const { orderid, userid, items } = job.data;
    try {
      await this.orderservice.createorder(orderid, userid, items);
      return { success: true };
    } catch (err) {
      await job.movetofailed({ message: err.message }, true);
      throw err;
    }
  }
}

await this.queue.add('create_order', { orderid, userid, items }, {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
});

// inventory.service.ts
@injectable()
export class inventoryservice {
  constructor(@injectrepository(product) private productrepo: repository<product>) {}
 
  async deductstockoptimistic(productid: string, quantity: number): promise<boolean> {
    const queryrunner = this.productrepo.manager.connection.createqueryrunner();
    await queryrunner.starttransaction();
 
    try {
      const product = await queryrunner.manager.findone(product, {
        where: { id: productid },
        lock: { mode: 'pessimistic_read' }, // 可选：加悲观读锁防止脏读
      });
 
      if (!product || product.stock < quantity) {
        throw new error('insufficient stock');
      }
 
      const affected = await queryrunner.manager.update(
        product,
        { id: productid, stock: morethanorequal(quantity), version: product.version },
        { stock: () => `stock - ${quantity}`, version: () => `version + 1` }
      );
 
      if (affected.affected === 0) {
        throw new error('concurrent update detected');
      }
 
      await queryrunner.committransaction();
      return true;
    } catch (err) {
      await queryrunner.rollbacktransaction();
      throw err;
    } finally {
      await queryrunner.release();
    }
  }
}

// redis.inventory.service.ts
@injectable()
export class redisinventoryservice {
  private readonly stock_key = (id: string) => `inventory:${id}`;
 
  constructor(private readonly redisservice: redisservice) {}
 
  async deductwithpessimisticlock(
    productid: string,
    quantity: number,
    ttlms = 5000
  ): promise<boolean> {
    const lockkey = `lock:inventory:${productid}`;
    const client = this.redisservice.getclient();
 
    // 获取分布式锁（setnx + expire）
    const acquired = await client.set(lockkey, '1', 'px', ttlms, 'nx');
    if (!acquired) {
      throw new error('failed to acquire inventory lock');
    }
 
    try {
      const stockscript = `
        local current = tonumber(redis.call('get', keys[1]))
        if not current or current < tonumber(argv[1]) then 
          return 0
        end
        redis.call('incrbyfloat', keys[1], -argv[1])
        return 1
      `;
 
      const result = await client.eval(
        stockscript,
        1,
        this.stock_key(productid),
        quantity
      );
 
      return result === 1;
    } finally {
      // 释放锁
      await client.del(lockkey);
    }
  }
}

// order.processor.ts
@processor('order_queue')
export class orderprocessor {
  constructor(
    private readonly inventoryservice: inventoryservice,
    private readonly eventpublisher: eventpublisher
  ) {}
 
  @process('create_order')
  async handleordercreation(job: job<orderpayload>): promise<void> {
    const { orderid, productid, quantity } = job.data;
 
    try {
      const success = await this.inventoryservice.deductstockoptimistic(productid, quantity);
      if (success) {
        await this.eventpublisher.emitasync(new ordercreatedevent(orderid));
      } else {
        await this.eventpublisher.emitasync(new orderfailedevent(orderid, 'out of stock'));
      }
    } catch (error) {
      // 失败重试机制
      if (job.attemptsmade < 3) {
        throw error; // 触发重试
      } else {
        await this.eventpublisher.emitasync(new orderfailedevent(orderid, error.message));
      }
    }
  }
}
 
// bootstrap with bullmq 
const queue = new queue('order_queue', { connection });
const worker = new worker('order_queue', new orderprocessor(...).handleordercreation, { connection });

// redis-lock.service.ts
import { injectable } from '@nestjs/common';
import redis from 'ioredis';
 
@injectable()
export class redislockservice {
  private readonly redis = new redis({
    host: 'redis-cluster-host',
    port: 6379,
  });
 
  private readonly lock_prefix = 'lock:';
  private readonly default_ttl = 5000; // 5s
 
  async acquirelock(key: string, ttlms: number = this.default_ttl): promise<string | null> {
    const lockkey = this.lock_prefix + key;
    const token = date.now().tostring();
 
    const script = `
      if redis.call('get', keys[1]) == false then
        return redis.call('set', keys[1], argv[1], 'px', argv[2])
      else 
        return false 
      end
    `;
 
    const result = await this.redis.eval(script, 1, lockkey, token, ttlms);
    return result ? token : null;
  }
 
  async releaselock(key: string, token: string): promise<boolean> {
    const lockkey = this.lock_prefix + key;
    const script = `
      if redis.call('get', keys[1]) == argv[1] then
        return redis.call('del', keys[1])
      else 
        return 0
      end
    `;
    const result = await this.redis.eval(script, 1, lockkey, token);
    return result === 1;
  }
}

// inventory.service.ts
import { injectable } from '@nestjs/common';
import { redislockservice } from './redis-lock.service';
 
@injectable()
export class inventoryservice {
  constructor(private readonly lockservice: redislockservice) {}
 
  async deductstock(productid: number, quantity: number): promise<boolean> {
    const lockkey = `stock:${productid}`;
    let token: string | null = null;
 
    try {
      token = await this.lockservice.acquirelock(lockkey, 3000);
      if (!token) {
        throw new error('failed to acquire lock');
      }
 
      // 模拟数据库检查与扣减（应在事务中）
      const current = await this.getcurrentstock(productid);
      if (current < quantity) {
        throw new error('insufficient stock');
      }
 
      await this.updatestockindatabase(productid, current - quantity);
      return true;
 
    } catch (err) {
      console.error(err);
      return false;
    } finally {
      if (token) {
        await this.lockservice.releaselock(lockkey, token);
      }
    }
  }
 
  private async getcurrentstock(productid: number): promise<number> {
    // 实际调用 typeorm / prisma 查询
    return 100; // mock
  }
 
  private async updatestockindatabase(productid: number, newstock: number): promise<void> {
    // 更新数据库逻辑 
    console.log(`updating stock for ${productid} to ${newstock}`);
  }
}

// order-producer.service.ts
import { injectable } from '@nestjs/common';
import { clientproxy } from '@nestjs/microservices';
 
@injectable()
export class orderproducerservice {
  constructor(private client: clientproxy) {}
 
  emitordercreated(payload: { userid: number; productid: number; count: number }) {
    this.client.emit('order_created', payload);
  }
}
 
// order-consumer.service.ts
@eventpattern('order_created')
async handleordercreated try {
    await this.inventoryservice.deductstock(data.productid, data.count);
    await this.orderrepository.save(data);
    console.log('order processed:', data);
  } catch (err) {
    // 发送失败则进入死信队列或重试机制 
    console.error('order processing failed:', err);
    throw err; // 触发重试
  }
}