belvedersky/queue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Change package path

Browse Source
This commit is contained in:
Кобелев Андрей Андреевич
2025-10-14 12:53:46 +05:00
parent e3c9643d05
commit 557fb40268
8 changed files with 20 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@@ -1,7 +1,11 @@
# queue # queue
[![Go Reference](https://pkg.go.dev/badge/git.belvedersky.ru/queue.svg)](https://pkg.go.dev/git.belvedersky.ru/queue)
Simple and lightweight queue implementation in Go with optional Dead Letter Queue (DLQ) support. Simple and lightweight queue implementation in Go with optional Dead Letter Queue (DLQ) support.
Repository: Repository:
https://git.belvedersky.ru/belvedersky/queue https://git.belvedersky.ru/belvedersky/queue

2
bench_test.go
View File

@@ -6,7 +6,7 @@ import (
"testing" "testing"
"time" "time"
"git.belvedersky.ru/belvedersky/queue" "git.belvedersky.ru/queue"
) )
type benchTask struct { type benchTask struct {

2
examples/example_basic_test.go
View File

@@ -5,7 +5,7 @@ import (
"fmt" "fmt"
"time" "time"
"git.belvedersky.ru/belvedersky/queue" "git.belvedersky.ru/queue"
) )
// Example_basic демонстрирует базовое использование очереди. // Example_basic демонстрирует базовое использование очереди.

2
examples/example_dlq_test.go
View File

@@ -6,7 +6,7 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"git.belvedersky.ru/belvedersky/queue" "git.belvedersky.ru/queue"
) )
// Example_DLQ показывает, как использовать Dead Letter Queue // Example_DLQ показывает, как использовать Dead Letter Queue

2
examples/example_parallel_test.go
View File

@@ -6,7 +6,7 @@ import (
"sync" "sync"
"time" "time"
"git.belvedersky.ru/belvedersky/queue" "git.belvedersky.ru/queue"
) )
// Example_parallel демонстрирует работу нескольких воркеров. // Example_parallel демонстрирует работу нескольких воркеров.

2
go.mod
View File

@@ -1,3 +1,3 @@
module git.belvedersky.ru/belvedersky/queue module git.belvedersky.ru/queue
go 1.25.1 go 1.25.1

21
queue.go
View File

