repository·v3·Indexed 23 days ago
https://github.com/vmihailenco/taskqA Golang asynchronous task/job queue supporting multiple backends including Redis, SQS, IronMQ, and in-memory. It features automatic scaling, rate limiting, retries, and deduplication. The library provides tools for producing and consuming tasks, custom handler implementation via the Handler interface or NewHandler, and observability integration through an OpenTelemetryHook.
To run the SQS backend example, you must provide AWS credentials via environment variables. The example requires the following variables to be set in your environment:
AWS_ACCOUNT_IDAWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEYTo run the taskq example using a Redis backend, ensure a Redis Server is running on port :6379. You must run both a consumer and a producer to see the task queue in action.
The SQS example is split into two separate processes: a consumer to process tasks and a producer to enqueue them. Run them in separate terminal sessions.
The consumerConfig type is used to manage consumer settings, specifically the number of fetchers and workers, while tracking performance metrics through an embedded perfProfile.
Key performance metrics available via the perfProfile methods are:
TPS(): Transactions Per Second (throughput).ErrorRate(): The ratio of retries to total processed tasks.Timing(): The duration of the last processing window.Note that consumerConfig also maintains a Score which is used by the internal configRoulette to determine the optimal configuration based on throughput, error rate, and timing.
The HandleMessage(msg *Message) method is the primary way to process an incoming message through the registry.
msg.TaskName.HandleMessage calculates a retry delay based on the task's options or the error type (if the error implements the Delayer interface).TaskQ supports deduplication via the Name field. If multiple messages are added with the same Name, only one will be processed.
Use OnceInPeriod to automatically generate a unique Name based on the provided arguments and a time period. This ensures that a specific task with specific arguments is only queued once per given duration.
If no args are provided to OnceInPeriod, it uses the existing m.Args to generate the name.
// Ensures the message is added only once every 10 minutes
msg.OnceInPeriod(10 * time.Minute)
// Or with specific args to define the uniqueness
msg.OnceInPeriod(10 * time.Minute, "unique-key", 123)To start processing tasks from a queue, call the Start method on your queue instance. This method is blocking and will begin fetching and executing tasks based on the queue's configuration.
// Start consuming the queue.
if err := MainQueue.Start(context.Background()); err != nil {
log.Fatal(err)
}To add tasks to a queue, you must first create a Factory for your chosen backend (e.g., redisq.NewFactory()), register a Queue using RegisterQueue, and define tasks using RegisterTask. Once the queue is initialized, use the Add method on the queue instance to enqueue tasks. Tasks can be configured with arguments using the WithArgs method.
import (
"github.com/vmihailenco/taskq/v3"
"github.com/vmihailenco/taskq/v3/redisq"
)
// Create a queue factory.
var QueueFactory = redisq.NewFactory()
// Create a queue.
var MainQueue = QueueFactory.RegisterQueue(&taskq.QueueOptions{
Name: "api-worker",
Redis: Redis, // go-redis client
})
// Register a task.
var CountTask = taskq.RegisterTask(&taskq.TaskOptions{
Name: "counter",
Handler: func() error {
IncrLocalCounter()
return nil
},
})
ctx := context.Background()
// And start producing.
for {
// Call the task without any args.
err := MainQueue.Add(CountTask.WithArgs(ctx))
if err != nil {
panic(err)
}
time.Sleep(time.Second)
}The TaskOptions struct defines how a specific task type is handled, retried, and recovered. When creating TaskOptions, you must provide a Name.
The Handler and FallbackHandler can be one of three function signatures:
func()func(arg1, arg2 Type)*Message argument: func(*Message)Note: Handlers may optionally take a context.Context as the first argument and may optionally return an error. If a handler returns a non-nil error, the message processing fails and will be retried according to the backoff settings.
RetryLimit: Number of retries before the message is permanently failed and deleted. Defaults to 64.MinBackoff: Minimum time between retries. Defaults to 30s.MaxBackoff: Maximum time between retries. Defaults to 30m.FallbackHandler: A function called after all retries have failed.DeferFunc: An optional function used by the Consumer to recover from panics.Use QueueOptions to configure the behavior of a task queue. You must provide a Name. The Init() method should be called to apply default values for workers, fetchers, timeouts, and storage.
Key configuration behaviors:
MinNumWorker and MaxNumWorker control goroutine counts. If WorkerLimit is set, it overrides both to enforce a global concurrency limit.MaxNumFetcher controls how many goroutines fetch messages. ReservationSize determines how many messages a fetcher grabs at once.ReservationTimeout is when a reserved message is returned to the queue. WaitTimeout is the long-polling duration.PauseErrorsThreshold defines how many consecutive failures trigger a queue pause (set to -1 to disable or 0 for the default 100).RateLimit (using redis_rate.Limit) and an optional RateLimiter.Version() function to retrieve the current release version of the taskq library as a string.*Task is registered, use its WithArgs method to construct a *Message ready to be enqueued. WithArgs takes a context.Context and a variadic list of arguments that match the signature of your registered Handler.