音频预标注异步架构
业务背景
现有一音频标注系统（标注结果用于语音模型训练），该系统有 n 个数据集，每个数据集包含 m 个上传任务，每个上传任务有大约 50 万条音频文件用于标注。为了提高标注人员的标注效率，现需要增加“预标注”功能。
业务需求
需求一
上传任务中的所有音频都需要发送给识别服务进行预标注
需求二
上传任务中所有音频完成预标注之后，该上传任务所属数据集才可以创建标注任务
业务实现
总体实现概览
上图是整个预标注业务的整体流程概览。标注系统是基于 Java 的 web 服务，识别服务是基于 C 语言的 ASR 服务。
需求一 实现
新建上传任务之后，标注系统遍历上传任务中的每一条音频，将该音频相关信息发送到消息队列，以供后续识别服务进行预标注。
识别服务从消息队列中消费音频信息，对音频进行预标注，最后将音频预标注结果发送给消息队列。
标注系统从消息队列消费音频预标注结果并持久化。
需求二 实现
给数据集表新增一个字段用于判断是否可以从该数据库创建新的标注任务，在新建上传记录之后，将该字段设置为”否“，在上传任务中所有音频都被预标注完成之后，再将该字段设置为”是“。
问题一
由于整个预标注流程是异步进行的，如何判断整个上传记录中的音频都完成预标注？可以在标注系统获取预标注结果时，对比总的音频数量和已经完成预标注的音频数量，如果总的音频数量=已经完成预标注的音频数量，那么就将数据集设置为可以创建标注任务。
理想情况下，解决完问题一之后，整个音频预标注业务就算实现了。但是实际环境往往是复杂的，由于需求二的实现需要任何消息都不能丢失，无论是标注系统发送给识别服务还是识别服务发送给标注系统。考虑到会出现系统服务重启、升级或者崩溃不可用等各种情况，我们需要额外的重试机制来保证每一条音频都完成预标注。由此，引申出了问题二。
问题二
一条音频的完整预标注流程包含四个步骤：
- 标注系统发送音频消息到消息队列
- 识别服务读取音频消息并进行预标注
- 识别服务发送音频标注消息到消息队列
- 标注系统读取音频预标注消息并持久化
以上四个步骤中任何一个步骤出问题都会导致最终音频预标注完成数量和音频总数量对不上，进而导致整个数据集无法新建标注任务。举个例子，在极端情况下，一个上传记录中有 50 万条音频，其中某一条音频的预标注流程出了问题，即使 499999 条都已经完成预标注，整个数据集还是无法新建标注任务。为了解决这个问题，我们需要确保四个步骤中的每一步都稳定可靠。
步骤 1
标注系统遍历上传记录中的所有音频，发送音频信息到消息队列。在这个过程中，可能存在两个问题。
第一个问题，如何确保消息成功发送给消息队列？解决这个问题可以使用消息队列的同步发送方式，消息成功发送给消息队列之后，消息队列会返回一个消息 ID（offsetMsgId），通过消息 ID 可以确认标注系统是否成功将消息发送给消息队列。
第二个问题，在发送音频信息消息的过程中，标注系统由于各种原因不可用，导致消息发送进程中断，如何保存以及恢复这种中间状态（已发送的消息不需要重复发送，未发送的消息需要继续发送）？笔者预想的解决方案是，使用支持从持久化恢复的定时任务，如 XXL-JOB 来处理步骤 1。
步骤 2 和步骤 3
步骤 2 和步骤 3 均为识别服务参与的步骤，为了减少系统的复杂度，笔者认为识别服务这边不需要额外的逻辑来判断步骤 2 中的音频信息消息是否被消费以及步骤 3 中的预标注结果消息是否被发送。因为步骤 2 和步骤 3 是否正常进行的评判标准正好为步骤 4 能否顺利进行。
步骤 4
在一切顺利的前提下，步骤 2 和 3 完成之后，消息队列中存在一条音频预标注结果的消息，步骤 4 中标注系统消费该条消息并将预标注结果持久化。至此，一条音频就完成了整个预标注业务。因此，我们可以将音频是否存在预标注结果作为步骤 4 是否顺利进行的标志。
真实环境往往是复杂的，会存在多种可能性导致步骤 4 无法顺利进行。第一个可能性就是识别服务的不可用（服务本身问题、机房维护等等）导致步骤 2 或 3 出问题。第二个可能性就是标注服务的不可用（服务本身问题，机房维护等等）导致音频预标注结果消息的消费或者预标注结果的持久化无法顺利进行。
具体是哪个服务出问题，出什么问题没有必要细究，我们需要考虑的是在标注系统和识别服务都恢复正常，即步骤 2、3、4 均能正常进行的前提下，如何恢复整个预标注流程。结合之前的说明，为了后续讨论方便，笔者将步骤 2、3、4 中无论哪个步骤出现问题都看待为步骤 4 出现问题，即标注系统无法获取音频的预标注结果。
从 总体实现概览 中可以发现，步骤 4 的进行需要步骤 1 发送音频信息消息。步骤 4 出问题换个角度可以看待成步骤 1 没有发送音频消息。因为要解决步骤 4 的问题，前提肯定是导致步骤 4 出现问题的服务恢复正常，即步骤 4 能够顺利进行得到保证，这样的前提下，只要步骤 1 再发送一次音频信息消息，整个预标注业务就能继续进行了。
再次进行步骤 1
步骤 1 中的定时任务通过 offsetMsgId 确保一个上传记录中的所有音频都发送一次给消息队列。为了解决步骤 4 无法顺利进行的问题，需要另一个机制让定时任务去重新发送被步骤 4 已经消费但是“忽略”的音频信息消息。
要实现这种机制，我们需要从实际的例子出发。假设一个上传记录有 10 条音频，其中 9 条音频已经顺利完成了预标注流程，只剩 1 条音频由于步骤 4 出现问题而无法完成整个流程。在步骤 1 的定时任务中，这 10 条音频的状态为，10 条音频都存在 offsetMsgId 字段，9 条音频存在预标注结果。那么我们是否可以用是否存在预标注结果作为再次发送消息的依据呢？在当前例子中，不存在预标注结果确实是由于步骤 4 出现问题，但是还存在着其他更复杂的情况导致暂时没有预标注结果。比如存在多个上传记录同时进行预标注流程，标注系统一直在读取其他上传记录中的预标注结果消息，当前例子中的那 1 条音频的预标注结果在消息队列中，但是标注系统尚未来得及处理。又比如只有当前例子中一个上传记录，在定时任务执行的时候，标注系统才将上一条音频的预标注结果持久化，所以只剩下 1 条音频的预标注结果还在消息队列中，未被标注系统消费并持久化。
因此，仅靠是否存在预标注结果并不能作为步骤 1 再次 发送消息的依据。只要标注系统还在消费预标注结果消息，就不能判断预标注结果是否不存在。换句话说，如果标注系统不再消费消息了，那么就可以确定，该条音频的预标注结果由于步骤 4 的问题而不存在，进而可以再次发送步骤 1 中的消息了。
总结下来，再次发送步骤 1 中的消息需要两个条件：
- 预标注结果未持久化
- 标注系统停止消费消息
条件 1 我们可以通过查询数据库轻松获取，条件 2 呢？条件 2 的获取也很简单，我们只需要去查询消息队列中未消费的消息数量即可。可惜的是，在实际开发中，笔者所使用的 RocketMQ 的 SDK 并未提供相应查询未消费消息数量的方法，因此只能采取另一种解决方案。为了知道标注系统何时停止消费消息，我们可以在消费消息的时候将当前时间存入 Redis。每次定时任务执行的时候，用当前时间对比 Redis 的最近消费时间，如果两者之间的之间差超过了 5 分钟，那么就可以说明标注系统已经停止消费消息了。为了准确性，可以将时间差的阈值提高，调整为 10 分钟。