萌新请教一个业务乐观锁问题，烦请大佬指点迷津！

关于使用数据库作为"乐观锁"的问题

业务背景

有一个业务流程，机构下多个商户按照日期分开进行作业，目前设计是按照商户+日期作为唯一主键生成商户当日监控，监控里有字段表示业务处理的阶段 STEP 和状态 Status ，举例商户 A+20220506 日，步骤 1 （待处理 /处理中 /成功 /失败），当前一个步骤成功后进行下一个步骤，在每个步骤里会进行不同业务处理。

步骤 1-成功 -> 步骤 2-初始化 -> 步骤 2-处理中 -> 步骤 2-成功 -> 终态

监控数据生成后，有定时任务 3 分钟触发继续一次执行处理步骤，在这里为了解决竞争问题，查出监控数据后，每条监控用线程池进行处理，STEP 阶段里加了一个步骤加锁状态 LOCK ，表示这个监控任务正在某个步骤处理中，采用的数据库乐观锁的方式

// 获取数据 按照步骤和状态分类
List<Monitor> monitorList = selectUnfinishMonitorList();
Map<StepHandler, List<Monitor>> monitorHandleMap = groupByStep(monitorList);
monitorHandleMap.entrySet()
	.forEach((StepHandler, list) -> threadPool.submit(StepHandler.apply(list)));

StepHandler1.apply(Monitor monitor) {
	// 此处假设为 Step1
	Step originStep = monitor.getStep();
	Status originStatus = monitor.getStatus();
	if (Step.LOCK.equals(originStep)) {
		// 当前监控已经在处理中
		return;
	}
	try {
		// 获取锁
		long update =
	                monitorDBManager.updateStepForLock(monitor.getId(), originStep,
	                        Step.LOCK, originStatus);
		if (update < 1) {
			log.info("STEP1 乐观锁修改失败，已被其他线程处理");
			return;
		}
		
		doBusiness...

		// 成功更新数据
		monitor.setStep(Step.Step1);
		monitor.setStatus(Status.SUCCESS);
		monitor,setUpdateTime(LocalDateTime.now());
		monitorDBManager.update(monitor);

	} catch (Exception e) {
		log.error("Step1 exception", e);
		monitorDBManager.updateStepForLock(monitor.getId(), Step.LOCK,
	                        originStep, originStatus);

	}

}

StepHandler2.apply(Monitor monitor) {
	// 此处假设为 Step2
	Step originStep = monitor.getStep();
	Status originStatus = monitor.getStatus();
	if (Step.LOCK.equals(originStep)) {
		// 当前监控已经在处理中
		return;
	}
	try {
		// 获取锁
		long update =
	                monitorDBManager.updateStepForLock(monitor.getId(), originStep,
	                        Step.LOCK, originStatus);
		if (update < 1) {
			log.info("STEP2 乐观锁修改失败，已被其他线程处理");
			return;
		}
		
		doBusiness...

		// 成功更新数据
		monitor.setStep(Step.Step2);
		monitor.setStatus(Status.SUCCESS);
		monitor,setUpdateTime(LocalDateTime.now());
		monitorDBManager.update(monitor);

	} catch (Exception e) {
		log.error("Step2 exception", e);
		monitorDBManager.updateStepForLock(monitor.getId(), Step.LOCK,
	                        originStep, originStatus);

	}

}

// 数据库 LOCK 操作
updateStepForLock(Long id, String originStep, String targetStep, String moniStatus) {
	UPDATE monitor_table
        SET STEP = #{targetStep},
            UPDATE_TIME = now()
        WHERE id = #{id}
          AND STEP = #{originStep}
          AND STATUS = #{moniStatus}
}

遇见问题

有个机构下有 2700+商户，由于休息日不能进行交易（支付通道不结算），五一期间 4.29-5.4 日六天时间在 5.5 日上午 10 点开启时，被机构连几秒钟内用 6 次，保存监控数据使用 mybatis-plus 提供的 save(list)方法，结果在几秒钟内调用 6 次 save 方法，提示保存成功（此方法没有返回值，根据我自己的日志判定，没有保存异常），保存成功后继续处理后续步骤，但是此时数据库里并没有这些监控数据，导致后续步骤处理的时候出现异常（无法更新监控状态，导致批次 1 任务疯狂执行），第三方 10 分钟调用一次，由于数据库并没有记录，导致这些数据生成了多次，都是数据库里没有数据，后来大概下午的时候数据库出现了这些数据
此后多次出现死锁现象，出现在不同步骤的获取锁的时候，有的是死锁，有的是获取锁超时，这个业务流程没有使用过数据库显示加锁（ for update 等）

org.springframework.dao.DeadlockLoserDataAccessException: 
### Error updating database.  Cause: com.mysql.jdbc.exceptions.jdbc4.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction
### The error may involve defaultParameterMap
### The error occurred while setting parameters
### SQL: UPDATE monitor_table         SET STEP = ?,             UPDATE_TIME = now()         WHERE id = ?           AND STEP = ?           AND STATUS = ?
### Cause: com.mysql.jdbc.exceptions.jdbc4.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction
; SQL []; Deadlock found when trying to get lock; try restarting transaction; nested exception is com.mysql.jdbc.exceptions.jdbc4.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction
	at org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator.doTranslate(SQLErrorCodeSQLExceptionTranslator.java:263)
	at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:73)
	at org.mybatis.spring.MyBatisExceptionTranslator.translateExceptionIfPossible(MyBatisExceptionTranslator.java:73)
	at org.mybatis.spring.SqlSessionTemplate$SqlSessionInterceptor.invoke(SqlSessionTemplate.java:446)

org.springframework.dao.CannotAcquireLockException: 
### Error updating database.  Cause: com.mysql.jdbc.exceptions.jdbc4.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction
### The error may involve defaultParameterMap
### The error occurred while setting parameters
### SQL: UPDATE monitor_table         SET STEP = ?,             UPDATE_TIME = now()         WHERE id = ?           AND STEP = ?           AND STATUS = ?
### Cause: com.mysql.jdbc.exceptions.jdbc4.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction
; SQL []; Lock wait timeout exceeded; try restarting transaction; nested exception is com.mysql.jdbc.exceptions.jdbc4.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction
	at org.springframework.jdbc.support.SQLErrorCodeSQLExceptionTranslator.doTranslate(SQLErrorCodeSQLExceptionTranslator.java:259)
	at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:73)
	at org.mybatis.spring.MyBatisExceptionTranslator.translateExceptionIfPossible(MyBatisExceptionTranslator.java:73)

疑惑

数据库瞬时达到 2000+QPS 时，出现这种保存成功（存疑），数据库没有数据，很久之后出现数据是什么问题？
这段乐观锁是否存在问题导致发生死锁，为什么会出现这种异常？如何在不改动数据库的情况下解决？

萌新请教一个业务乐观锁问题，烦请大佬指点迷津！

关于使用数据库作为"乐观锁"的问题

业务背景

遇见问题

疑惑

恳求大佬帮忙解疑，感谢大家！！！