odboy
8979c744a8
- 重构后端与前端，全程大白话注释（不懂程序也能看懂） - 移动阶段改为全局并行（MoveAll，工人数=CPU核心数） - 去掉记录每个文件搬去哪了（删除去向回填，解决写库卡顿） - 去掉重复文件标记与统计：重复判定改为 MD5+路径都相同才算，重复则原地忽略不移动；MD5 相同但路径不同仍照常搬 - 媒体库去掉归类目录列，看板去掉重复/归类分布统计
63 lines
2.0 KiB
Go
63 lines
2.0 KiB
Go
// 状态订阅/发布中心：把 ScanService 的状态变更实时"广播"给所有订阅者。
|
|
//
|
|
// 大白话理解：这是个广播站。后台干活时进度一变，就通过广播站喊一嗓子；
|
|
// 谁订阅了（比如 WebSocket 连接），谁就收到。没有人听的时候也不影响干活。
|
|
//
|
|
// 还有个贴心小设计：每个订阅者面前的信箱最多存一条消息。
|
|
// 如果进度变得太快、接收方来不及处理，就只留最新一条——反正看进度的人
|
|
// 关心的是"现在到哪了"，而不是"中间每一刻到哪了"。
|
|
package service
|
|
|
|
import "sync"
|
|
|
|
// StatusListener 订阅者的"信箱"（一个缓冲 1 的管道）。
|
|
type StatusListener chan ScanStatus
|
|
|
|
// StatusHub 广播站：管理所有订阅者，负责把状态推给每个人。
|
|
type StatusHub struct {
|
|
mu sync.RWMutex // 保护订阅者名单
|
|
subs map[StatusListener]struct{} // 当前所有订阅者
|
|
}
|
|
|
|
// NewStatusHub 建一个广播站。
|
|
func NewStatusHub() *StatusHub {
|
|
return &StatusHub{subs: make(map[StatusListener]struct{})}
|
|
}
|
|
|
|
// Subscribe 让调用方订阅进度。返回它的信箱，和一个"退订"函数。
|
|
// 退订后信箱会被关闭（写方要自己注意别再往里面塞）。
|
|
func (h *StatusHub) Subscribe() (StatusListener, func()) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
ch := make(StatusListener, 1)
|
|
h.subs[ch] = struct{}{}
|
|
return ch, func() {
|
|
h.mu.Lock()
|
|
if _, ok := h.subs[ch]; ok {
|
|
delete(h.subs, ch)
|
|
close(ch)
|
|
}
|
|
h.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
// Publish 把最新状态推给所有订阅者。
|
|
// 谁的信箱满了（来不及消费），就把里面的旧消息挤掉、换上最新的一条。
|
|
// 整个过程不阻塞，广播站永远不会因为某个订阅者太慢而卡住。
|
|
func (h *StatusHub) Publish(st ScanStatus) {
|
|
h.mu.RLock()
|
|
defer h.mu.RUnlock()
|
|
for ch := range h.subs {
|
|
select {
|
|
case ch <- st:
|
|
default:
|
|
// 信箱满了：清掉旧消息，放最新的进去
|
|
select {
|
|
case <-ch:
|
|
default:
|
|
}
|
|
ch <- st
|
|
}
|
|
}
|
|
}
|