@@ -11,17 +11,17 @@ import (
type QueueHooks struct { type QueueHooks struct {
OnProduced func() // вызывается при успешном добавлении элемента OnProduced func() // вызывается при успешном добавлении элемента
OnHandled func() // вызывается при успешной обработке OnHandled func() // вызывается при успешной обработке
OnDropped func(interface{}) // вызывается при пропуске (cancel/timeout) OnDropped func(any) // вызывается при пропуске (cancel/timeout)
} }
// Queue — потокобезопасная обобщённая FIFO очередь с DLQ, параллельной обработкой // Queue — потокобезопасная обобщённая FIFO очередь с DLQ, параллельной обработкой
// и корректным завершением без data race (даже под -race). // и корректным завершением без data race
type Queue[T any] struct { type Queue[T any] struct {
ch chan T ch chan T
capacity int capacity int
mu sync.RWMutex // защищает: handlers/errorFunc/dlq/hooks/closed mu sync.RWMutex // защищает: handlers/errorFunc/dlq/hooks/closed
sendMu sync.RWMutex // синхронизирует ВСЕ записи в канал и его закрытие sendMu sync.RWMutex // синхронизирует записи в канал и его закрытие
handlers []func(T) handlers []func(T)
errorFunc func(error) errorFunc func(error)
@@ -47,7 +47,6 @@ func NewQueue[T any](capacity int) *Queue[T] {
} }
// ---------- Конфигурация ---------- // ---------- Конфигурация ----------
func (q *Queue[T]) Register(handler func(T)) { func (q *Queue[T]) Register(handler func(T)) {
if handler == nil { if handler == nil {
panic("handler cannot be nil") panic("handler cannot be nil")
@@ -57,24 +56,28 @@ func (q *Queue[T]) Register(handler func(T)) {
q.mu.Unlock() q.mu.Unlock()
} }
// ---------- При ошибке ----------
func (q *Queue[T]) OnError(fn func(error)) { func (q *Queue[T]) OnError(fn func(error)) {
q.mu.Lock() q.mu.Lock()
q.errorFunc = fn q.errorFunc = fn
q.mu.Unlock() q.mu.Unlock()
} }
// ---------- Установка DLQ ----------
func (q *Queue[T]) SetDLQ(dlq *Queue[T]) { func (q *Queue[T]) SetDLQ(dlq *Queue[T]) {
q.mu.Lock() q.mu.Lock()
q.dlq = dlq q.dlq = dlq
q.mu.Unlock() q.mu.Unlock()
} }
// ---------- Установка Хуков ----------
func (q *Queue[T]) SetHooks(h QueueHooks) { func (q *Queue[T]) SetHooks(h QueueHooks) {
q.mu.Lock() q.mu.Lock()
q.hooks = h q.hooks = h
q.mu.Unlock() q.mu.Unlock()
} }
// ---------- Установка RateLimit ----------
func (q *Queue[T]) SetRateLimit(rps int) { func (q *Queue[T]) SetRateLimit(rps int) {
if rps > 0 { if rps > 0 {
q.rateLimit = time.Tick(time.Second / time.Duration(rps)) q.rateLimit = time.Tick(time.Second / time.Duration(rps))
@@ -82,7 +85,6 @@ func (q *Queue[T]) SetRateLimit(rps int) {
} }
// ---------- Запуск обработчиков ---------- // ---------- Запуск обработчиков ----------
func (q *Queue[T]) HandleParallel(ctx context.Context, workers int) { func (q *Queue[T]) HandleParallel(ctx context.Context, workers int) {
if workers <= 0 { if workers <= 0 {
workers = 1 workers = 1
@@ -110,12 +112,10 @@ func (q *Queue[T]) HandleParallel(ctx context.Context, workers int) {
} }
// ---------- Отправка в канал (синхронизированная) ---------- // ---------- Отправка в канал (синхронизированная) ----------
// sendBlocking — блокирующая отправка (используется Produce). // sendBlocking — блокирующая отправка (используется Produce).
// Берём RLock на всё время операции записи, чтобы Close() (Lock) не мог выполнить close(q.ch) // Берём RLock на всё время операции записи, чтобы Close() (Lock) не мог выполнить close(q.ch)
// одновременно с записью — это устраняет data race.
func (q *Queue[T]) sendBlocking(item T) bool { func (q *Queue[T]) sendBlocking(item T) bool {
// Быстрая проверка закрытия под RLock (необязательна, но экономит работу) // Быстрая проверка закрытия под RLock
q.sendMu.RLock() q.sendMu.RLock()
q.mu.RLock() q.mu.RLock()
closed := q.closed closed := q.closed
@@ -167,7 +167,6 @@ func (q *Queue[T]) sendNonBlocking(item T) bool {
} }
// ---------- Паблик-API продюсеров ---------- // ---------- Паблик-API продюсеров ----------
// Produce — блокирующая постановка. Если очередь закрыта — элемент игнорируется. // Produce — блокирующая постановка. Если очередь закрыта — элемент игнорируется.
func (q *Queue[T]) Produce(item T) { func (q *Queue[T]) Produce(item T) {
_ = q.sendBlocking(item) _ = q.sendBlocking(item)
@@ -212,7 +211,6 @@ func (q *Queue[T]) TryProduce(item T, timeout time.Duration) bool {
} }
// ---------- Завершение ---------- // ---------- Завершение ----------
func (q *Queue[T]) Shutdown(ctx context.Context) error { func (q *Queue[T]) Shutdown(ctx context.Context) error {
q.Close() q.Close()
done := make(chan struct{}) done := make(chan struct{})
@@ -245,7 +243,6 @@ func (q *Queue[T]) Close() {
func (q *Queue[T]) Wait() { q.wg.Wait() } func (q *Queue[T]) Wait() { q.wg.Wait() }
// ---------- Инфо-методы ---------- // ---------- Инфо-методы ----------
func (q *Queue[T]) Len() int { return len(q.ch) } func (q *Queue[T]) Len() int { return len(q.ch) }
func (q *Queue[T]) Cap() int { return q.capacity } func (q *Queue[T]) Cap() int { return q.capacity }
func (q *Queue[T]) IsEmpty() bool { return len(q.ch) == 0 } func (q *Queue[T]) IsEmpty() bool { return len(q.ch) == 0 }
@@ -261,7 +258,6 @@ func (q *Queue[T]) String() string {
} }
// ---------- Внутренняя обработка ---------- // ---------- Внутренняя обработка ----------
func (q *Queue[T]) processItem(item T) { func (q *Queue[T]) processItem(item T) {
q.mu.RLock() q.mu.RLock()
handlers := append([]func(T){}, q.handlers...) handlers := append([]func(T){}, q.handlers...)
@@ -269,7 +265,6 @@ func (q *Queue[T]) processItem(item T) {
dlq := q.dlq dlq := q.dlq
hooks := q.hooks hooks := q.hooks
q.mu.RUnlock() q.mu.RUnlock()
for _, h := range handlers { for _, h := range handlers {
func() { func() {
defer func() { defer func() {

2
queue_test.go
View File

@@ -7,7 +7,7 @@ import (
"testing" "testing"
"time" "time"
"git.belvedersky.ru/belvedersky/queue" "git.belvedersky.ru/queue"
) )
type Task struct { type Task struct {