Go语言IoT设备队列去重：避免重复时间命令重复入队方案问询

问题背景 #

在IoT设备上运行的Go应用会接收云端推送的命令，命令被存入一个队列：

var queue chan time.Time

设备上的Worker进程负责处理该队列，任务是向云端回传某一时间段的数据，通道中的时间即为该时间段的起始时间。由于IoT设备采用移动网络连接，数据可能丢失无法送达云端，而云端无法确认设备是否已接收命令，因此会重复发送命令。

我们需要实现的核心需求是：确保若原命令仍在队列中，相同的命令无法被再次推入队列。现有代码框架如下：

func addToQueue(periodStart time.Time) error {
    if alreadyOnQueue(queue, periodStart) {
        return errors.New("periodStart was already on the queue, not adding it again")
    }
    queue <- periodStart
    return nil
}

func alreadyOnQueue(queue chan time.Time, t time.Time) bool {
    return false // todo
}

实现方案 #

Go语言的通道无法直接遍历检查内部元素，因此需要额外维护一个同步集合来跟踪当前队列中的命令，同时保证并发操作的安全性。以下是具体实现：

1. 初始化同步结构 #

新增一个互斥锁保护的map，用于记录当前队列中存在的时间命令，同时初始化队列：

import (
    "errors"
    "sync"
    "time"
)

var (
    queue    chan time.Time
    inQueue  = make(map[time.Time]struct{}) // 记录队列中存在的命令
    queueMu  sync.Mutex                     // 保护map和队列操作的互斥锁
)

// 程序启动时调用，初始化队列缓冲大小
func InitQueue(bufferSize int) {
    queue = make(chan time.Time, bufferSize)
}

2. 改造addToQueue函数 #

在添加命令前，先通过互斥锁保护的map检查是否已存在，确认不存在后再加入队列，同时更新map：

func addToQueue(periodStart time.Time) error {
    queueMu.Lock()
    defer queueMu.Unlock()

    // 检查命令是否已在队列中
    if _, exists := inQueue[periodStart]; exists {
        return errors.New("periodStart was already on the queue, not adding it again")
    }

    // 先将命令加入跟踪map，再尝试发送到队列
    inQueue[periodStart] = struct{}{}
    select {
    case queue <- periodStart:
        return nil
    default:
        // 队列已满，发送失败，回滚跟踪map
        delete(inQueue, periodStart)
        return errors.New("queue is full, cannot add periodStart")
    }
}

3. 改造Worker进程 #

Worker处理完命令后，必须从跟踪map中移除对应的记录，保证队列和map的状态一致：

func Worker() {
    for periodStart := range queue {
        // 执行向云端回传数据的业务逻辑
        uploadPeriodData(periodStart)

        // 处理完成后，从跟踪map中移除该命令
        queueMu.Lock()
        delete(inQueue, periodStart)
        queueMu.Unlock()
    }
}

// 示例业务函数：回传指定时间段的数据
func uploadPeriodData(start time.Time) {
    // 此处实现具体的回传逻辑
}

关键说明 #

• 并发安全：使用sync.Mutex保证inQueue map的读写操作互斥，避免并发场景下的状态不一致。
• 状态一致性：命令加入队列前先更新跟踪map，发送失败时回滚；处理完成后及时从map中移除，确保map和队列的状态始终同步。
• 队列阻塞处理：采用非阻塞发送（select+default）避免锁持有时间过长，若需要阻塞等待队列空闲，可移除default分支，但需注意这会导致addToQueue在队列满时阻塞，且锁会持有到发送完成，可能影响并发性能。

内容的提问来源于stack exchange，提问作者Martijn de Munnik