Typeorm订阅者无法获取最新数据及QueryRunner报错问题咨询

问题描述 #

编写了监听review变更的TypeORM订阅者，用于更新目标列的平均评分，但出现两个异常：

1. 当把评分从2更新为3时，查询结果res仍存储旧评分，推测rating表未同步最新数据；
2. 若将const queryRunner = event.connection.createQueryRunner();替换为const queryRunner = event.queryRunner，能得到正确平均评分，但执行update操作时触发QueryRunnerAlreadyReleasedError: Query runner already released. Cannot run queries anymore.错误。

原代码实现：

import { EntityTargetType } from '@src/core/enums/entity-target-type.enum';
import { CourseEntity } from '@src/modules/course/entities/course.entity';
import { HomeworkEntity } from '@src/modules/homework/entities/homework.entity';
import { LectureEntity } from '@src/modules/lecture/entities/lecture.entity';
import { ReviewEntity } from '@src/modules/review/entities/review.entity';
import { EntitySubscriberInterface, EventSubscriber, InsertEvent, RemoveEvent, UpdateEvent } from 'typeorm';

@EventSubscriber()
export class ReviewSubscriber implements EntitySubscriberInterface<ReviewEntity> {
  constructor() {}

  listenTo() {
    return ReviewEntity;
  }

  async afterUpdate(event: UpdateEvent<ReviewEntity>) {
    await this.updateTarget(event);
  }

  async afterRemove(event: RemoveEvent<ReviewEntity>) {
    await this.updateTarget(event);
  }

  async afterInsert(event: InsertEvent<ReviewEntity>) {
    await this.updateTarget(event);
  }

  async updateTarget(event: UpdateEvent<ReviewEntity> | RemoveEvent<ReviewEntity> | InsertEvent<ReviewEntity>) {
    const queryRunner = event.connection.createQueryRunner();
    await queryRunner.connect();
    await queryRunner.startTransaction();

    console.log(event.entity);

    try {
      let lecture: LectureEntity | null = null;
      const query = queryRunner.manager.createQueryBuilder(ReviewEntity, 'review');
      query.select('COALESCE(ROUND(AVG(review.rating), 2), 0)', 'rating');
      query.addSelect('COALESCE(COUNT(review.id), 0)', 'count');

      if (event.entity.targetType === EntityTargetType.LECTURE_BOOK) {
        query.leftJoin('lecture_book_entity', 'lectureBook', 'lectureBook.id = review.targetId');
        query.leftJoin('lectureBook.lecture', 'lecture');

        lecture = await queryRunner.manager.findOne(LectureEntity, {
          where: { bookings: { id: event.entity.targetId } },
        });

        if (!lecture) {
          throw new Error(`Lecture not found for booking id ${event.entity.targetId}`);
        }

        query.where('lecture.id = :lectureId', { lectureId: lecture.id });
      } else {
        query.where('review.targetId = :targetId', { targetId: event.entity.targetId });
      }

      // Perform a select query to get the latest review information
      const res = await query.getRawOne();
      const ratings = {
        ratingsCount: res['count'],
        ratingsAvg: res['rating'],
      };

      console.log(ratings, lecture?.id);

      if (event.entity.targetType === EntityTargetType.COURSE) {
        await queryRunner.manager.update(CourseEntity, { id: event.entity.targetId }, ratings);
      } else if (event.entity.targetType === EntityTargetType.HOMEWORK) {
        await queryRunner.manager.update(HomeworkEntity, { id: event.entity.targetId }, ratings);
      } else if (event.entity.targetType === EntityTargetType.LECTURE_BOOK) {
        if (lecture) {
          await queryRunner.manager.update(LectureEntity, { id: lecture.id }, ratings);
        }
      }

      await queryRunner.commitTransaction();
    } catch (err) {
      await queryRunner.rollbackTransaction();
      console.error('Error updating target:', err);
    } finally {
      await queryRunner.release();
    }
  }
}

