Geliebte
1ea8c9da9c
- Wails v2.16.0 + React 19 + TypeScript + Vite 7 - 插件系统（build tag 控制编译，plugin.json 运行时配置） - 支付插件（授权、账号、订单、退款、账单） - 高并发基础设施（任务池、批量处理器、限流器、环形缓冲、管道） - Windows 系统托盘（Win32 API） - CodeGraph 索引 + .memory 记忆包 - 默认图标 assets/icon.png
370 lines
7.4 KiB
Go
370 lines
7.4 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
)
|
||
|
||
// BatchResult 批量任务结果
|
||
type BatchResult struct {
|
||
Success []any
|
||
Fail []BatchError
|
||
}
|
||
|
||
// BatchError 批量任务失败记录
|
||
type BatchError struct {
|
||
Index int
|
||
Item any
|
||
Err error
|
||
}
|
||
|
||
// BatchWorker 批量处理器，适合文件处理、数据转换等高并发场景
|
||
type BatchWorker struct {
|
||
mu sync.Mutex
|
||
running atomic.Int64
|
||
total atomic.Int64
|
||
success atomic.Int64
|
||
failed atomic.Int64
|
||
maxConcurrency int
|
||
}
|
||
|
||
// NewBatchWorker 创建批量处理器
|
||
func NewBatchWorker(maxConcurrency int) *BatchWorker {
|
||
if maxConcurrency <= 0 {
|
||
maxConcurrency = 10
|
||
}
|
||
return &BatchWorker{
|
||
maxConcurrency: maxConcurrency,
|
||
}
|
||
}
|
||
|
||
// ProcessConcurrent 并发处理一批数据
|
||
func (bw *BatchWorker) ProcessConcurrent(
|
||
ctx context.Context,
|
||
items []any,
|
||
handler func(ctx context.Context, item any) (any, error),
|
||
) BatchResult {
|
||
bw.total.Add(int64(len(items)))
|
||
results := BatchResult{}
|
||
sem := make(chan struct{}, bw.maxConcurrency)
|
||
var wg sync.WaitGroup
|
||
var mu sync.Mutex
|
||
|
||
for i, item := range items {
|
||
select {
|
||
case <-ctx.Done():
|
||
break
|
||
default:
|
||
}
|
||
wg.Add(1)
|
||
sem <- struct{}{}
|
||
go func(index int, data any) {
|
||
defer wg.Done()
|
||
defer func() { <-sem }()
|
||
bw.running.Add(1)
|
||
defer bw.running.Add(-1)
|
||
|
||
result, err := handler(ctx, data)
|
||
bw.success.Add(1)
|
||
|
||
mu.Lock()
|
||
defer mu.Unlock()
|
||
if err != nil {
|
||
bw.failed.Add(1)
|
||
results.Fail = append(results.Fail, BatchError{
|
||
Index: index,
|
||
Item: data,
|
||
Err: err,
|
||
})
|
||
} else {
|
||
results.Success = append(results.Success, result)
|
||
}
|
||
}(i, item)
|
||
}
|
||
|
||
wg.Wait()
|
||
return results
|
||
}
|
||
|
||
// ProcessSequential 顺序处理（用于需要严格顺序的场景）
|
||
func (bw *BatchWorker) ProcessSequential(
|
||
ctx context.Context,
|
||
items []any,
|
||
handler func(ctx context.Context, item any) (any, error),
|
||
) BatchResult {
|
||
results := BatchResult{}
|
||
for i, item := range items {
|
||
select {
|
||
case <-ctx.Done():
|
||
return results
|
||
default:
|
||
}
|
||
result, err := handler(ctx, item)
|
||
if err != nil {
|
||
results.Fail = append(results.Fail, BatchError{Index: i, Item: item, Err: err})
|
||
} else {
|
||
results.Success = append(results.Success, result)
|
||
}
|
||
}
|
||
return results
|
||
}
|
||
|
||
// Stats 当前处理统计
|
||
func (bw *BatchWorker) Stats() (running, total, success, failed int64) {
|
||
return bw.running.Load(), bw.total.Load(), bw.success.Load(), bw.failed.Load()
|
||
}
|
||
|
||
// RingBuffer 无锁环形缓冲区，适合高频消息传递
|
||
type RingBuffer struct {
|
||
data []any
|
||
head int64
|
||
tail int64
|
||
capacity int
|
||
size atomic.Int64
|
||
}
|
||
|
||
// NewRingBuffer 创建环形缓冲区
|
||
func NewRingBuffer(capacity int) *RingBuffer {
|
||
if capacity <= 0 {
|
||
capacity = 1024
|
||
}
|
||
rb := &RingBuffer{
|
||
data: make([]any, capacity),
|
||
capacity: capacity,
|
||
}
|
||
return rb
|
||
}
|
||
|
||
// Enqueue 入队（阻塞直到有空位）
|
||
func (rb *RingBuffer) Enqueue(item any) {
|
||
for {
|
||
tail := atomic.LoadInt64(&rb.tail)
|
||
head := atomic.LoadInt64(&rb.head)
|
||
if tail-head >= int64(rb.capacity) {
|
||
time.Sleep(time.Microsecond)
|
||
continue
|
||
}
|
||
if atomic.CompareAndSwapInt64(&rb.tail, tail, tail+1) {
|
||
rb.data[tail%int64(rb.capacity)] = item
|
||
rb.size.Add(1)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// Dequeue 出队（阻塞直到有数据）
|
||
func (rb *RingBuffer) Dequeue() (any, bool) {
|
||
for {
|
||
head := atomic.LoadInt64(&rb.head)
|
||
tail := atomic.LoadInt64(&rb.tail)
|
||
if head >= tail {
|
||
return nil, false
|
||
}
|
||
if atomic.CompareAndSwapInt64(&rb.head, head, head+1) {
|
||
item := rb.data[head%int64(rb.capacity)]
|
||
rb.data[head%int64(rb.capacity)] = nil
|
||
rb.size.Add(-1)
|
||
return item, true
|
||
}
|
||
}
|
||
}
|
||
|
||
// Size 当前元素数量
|
||
func (rb *RingBuffer) Size() int64 {
|
||
return rb.size.Load()
|
||
}
|
||
|
||
// RateLimiter 令牌桶限流器
|
||
type RateLimiter struct {
|
||
tokens int64
|
||
maxTokens int64
|
||
refillRate int64 // 每秒补充数
|
||
lastRefill time.Time
|
||
mu sync.Mutex
|
||
}
|
||
|
||
// NewRateLimiter 创建限流器
|
||
func NewRateLimiter(maxTokens, refillRate int64) *RateLimiter {
|
||
rl := &RateLimiter{
|
||
tokens: maxTokens,
|
||
maxTokens: maxTokens,
|
||
refillRate: refillRate,
|
||
lastRefill: time.Now(),
|
||
}
|
||
go rl.refillLoop()
|
||
return rl
|
||
}
|
||
|
||
func (rl *RateLimiter) refillLoop() {
|
||
ticker := time.NewTicker(time.Second)
|
||
defer ticker.Stop()
|
||
for range ticker.C {
|
||
rl.mu.Lock()
|
||
elapsed := time.Since(rl.lastRefill).Seconds()
|
||
add := int64(elapsed * float64(rl.refillRate))
|
||
current := rl.tokens
|
||
if current+add > rl.maxTokens {
|
||
add = rl.maxTokens - current
|
||
}
|
||
rl.tokens += add
|
||
rl.lastRefill = time.Now()
|
||
rl.mu.Unlock()
|
||
}
|
||
}
|
||
|
||
// Acquire 获取一个令牌，返回是否成功
|
||
func (rl *RateLimiter) Acquire() bool {
|
||
rl.mu.Lock()
|
||
defer rl.mu.Unlock()
|
||
if rl.tokens <= 0 {
|
||
return false
|
||
}
|
||
rl.tokens--
|
||
return true
|
||
}
|
||
|
||
// Limit 带限制的包装函数
|
||
func (rl *RateLimiter) Limit(fn func() error) error {
|
||
for !rl.Acquire() {
|
||
time.Sleep(time.Millisecond * 10)
|
||
}
|
||
return fn()
|
||
}
|
||
|
||
// ConcurrentMap 线程安全的 map，支持并发读写
|
||
type ConcurrentMap struct {
|
||
mu sync.RWMutex
|
||
items map[any]any
|
||
}
|
||
|
||
// NewConcurrentMap 创建线程安全 map
|
||
func NewConcurrentMap() *ConcurrentMap {
|
||
return &ConcurrentMap{items: make(map[any]any)}
|
||
}
|
||
|
||
func (cm *ConcurrentMap) Set(k, v any) {
|
||
cm.mu.Lock()
|
||
defer cm.mu.Unlock()
|
||
cm.items[k] = v
|
||
}
|
||
|
||
func (cm *ConcurrentMap) Get(k any) (any, bool) {
|
||
cm.mu.RLock()
|
||
defer cm.mu.RUnlock()
|
||
v, ok := cm.items[k]
|
||
return v, ok
|
||
}
|
||
|
||
func (cm *ConcurrentMap) Delete(k any) {
|
||
cm.mu.Lock()
|
||
defer cm.mu.Unlock()
|
||
delete(cm.items, k)
|
||
}
|
||
|
||
func (cm *ConcurrentMap) Len() int {
|
||
cm.mu.RLock()
|
||
defer cm.mu.RUnlock()
|
||
return len(cm.items)
|
||
}
|
||
|
||
// Pipeline 函数式管道，链式处理数据流
|
||
type Pipeline struct {
|
||
stages []func(any) any
|
||
input chan any
|
||
output chan any
|
||
}
|
||
|
||
// NewPipeline 创建管道
|
||
func NewPipeline() *Pipeline {
|
||
return &Pipeline{
|
||
stages: make([]func(any) any, 0),
|
||
input: make(chan any, 256),
|
||
output: make(chan any, 256),
|
||
}
|
||
}
|
||
|
||
// AddStage 添加处理阶段
|
||
func (p *Pipeline) AddStage(fn func(any) any) *Pipeline {
|
||
p.stages = append(p.stages, fn)
|
||
return p
|
||
}
|
||
|
||
// Run 启动管道（阻塞，直到 input 关闭）
|
||
func (p *Pipeline) Run(items []any) []any {
|
||
go func() {
|
||
for _, item := range items {
|
||
p.input <- item
|
||
}
|
||
close(p.input)
|
||
}()
|
||
|
||
go p.process()
|
||
results := make([]any, 0)
|
||
for v := range p.output {
|
||
results = append(results, v)
|
||
}
|
||
return results
|
||
}
|
||
|
||
func (p *Pipeline) process() {
|
||
defer close(p.output)
|
||
for item := range p.input {
|
||
for _, stage := range p.stages {
|
||
item = stage(item)
|
||
}
|
||
p.output <- item
|
||
}
|
||
}
|
||
|
||
// GoCtx 便捷并发模式
|
||
type GoCtx struct {
|
||
wg sync.WaitGroup
|
||
errs sync.Map
|
||
results sync.Map
|
||
}
|
||
|
||
// Go 启动一个并发任务
|
||
func (gc *GoCtx) Go(id string, fn func() error) {
|
||
gc.wg.Add(1)
|
||
go func() {
|
||
defer gc.wg.Done()
|
||
if err := fn(); err != nil {
|
||
gc.errs.Store(id, err)
|
||
}
|
||
}()
|
||
}
|
||
|
||
// Wait 等待所有任务完成，返回错误列表
|
||
func (gc *GoCtx) Wait() []error {
|
||
gc.wg.Wait()
|
||
var errs []error
|
||
gc.errs.Range(func(_, v any) bool {
|
||
errs = append(errs, v.(error))
|
||
return true
|
||
})
|
||
return errs
|
||
}
|
||
|
||
// WaitWithTimeout 带超时的等待
|
||
func (gc *GoCtx) WaitWithTimeout(timeout time.Duration) []error {
|
||
done := make(chan struct{})
|
||
go func() {
|
||
gc.wg.Wait()
|
||
close(done)
|
||
}()
|
||
select {
|
||
case <-done:
|
||
var errs []error
|
||
gc.errs.Range(func(_, v any) bool {
|
||
errs = append(errs, v.(error))
|
||
return true
|
||
})
|
||
return errs
|
||
case <-time.After(timeout):
|
||
return []error{fmt.Errorf("并发任务超时: %v", timeout)}
|
||
}
|
||
}
|