监控与运维

inspector := asynq.NewInspector(asynq.RedisClientOpt{Addr: "localhost:6379"})
defer inspector.Close()

// 获取单个队列信息
qinfo, err := inspector.GetQueueInfo("default")
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Queue: %s\n", qinfo.Queue)
fmt.Printf("  Pending:    %d\n", qinfo.Pending)     // 等待执行
fmt.Printf("  Active:     %d\n", qinfo.Active)      // 正在执行
fmt.Printf("  Scheduled:  %d\n", qinfo.Scheduled)   // 已排期
fmt.Printf("  Retry:      %d\n", qinfo.Retry)       // 等待重试
fmt.Printf("  Archived:   %d\n", qinfo.Archived)    // 已归档
fmt.Printf("  Completed:  %d\n", qinfo.Completed)   // 已完成（未清理）
fmt.Printf("  Aggregating:%d\n", qinfo.Aggregating) // 等待聚合
fmt.Printf("  Paused:     %v\n", qinfo.Paused)      // 是否暂停
fmt.Printf("  Memory:     %d bytes\n", qinfo.MemoryUsage)
fmt.Printf("  Latency:    %v\n", qinfo.Latency)

// 获取所有队列信息
allQueues, err := inspector.Queues()

// 按 ID 查询任务详情
taskInfo, err := inspector.GetTaskInfo("default", "task-uuid-here")

// 列出活跃任务
activeTasks, err := inspector.ListActiveTasks("default",
    asynq.PageSize(50),
    asynq.Page(1),
)

// 列出等待中的任务
pendingTasks, err := inspector.ListPendingTasks("default",
    asynq.PageSize(50),
)

// 列出已排期任务
scheduledTasks, err := inspector.ListScheduledTasks("default",
    asynq.PageSize(50),
)

// 列出重试任务
retryTasks, err := inspector.ListRetryTasks("default",
    asynq.PageSize(50),
)

// 列出归档任务
archivedTasks, err := inspector.ListArchivedTasks("default",
    asynq.PageSize(50),
)

// 列出已完成任务
completedTasks, err := inspector.ListCompletedTasks("default",
    asynq.PageSize(50),
)

// 列出聚合中的任务
aggregatingTasks, err := inspector.ListAggregatingTasks("default",
    asynq.PageSize(50),
)

asynq.PageSize(n) // 每页数量，默认 20
asynq.Page(n)     // 页码，从 1 开始

// 取消一个等待中的任务
err := inspector.CancelTask("default", "task-id")

// 删除任务
err := inspector.DeleteTask("default", "task-id")

// 立即运行一个已排期的任务
err := inspector.RunTask("default", "task-id")

// 归档一个任务
err := inspector.ArchiveTask("default", "task-id")

// 重试一个已归档的任务
err := inspector.RetryTask("default", "task-id")

// 删除队列中的所有任务
deletedCount, err := inspector.DeleteAllPendingTasks("default")
deletedCount, err := inspector.DeleteAllScheduledTasks("default")
deletedCount, err := inspector.DeleteAllRetryTasks("default")
deletedCount, err := inspector.DeleteAllArchivedTasks("default")
deletedCount, err := inspector.DeleteAllCompletedTasks("default")

// 批量运行
runCount, err := inspector.RunAllScheduledTasks("default")
runCount, err := inspector.RunAllRetryTasks("default")
runCount, err := inspector.RunAllArchivedTasks("default")

// 批量归档
archiveCount, err := inspector.ArchiveAllPendingTasks("default")
archiveCount, err := inspector.ArchiveAllScheduledTasks("default")
archiveCount, err := inspector.ArchiveAllRetryTasks("default")

// 批量重试
retryCount, err := inspector.RetryAllArchivedTasks("default")

// 暂停队列
err := inspector.PauseQueue("low")

// 恢复队列
err := inspector.UnpauseQueue("low")

newPayload, _ := json.Marshal(UpdatedPayload{...})
err := inspector.UpdateTaskPayload("default", "task-id", newPayload)

// 获取系统概览
func GetSystemOverview(inspector *asynq.Inspector) (*SystemOverview, error) {
    queues, err := inspector.Queues()
    if err != nil {
        return nil, err
    }

    overview := &SystemOverview{}
    for _, q := range queues {
        info, err := inspector.GetQueueInfo(q)
        if err != nil {
            continue
        }
        overview.TotalPending += info.Pending
        overview.TotalActive += info.Active
        overview.TotalArchived += info.Archived
        overview.TotalRetry += info.Retry
    }
    return overview, nil
}

// 清理旧归档任务
func CleanOldArchivedTasks(inspector *asynq.Inspector, queue string, olderThan time.Duration) error {
    tasks, err := inspector.ListArchivedTasks(queue, asynq.PageSize(100))
    if err != nil {
        return err
    }
    for _, t := range tasks {
        if time.Since(t.LastFailedAt) > olderThan {
            inspector.DeleteTask(queue, t.ID)
        }
    }
    return nil
}