原因分析 #

1. 新QueryRunner查不到最新数据的原因 #

TypeORM默认事务隔离级别为READ COMMITTED，而afterUpdate/afterInsert/afterRemove事件默认在原事务提交前触发。此时review的变更操作还在原事务中未提交，新创建的QueryRunner属于独立数据库连接，无法看到未提交的事务数据，因此查询到的是旧评分。

2. 使用event.queryRunner报错的原因 #

event.queryRunner是原review操作使用的QueryRunner，TypeORM会在事件执行完成后自动释放该QueryRunner。代码中用它执行update操作时，原事务已经完成，QueryRunner被回收，后续操作自然触发"已释放"错误。

修复方案 #

方案一：复用原事务的EntityManager（推荐） #

直接使用事件上下文的event.manager，它属于原事务的上下文，能看到未提交的最新review数据，且无需手动管理事务（原事务会统一处理提交/回滚）。

修改后的updateTarget方法：

async updateTarget(event: UpdateEvent<ReviewEntity> | RemoveEvent<ReviewEntity> | InsertEvent<ReviewEntity>) {
  try {
    let lecture: LectureEntity | null = null;
    const query = event.manager.createQueryBuilder(ReviewEntity, 'review');
    query.select('COALESCE(ROUND(AVG(review.rating), 2), 0)', 'rating');
    query.addSelect('COALESCE(COUNT(review.id), 0)', 'count');

    if (event.entity.targetType === EntityTargetType.LECTURE_BOOK) {
      query.leftJoin('lecture_book_entity', 'lectureBook', 'lectureBook.id = review.targetId');
      query.leftJoin('lectureBook.lecture', 'lecture');

      lecture = await event.manager.findOne(LectureEntity, {
        where: { bookings: { id: event.entity.targetId } },
      });

      if (!lecture) {
        throw new Error(`Lecture not found for booking id ${event.entity.targetId}`);
      }

      query.where('lecture.id = :lectureId', { lectureId: lecture.id });
    } else {
      query.where('review.targetId = :targetId', { targetId: event.entity.targetId });
    }

    const res = await query.getRawOne();
    const ratings = {
      ratingsCount: res['count'],
      ratingsAvg: res['rating'],
    };

    console.log(ratings, lecture?.id);

    if (event.entity.targetType === EntityTargetType.COURSE) {
      await event.manager.update(CourseEntity, { id: event.entity.targetId }, ratings);
    } else if (event.entity.targetType === EntityTargetType.HOMEWORK) {
      await event.manager.update(HomeworkEntity, { id: event.entity.targetId }, ratings);
    } else if (event.entity.targetType === EntityTargetType.LECTURE_BOOK) {
      if (lecture) {
        await event.manager.update(LectureEntity, { id: lecture.id }, ratings);
      }
    }
  } catch (err) {
    console.error('Error updating target:', err);
    // 抛出错误让TypeORM回滚原事务
    throw err;
  }
}

方案二：配置事件在事务提交后执行 #

通过transaction: false配置，让事件在原事务提交后触发，此时数据库已同步最新数据，新创建的QueryRunner能正常查询和更新。

修改事件方法的装饰器配置：

import { AfterInsert, AfterRemove, AfterUpdate } from 'typeorm';

// ...

@AfterUpdate({ transaction: false })
async afterUpdate(event: UpdateEvent<ReviewEntity>) {
  await this.updateTarget(event);
}

@AfterRemove({ transaction: false })
async afterRemove(event: RemoveEvent<ReviewEntity>) {
  await this.updateTarget(event);
}

@AfterInsert({ transaction: false })
async afterInsert(event: InsertEvent<ReviewEntity>) {
  await this.updateTarget(event);
}

此方案下，原updateTarget方法的QueryRunner逻辑无需修改，因为事件触发时原事务已提交，新QueryRunner能查到最新数据，且独立管理事务不会冲突。

内容的提问来源于stack exchange，提问作者Korer