Data Science. SQL hub

🛠️ История создания “storage-agnostic” message queue


Контекст:
Работая на Go, автор вдохновился инструментами из Node.js экосистемы (BullMQ, RabbitMQ) и захотел сделать что-то похожее, но с нуля, без зависимостей. Так родилась идея — сначала он создал Gocq (Go Concurrent Queue): простую concurrent-очередь, работающую через каналы.

⚡ Основная проблема

Gocq отлично работал в памяти, но не поддерживал устойчивое хранение задач.
Автор задумался: а можно ли сделать очередь, не зависящую от конкретного хранилища — так, чтобы её можно было подключить к Redis, SQLite или совсем без них?

🧱 Как это реализовано в VarMQ

После рефакторинга Gocq был разделён на два компонента:
1) Worker pool — пул воркеров, обрабатывающих задачи
2) Queue interface — абстракция над очередью, не зависящая от реализации

Теперь воркер просто берёт задачи из очереди, не зная, где они хранятся.

🧠 Пример использования

• In-memory очередь:

w := varmq.NewVoidWorker(func(data any) {
   // обработка задачи
}, 2)
q := w.BindQueue()

• С SQLite-поддержкой:

import "github.com/goptics/sqliteq"

db := sqliteq.New("test.db")
pq, _ := db.NewQueue("orders")
q := w.WithPersistentQueue(pq)

• С Redis (для распределённой обработки):

import "github.com/goptics/redisq"

rdb := redisq.New("redis://localhost:6379")
pq := rdb.NewDistributedQueue("transactions")
q := w.WithDistributedQueue(pq)

В итоге воркер обрабатывает задачи одинаково — независимо от хранилища.

✅ Почему это круто

• Гибкость: адаптеры позволяют легко менять хранилище без правок воркера
• Минимальные зависимости: в яд

📌 Читать

www.tgoop.com/sqlhub/1887

5.28K viewsMay 22 at 17:03

🛠️ История создания “storage-agnostic” message queue


Контекст:
Работая на Go, автор вдохновился инструментами из Node.js экосистемы (BullMQ, RabbitMQ) и захотел сделать что-то похожее, но с нуля, без зависимостей. Так родилась идея — сначала он создал Gocq (Go Concurrent Queue): простую concurrent-очередь, работающую через каналы.

⚡ Основная проблема

Gocq отлично работал в памяти, но не поддерживал устойчивое хранение задач.
Автор задумался: а можно ли сделать очередь, не зависящую от конкретного хранилища — так, чтобы её можно было подключить к Redis, SQLite или совсем без них?

🧱 Как это реализовано в VarMQ

После рефакторинга Gocq был разделён на два компонента:
1) Worker pool — пул воркеров, обрабатывающих задачи
2) Queue interface — абстракция над очередью, не зависящая от реализации

Теперь воркер просто берёт задачи из очереди, не зная, где они хранятся.

🧠 Пример использования

• In-memory очередь:

w := varmq.NewVoidWorker(func(data any) {
   // обработка задачи
}, 2)
q := w.BindQueue()

import "github.com/goptics/sqliteq"

db := sqliteq.New("test.db")
pq, _ := db.NewQueue("orders")
q := w.WithPersistentQueue(pq)

import "github.com/goptics/redisq"

rdb := redisq.New("redis://localhost:6379")
pq := rdb.NewDistributedQueue("transactions")
q := w.WithDistributedQueue(pq)

BY Data Science. SQL hub