使用Ratus实现Golang RESTful异步任务队列服务

Ratus是一个轻量级的Golang异步任务队列服务框架，它提供了RESTful API接口，非常适合构建分布式任务处理系统。下面我将详细介绍如何使用Ratus实现异步任务队列服务。

Ratus核心概念

1. 任务(Task): 需要异步执行的工作单元
2. 承诺(Promise): 任务被消费者获取后的状态
3. 主题(Topic): 任务的分类/队列名称

安装Ratus

go get github.com/hyperonym/ratus

基本使用示例

1. 创建Ratus服务

package main

import (
	"context"
	"log"
	"net/http"

	"github.com/hyperonym/ratus"
	"github.com/hyperonym/ratus/engine/memory"
)

func main() {
	// 创建内存引擎实例(生产环境可用Redis或MongoDB引擎)
	engine := memory.New()

	// 创建Ratus服务实例
	service := ratus.New(engine)

	// 配置HTTP服务器
	server := &http.Server{
		Addr:    ":8080",
		Handler: service,
	}

	// 启动服务
	log.Println("Ratus service starting on :8080")
	if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
		log.Fatalf("Failed to start server: %v", err)
	}
}

2. 生产者代码示例

package main

import (
	"context"
	"log"
	"time"

	"github.com/hyperonym/ratus"
)

func produceTasks() {
	// 创建Ratus客户端
	client := ratus.NewClient("http://localhost:8080")

	// 创建任务
	task := &ratus.Task{
		ID:      "task-1",
		Topic:   "email-queue",
		Payload: "{\"to\":\"user@example.com\",\"subject\":\"Welcome\"}",
	}

	// 提交任务
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	created, err := client.CreateTask(ctx, task)
	if err != nil {
		log.Printf("Failed to create task: %v", err)
		return
	}

	log.Printf("Task created: %s", created.ID)
}

3. 消费者代码示例

package main

import (
	"context"
	"log"
	"time"

	"github.com/hyperonym/ratus"
)

func consumeTasks() {
	// 创建Ratus客户端
	client := ratus.NewClient("http://localhost:8080")

	for {
		// 获取任务
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()

		// 从"email-queue"主题获取任务
		promise, err := client.Poll(ctx, "email-queue", &ratus.PollRequest{
			Limit: 1,
		})
		if err != nil {
			log.Printf("Failed to poll task: %v", err)
			time.Sleep(5 * time.Second)
			continue
		}

		if len(promise.Tasks) == 0 {
			log.Println("No tasks available, waiting...")
			time.Sleep(5 * time.Second)
			continue
		}

		task := promise.Tasks[0]

		// 处理任务
		log.Printf("Processing task %s: %s", task.ID, task.Payload)
		// TODO: 实际业务逻辑处理

		// 标记任务完成
		_, err = client.Commit(ctx, &ratus.CommitRequest{
			ID:     task.ID,
			State:  ratus.TaskStateCompleted,
			Result: "{\"status\":\"success\"}",
		})
		if err != nil {
			log.Printf("Failed to commit task: %v", err)
		}
	}
}

高级功能

1. 延迟任务

// 创建延迟5分钟执行的任务
task := &ratus.Task{
	ID:      "delayed-task",
	Topic:   "delayed-queue",
	Payload: "{\"action\":\"reminder\"}",
	Scheduled: time.Now().Add(5 * time.Minute).Format(time.RFC3339),
}

2. 任务重试

// 创建带重试策略的任务
task := &ratus.Task{
	ID:      "retry-task",
	Topic:   "retry-queue",
	Payload: "{\"action\":\"process\"}",
	Policy: &ratus.Policy{
		RetryLimit: 3,  // 最大重试次数
		RetryDelay: "10s", // 重试间隔
	},
}

3. 使用Redis引擎

import "github.com/hyperonym/ratus/engine/redis"

func main() {
	// 创建Redis引擎
	engine, err := redis.New(&redis.Options{
		Addr:     "localhost:6379",
		Password: "", // 无密码
		DB:       0,  // 默认DB
	})
	if err != nil {
		log.Fatal(err)
	}

	// 创建Ratus服务
	service := ratus.New(engine)
	// ... 其余代码相同
}

最佳实践

1. 任务ID生成: 使用UUID或其他分布式唯一ID生成算法
2. 错误处理: 消费者应妥善处理错误并适当重试
3. 监控: 实现任务处理监控和告警机制
4. 幂等性: 确保任务处理逻辑是幂等的
5. 资源限制: 控制并发任务数量防止资源耗尽

RESTful API示例

Ratus提供了标准的RESTful接口:

• POST /tasks - 创建任务
• GET /topics/{topic}/tasks - 获取任务
• DELETE /tasks/{id} - 删除任务
• PATCH /tasks/{id} - 更新任务状态

通过以上代码和说明，您可以快速构建一个基于Ratus的异步任务队列服务。根据实际需求，您可以选择内存引擎(适合开发和测试)或Redis/MongoDB引擎(适合生产环境)。