Documentation
¶
Overview ¶
Package queue defines the queue abstraction with priority ordering and per-queue / per-tenant rate limiting.
Queues are named channels that group related jobs. Jobs carry a Queue field that determines which queue they belong to. The dispatcher polls the queues listed in [dispatch.Config.Queues] (default: ["default"]).
Per-Queue Configuration ¶
Use Config to set per-queue rate limits and concurrency caps:
queue.Config{
Name: "email",
MaxConcurrency: 5, // max 5 concurrent email jobs
RateLimit: 10, // max 10 jobs/s dequeued from this queue
RateBurst: 20, // allow bursts up to 20
}
Pass configs when building the engine:
engine.Build(d,
engine.WithQueueConfig(
queue.Config{Name: "critical", MaxConcurrency: 20},
queue.Config{Name: "bulk", RateLimit: 5, RateBurst: 10},
),
)
Manager ¶
Manager enforces per-queue and per-tenant limits at dequeue time. It uses a token-bucket rate limiter (golang.org/x/time/rate) and an active-count gate for concurrency limits.
m := queue.NewManager(configs...)
if m.Acquire(queueName, tenantID) {
defer m.Release(queueName, tenantID)
// process the job
}
Queues without a Config have no limits beyond the pool-wide concurrency.
Index ¶
- type Config
- type Manager
- func (m *Manager) Acquire(queue, tenantID string) bool
- func (m *Manager) ActiveCount(queue string) int
- func (m *Manager) QueueConfig(queue string) (Config, bool)
- func (m *Manager) QueueNames() []string
- func (m *Manager) Release(queue, tenantID string)
- func (m *Manager) SetQueueConfig(cfg Config)
- func (m *Manager) SetTenantConfig(cfg TenantConfig)
- func (m *Manager) TenantActiveCount(queue, tenantID string) int
- type TenantConfig
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// Name is the queue identifier (must match the job.Queue field).
Name string
// MaxConcurrency limits how many jobs from this queue may run
// simultaneously across the local worker pool. Zero means no
// queue-specific limit (pool-wide concurrency still applies).
MaxConcurrency int
// RateLimit is the maximum sustained jobs per second that may be
// dequeued from this queue. Zero disables rate limiting.
RateLimit float64
// RateBurst is the burst size for the token-bucket rate limiter.
// Defaults to 1 if RateLimit is set but RateBurst is zero.
RateBurst int
}
Config defines per-queue behaviour such as rate limiting and concurrency.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager controls per-queue and per-tenant rate limiting and concurrency. It is safe for concurrent use.
func NewManager ¶
NewManager creates a Manager with the given queue configurations. Queues not listed here have no limits.
func (*Manager) Acquire ¶
Acquire checks rate limits and concurrency for the given queue and tenant. If the job is allowed to proceed it increments the active counter and returns true. The caller MUST call Release when the job completes.
func (*Manager) ActiveCount ¶
ActiveCount returns the current number of active jobs for a queue.
func (*Manager) QueueConfig ¶
QueueConfig returns the configuration for the named queue and true, or a zero Config and false if the queue has no explicit configuration.
func (*Manager) QueueNames ¶
QueueNames returns the names of all configured queues.
func (*Manager) SetQueueConfig ¶
SetQueueConfig dynamically updates (or creates) a queue configuration.
func (*Manager) SetTenantConfig ¶
func (m *Manager) SetTenantConfig(cfg TenantConfig)
SetTenantConfig configures rate limits and concurrency for a specific tenant on a specific queue. Calling this multiple times for the same queue+tenant replaces the previous configuration.
func (*Manager) TenantActiveCount ¶
TenantActiveCount returns the current number of active jobs for a queue+tenant pair.
type TenantConfig ¶
type TenantConfig struct {
// QueueName is the queue this config applies to.
QueueName string
// TenantID is the tenant identifier (typically job.ScopeOrgID).
TenantID string
// RateLimit is the sustained jobs per second for this tenant.
RateLimit float64
// RateBurst is the burst size for the tenant's rate limiter.
RateBurst int
// MaxConcurrency limits simultaneous jobs for this tenant on this
// queue. Zero means no tenant-specific concurrency limit.
MaxConcurrency int
}
TenantConfig defines rate limits and concurrency for a specific tenant on a specific queue, identified by the job's ScopeOrgID.
Source Files