这并不像听起来那么简单，我们不能访问通道队列来修改它的内容，而且，即使我们可以，也不容易找到一个具有相同id的项，我们必须迭代整个队列。distinctUntilChangedBy()是一个非常不同的情况，因为它只比较最后一项，而不是遍历整个队列。
我认为我们最好的选择是不使用通道提供的队列，而是自己在Map中存储数据，并且只提供发送和接收功能。我将其实现为一个类似流的操作符，并且使其通用，因此它可以用于其他类似的情况：

context(CoroutineScope)
fun <T : Any, K> ReceiveChannel<T>.groupingReduce(keySelector: (T) -> K, reduce: (T, T) -> T): ReceiveChannel<T> = produce {
    val items = mutableMapOf<K, T>()
    while (!isClosedForReceive) {
        select<Unit> {
            if (items.isNotEmpty()) {
                val (key, item) = items.entries.first()
                onSend(item) {
                    items.remove(key)
                }
            }
            onReceiveCatching { result ->
                val item = result.getOrElse { return@onReceiveCatching }
                items.merge(keySelector(item), item, reduce)
            }
        }
    }
    items.values.forEach { send(it) }
}

它将数据保存在map中，尝试同时发送和接收，无论哪个先完成。如果接收到一个项，并且键已经在map中，它允许以调用者提供的方式合并两个值。它按照项在源通道中第一次出现的顺序发送项，因此相同键的新值不会将该项推回到队列中的最后一个位置。
这就是我们如何在您提供的示例中使用它。我对它做了一些修改，因为您的版本让我感到困惑。它在生成(1, 17)之前消耗(1, 10)，所以实际上示例是不正确的。此外，生产者和消费者不会同时运行，因此并发启动它们并添加延迟不会有太大变化：

suspend fun main(): Unit = coroutineScope {
    val channel = Channel<Item>(Channel.UNLIMITED)
    val channel2 = channel.groupingReduce(
        keySelector = { it.id },
        reduce = { it1, it2 -> if (it1.value > it2.value) it1 else it2 }
    )

    for(item in testData) {
        println("Producing item $item")
        channel.send(item)
    }
    channel.close()

    // Needed because while using `UNLIMITED` sending is almost immediate,
    // so it actually starts consuming at the same time it is producing.
    delay(100)

    for (item in channel2) {
        println(item.toString())
    }
}

我创建了另一个生产者和消费者实际上并发运行的例子，每100ms生产一个项目，每200ms消费一个项目，初始延迟为50ms。

suspend fun main(): Unit = coroutineScope {
    val channel = Channel<Item>(Channel.UNLIMITED)
    val channel2 = channel.groupingReduce(
        keySelector = { it.id },
        reduce = { it1, it2 -> if (it1.value > it2.value) it1 else it2 }
    )

    launch {
        delay(50)
        for (item in channel2) {
            println(item.toString())
            delay(200)
        }
    }

    launch {
        listOf(
            Item(1, 10),
            // consume: 1, 10
            Item(2, 20),
            Item(1, 30),
            // consume: 2, 20
            Item(3, 40),
            Item(1, 50),
            // consume: 1, 50
            Item(4, 60),
            Item(1, 70),
            // consume: 3, 40
            Item(5, 80),
            // consume: 4, 60
            // consume: 1, 70
            // consume: 5, 80
        ).forEach {
            channel.send(it)
            delay(100)
        }
        channel.close()
    }
}

也许有一个更好的方法来解决这个问题。而且，老实说，我不是100%肯定这个代码是正确的。也许我错过了一些角落的情况下，频道关闭，取消或失败。此外，我不知道是否select { onSend() }保证，如果代码块没有被执行，那么该项目没有被发送。如果我们取消send()，我们不能保证该商品没有被发送。在这种情况下可能是相同的。