29 的内存索引可以拒绝同一任务的重复完成,仍缺少一个问题:生产者提交任务时,谁来限制待处理项?如果一个 `Channel.send` 返回成功,是**队列已有位置**,还是消费者已将业务状态写成 Succeeded?这两个完成点分不开,服务往往在队列入列时误报成功;消费者故障后,先前的“发送成功”便成了一条没有去向的承诺。

冻结的 Kotlin/JVM 组合不变:编译器/KGP/stdlib 2.2.21、Gradle 8.14、JDK/toolchain 21、language/API 2.2、jvmTarget 21,kotlinx.coroutines core/test 1.10.2,库源码 tag 5f8900478a8e20c073145b1608fbc71fe3d7378b。先核对 官方 Channel 指南 和 Channel、trySend、receiveCatching、close API。后文的排队与丢弃观察来自本工程受控闸门,不是所有持久消息系统的保证。

一个容量为 1 的任务队列

新建的 BoundedTaskQueue(capacity = 1) 包装 Channel<TaskId>(1)。enqueue 调用挂起的 send,tryEnqueue 暴露 ChannelResult<Unit>;消费入口 consumeOne(process) 先用 receiveCatching() 取得 ID,再显式运行由调用方提供的挂起业务步骤:

1
2
3
4
suspend fun consumeOne(process: suspend (TaskId) -> TaskState): TaskState? {
val id = channel.receiveCatching().getOrNull() ?: return null
return process(id)
}

返回 null 表示此处已经从关闭/取消的队列中读不到下一项,不是“这批任务执行成功”。process 抛出异常时,consumeOne 让异常原样往外走;不会因为 ID 曾入列,就返回一个伪造的 TaskState.Succeeded。这里没有后台消费者常驻的服务循环,只有由测试明确启动的调用者。队列是当前进程的一段内存;崩溃、重启、消息重试与去重日志都未引入。

第一组实验先把 task-30-first 送入容量 1 的队列,再异步 enqueue(task-30-second);执行测试调度器的当前待办后,第二个发送者尚未完成。随后启动消费者,消费者取走第一项,通知已经进入业务步骤,再等 release 闸门;队列空出一个槽位,第二个 send 于是返回,但第一条业务处理的 Deferred 仍没完成。放开业务闸门后,第一项才返回 Succeeded("completed:task-30-first");关闭队列后,第二项仍可被消费并显式完成。记录为 CHANNEL full_pending=true slot_freed_send_done=true process_pending=true ...。一个槽位的背压约束的是待接纳任务数,不是同时处理中的业务步骤数;已经被消费者取走并卡在业务步骤中的第一项不再占着队列槽位。

相同测试再起一个独立容量 1 的队列:已占槽位时第二个发送者挂起,确认它还没返回后取消并 join() 它,再关闭队列,消费既有的第一项并确认下一次 consumeOne 为 null。输出 cancelled_sender_dropped=true 对应这个明确次序。它不承诺同时取消/取走槽位的所有交错,更不能把“发送者取消”解释为撤销已经消费的任务;发送成功与撤销成功必须分开设计。

close 与 cancel:旧项还有谁负责

普通 close() 拒绝后续 trySend,但已成功放入队列的元素仍可取出。第二组测试在 close 前 tryEnqueue(task-30-drain) 为 success,close 后 tryEnqueue(too-late) 为失败;前一项仍经 consumeOne 得到 Succeeded,再取时为空。记录 CHANNEL close_rejects_new=true accepted_item_drained=true terminal_receive_empty=true。没有声称 close 会等待处理中的消费者;第一组业务闸门恰恰在槽位释放、下一项发送返回后仍需要单独放行。想让服务优雅退出,还需停止新请求、等待/取消在途任务、核对已经入列却未处理项的归属。

cancel() 与 close 的意图也不能混用。第三组在另一条队列中先入列 task-30-undelivered,再调用 cancel(),构造时提供的 onUndeliveredElement 收到这一条未被消费的 ID,计数 1。回调只在本机记录“丢弃”,不持久化、也不自动补偿外部副作用。回调内部再次抛异常、消费者已经取走条目时的时序、重启后重投都还没有验收;需要独立责任表,而不是依赖内存 Channel 的回调充当可靠日志。

发送成功之后消费者抛异常

同一个第三组测试先 tryEnqueue(task-30-failed),确认返回 success。随后在 supervisorScope 中启动消费者,process 在取到 ID 后抛合成 IllegalStateException("injected task-30 consumer failure");调用者 await() 收到同消息的异常。关闭队列并再次消费时为空。输出 CHANNEL send_accepted_consumer_failed=true cancel_undelivered=1 的前半段不是一次对 trySend 的负例:发送确实成功了,后面失败的是业务处理;该项已经由消费者拿走,不应凭空把它当成“尚未交付”的项计入 cancel 回调。

检查点 本篇实测 对外能报告什么 还没验证什么
send/trySend 成功 任务 ID 已获一个槽位或交给等待的接收者 内存队列接受了提交 消费者是否已运行、结果是否持久化
receiveCatching 取得 ID 槽位空出来;另一个发送者可返回 消费者拿到了某个 ID process 是否成功;失败后是否重试
process 返回 Succeeded 合成步骤完成 本次内存业务步骤成功 是否写 Store、发外部消息或提交事务
process 抛异常 发送仍是已发生的事实 应报告执行失败并确定责任 自动重投、幂等补偿、持久化确认

TaskState.Succeeded 依旧来自受管任务的本地返回值,没有数据库写入。这与 27 的“Job 已取消不代表底层操作停止”是同一类误读:一次 API 调用的完成点不能代替下游外部副作用的完成点。

练习答案与验收

  1. 一个 Channel 槽位空出、第二次 send 返回时,第一条任务能报业务成功吗?不能。 第一项可能已被取出但正等着 release,定向测试此时 process_pending=true;要等处理函数显式完成才知道结果。
  2. 服务关闭接纳时,要用 close() 还是 cancel()?取决于既有条目的处理策略。 本例 close 后旧项仍被消费,cancel 后未消费项由 onUndeliveredElement 计一次丢弃;两条路径都不能自动替你确认已有外部副作用。
  3. trySend 返回 success 后消费者抛异常,重发任务会不会确保“只执行一次”?没有这个保证。 当前只验证合成故障的传播与队列清空;跨任务去重、事务提交和重试幂等仍需后续协议。

verification/30/ 保存 32 组累计 JUnit、三个定向用例的闸门输出、运行环境、Gradle/CLI 正反命令与退出码。同名素材 kotlin-language-30.tar.gz 提供完整累计工程、Wrapper、RUN/VERSIONS、SHA-256 和解包复跑步骤。下一篇用 Flow 发布进度:要先问何时执行、谁触发收集,不能把队列交付和冷流收集又混为同一次业务完成。