使用celeriac监控Golang与Celery工作节点交互

celeriac是一个Golang库，用于与Celery工作节点交互并监控其状态。它允许Golang应用程序订阅Celery事件，获取任务状态和工作节点信息。

安装celeriac

go get github.com/svcodestream/celeriac

基本用法

1. 创建监控器

package main

import (
	"log"
	"time"

	"github.com/svcodestream/celeriac"
)

func main() {
	// 创建监控器
	monitor, err := celeriac.NewMonitor(
		"amqp://guest:guest@localhost:5672//", // RabbitMQ地址
		celeriac.MonitorBufferSize(100),      // 事件缓冲区大小
	)
	if err != nil {
		log.Fatalf("Failed to create monitor: %v", err)
	}
	defer monitor.Close()

	// 启动监控器
	go monitor.Start()

	// 处理事件
	for {
		select {
		case event := <-monitor.Events():
			handleEvent(event)
		case <-time.After(30 * time.Second):
			log.Println("No events received in the last 30 seconds")
		}
	}
}

func handleEvent(event celeriac.Event) {
	switch e := event.(type) {
	case *celeriac.WorkerOnlineEvent:
		log.Printf("Worker %s came online", e.Hostname)
	case *celeriac.WorkerOfflineEvent:
		log.Printf("Worker %s went offline", e.Hostname)
	case *celeriac.TaskReceivedEvent:
		log.Printf("Task %s received by worker %s", e.UUID, e.Hostname)
	case *celeriac.TaskStartedEvent:
		log.Printf("Task %s started on worker %s", e.UUID, e.Hostname)
	case *celeriac.TaskSucceededEvent:
		log.Printf("Task %s succeeded with result: %v", e.UUID, e.Result)
	case *celeriac.TaskFailedEvent:
		log.Printf("Task %s failed: %s", e.UUID, e.Exception)
	case *celeriac.TaskRetriedEvent:
		log.Printf("Task %s retried: %s", e.UUID, e.Exception)
	default:
		log.Printf("Received unknown event: %T", e)
	}
}

2. 获取工作节点信息

// 获取所有工作节点
workers := monitor.GetWorkers()
for _, worker := range workers {
	log.Printf("Worker: %s, Active: %v, Processed: %d", 
		worker.Hostname, worker.IsActive(), worker.TotalTasksProcessed())
}

// 获取特定工作节点
if worker, exists := monitor.GetWorker("worker1@host"); exists {
	log.Printf("Worker %s last seen at %v", worker.Hostname, worker.LastSeen())
}

3. 获取任务状态

// 获取所有已知任务
tasks := monitor.GetTasks()
for _, task := range tasks {
	log.Printf("Task %s: %s, Worker: %s", task.UUID, task.State, task.Worker)
}

// 获取特定任务
if task, exists := monitor.GetTask("task-uuid-123"); exists {
	log.Printf("Task %s state: %s", task.UUID, task.State)
}

高级功能

自定义事件处理

// 创建自定义事件处理器
type CustomEventHandler struct{}

func (h *CustomEventHandler) HandleEvent(event celeriac.Event) {
	switch e := event.(type) {
	case *celeriac.TaskSucceededEvent:
		// 只处理成功任务
		log.Printf("Custom handler: Task %s succeeded", e.UUID)
	}
}

// 注册处理器
monitor.AddEventHandler(&CustomEventHandler{})

过滤特定事件

// 只接收特定类型的事件
filteredEvents := monitor.FilterEvents(
	func(event celeriac.Event) bool {
		_, isTaskEvent := event.(celeriac.TaskEvent)
		return isTaskEvent
	},
)

go func() {
	for event := range filteredEvents {
		handleEvent(event)
	}
}()

最佳实践

1. 错误处理：始终检查监控器创建和事件处理中的错误
2. 资源清理：使用defer确保监控器正确关闭
3. 并发安全：celeriac是并发安全的，可以在多个goroutine中使用
4. 性能考虑：对于高负载系统，适当调整缓冲区大小
5. 重连逻辑：实现网络中断时的自动重连机制

注意事项

1. celeriac需要访问Celery使用的同一个RabbitMQ实例
2. Celery必须配置为发送事件(CELERY_SEND_EVENTS=True)
3. 监控器不会持久化事件，重启后会丢失历史数据

通过celeriac，Golang应用程序可以有效地监控Celery集群，实现任务跟踪、工作节点监控和系统集成等功能。