go install github.com/hibiken/asynq/tools/asynq@latest

# 查看所有队列统计
asynq stats --uri redis://localhost:6379/0

# 列出指定状态的任务
asynq ls pending    --queue=default
asynq ls active     --queue=default
asynq ls scheduled  --queue=default
asynq ls retry      --queue=default
asynq ls archived   --queue=default
asynq ls completed  --queue=default

# 暂停/恢复队列
asynq pause critical
asynq unpause critical

# 任务操作
asynq cancel --id=<task-id> --queue=default    # 取消任务
asynq delete --id=<task-id> --queue=default    # 删除任务
asynq run --id=<task-id> --queue=default       # 立即运行
asynq retry --id=<task-id> --queue=default     # 重试归档任务

# 批量操作
asynq delete-all pending   --queue=default
asynq run-all scheduled    --queue=default
asynq retry-all archived   --queue=default

# 清理已完成任务
asynq clean completed --queue=default --expiration=24h

# 单机
asynq stats --uri redis://localhost:6379/0

# 带密码
asynq stats --uri redis://:password@localhost:6379/0

# Sentinel
asynq stats --uri redis-sentinel://:password@sentinel1:26379,sentinel2:26379/mymaster/0

# Cluster
asynq stats --uri redis-cluster://:password@node1:7000,node2:7001,node3:7002/0

asynq dash --uri redis://localhost:6379/0

# 基础部署
docker run -d \
    --name asynqmon \
    -p 8080:8080 \
    hibiken/asynqmon \
    --redis-addr=host.docker.internal:6379

# 多 Redis 实例
docker run -d \
    --name asynqmon \
    -p 8080:8080 \
    hibiken/asynqmon \
    --redis-addr=redis-prod-1:6379 \
    --redis-addr=redis-prod-2:6379

version: '3.8'
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  asynqmon:
    image: hibiken/asynqmon
    ports:
      - "8080:8080"
    command: --redis-addr=redis:6379
    depends_on:
      - redis

asynq dash --uri redis://localhost:6380/0 --tls

asynq dash --uri redis://username:password@localhost:6379/0

import "github.com/hibiken/asynq/x/metrics"

srv := asynq.NewServer(redisOpt, asynq.Config{
    MetricsCollector: metrics.NewPrometheusCollector("asynq_"),
})

# 队列积压告警
asynq_queue_pending{queue="critical"} > 1000

# 归档任务增长告警
rate(asynq_tasks_archived_total[5m]) > 0.5

# 任务失败率告警
rate(asynq_tasks_failed_total[5m]) / rate(asynq_tasks_processed_total[5m]) > 0.1

srv := asynq.NewServer(redisOpt, asynq.Config{
    HealthCheckInterval: 15 * time.Second,
    HealthCheckFunc: func(err error) {
        if err != nil {
            healthStatus.Store(false)
            log.Printf("health check failed: %v", err)
        } else {
            healthStatus.Store(true)
        }
    },
})

var healthStatus atomic.Bool
healthStatus.Store(true)

http.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
    if healthStatus.Load() {
        w.WriteHeader(http.StatusOK)
        w.Write([]byte("OK"))
    } else {
        w.WriteHeader(http.StatusServiceUnavailable)
        w.Write([]byte("Unhealthy"))
    }
})

http.HandleFunc("/ready", func(w http.ResponseWriter, r *http.Request) {
    // 检查 Redis 连接
    if err := redisClient.Ping(r.Context()).Err(); err != nil {
        w.WriteHeader(http.StatusServiceUnavailable)
        w.Write([]byte("Redis unreachable"))
        return
    }
    w.WriteHeader(http.StatusOK)
    w.Write([]byte("Ready"))
})

livenessProbe:
  httpGet:
    path: /health
    port: 8080
  initialDelaySeconds: 10
  periodSeconds: 15

readinessProbe:
  httpGet:
    path: /ready
    port: 8080
  initialDelaySeconds: 5
  periodSeconds: 10

func CleanDeadLetters(inspector *asynq.Inspector, retentionDays int) {
    cutoff := time.Now().Add(-time.Duration(retentionDays) * 24 * time.Hour)

    queues, _ := inspector.Queues()
    for _, q := range queues {
        tasks, err := inspector.ListArchivedTasks(q, asynq.PageSize(100))
        if err != nil {
            continue
        }
        for _, t := range tasks {
            if t.LastFailedAt.Before(cutoff) {
                inspector.DeleteTask(q, t.ID)
                log.Printf("cleaned dead letter: %s (failed %v ago)", t.ID, time.Since(t.LastFailedAt))
            }
        }
    }
}

func MonitorQueueBacklog(inspector *asynq.Inspector, threshold int, callback func(string, int)) {
    queues, _ := inspector.Queues()
    for _, q := range queues {
        info, _ := inspector.GetQueueInfo(q)
        if info.Pending > threshold {
            callback(q, info.Pending)
        }
    }
}