Senior
// Klientas perduoda objektą (Task) su duomenimis, skirtą užduočiai atlikti, // mūsų pavyzdyje naudosime tuščią struktūrą. // // Valdytojas gali apdoroti ne daugiau kaip N užduočių vienu metu, // ir ne daugiau kaip X užduočių gali būti eilėje. // Jei nėra vietos eilėje, iš karto grąžiname klaidą klientui. // // Užduotis imama į apdorojimą, jei yra laisvų apdorotojų. // Simuliuojame apdorojimo trukmę naudodami time.Sleep(5*time.Second). // Kai viena užduotis baigiasi, paimame kitą iš eilės. // Jei eilė tuščia, laukiame naujų užduočių iš klientų. // // Su žvaigždute: papildyti Task struktūrą ir gauti užduočių būsenas per atskirą užklausą. package scheduler type Scheduler interface { AddTask(t Task) error Close() } type scheduler struct { queue chan Task wg sync.WaitGroup } func New(workers, queue int) Scheduler { // laukiama įgyvendinimo } type Task interface { Do() }
sobes.tech AI
Atsakymas iš AI
Ниже пример реализации Scheduler в Go, который ограничивает количество одновременно обрабатываемых задач (workers) и размер очереди (queue). Если очередь заполнена, возвращается ошибка. Задачи обрабатываются параллельно, после завершения задачи берётся следующая из очереди.
package scheduler
import (
"errors"
"sync"
"time"
)
type Task interface {
Do()
}
type scheduler struct {
queue chan Task
workers int
wg sync.WaitGroup
quit chan struct{}
}
func New(workers, queue int) Scheduler {
s := &scheduler{
queue: make(chan Task, queue),
workers: workers,
quit: make(chan struct{}),
}
for i := 0; i < workers; i++ {
go s.worker()
}
return s
}
func (s *scheduler) worker() {
for {
select {
case task := <-s.queue:
s.wg.Add(1)
task.Do()
s.wg.Done()
case <-s.quit:
return
}
}
}
func (s *scheduler) AddTask(t Task) error {
select {
case s.queue <- t:
return nil
default:
return errors.New("queue is full")
}
}
func (s *scheduler) Close() {
close(s.quit)
// Ждём завершения всех задач
s.wg.Wait()
}
// Пример реализации Task
// type MyTask struct{}
// func (t MyTask) Do() {
// time.Sleep(5 * time.Second) // имитация работы
// }
Пояснения:
queue— буферизированный канал для задач, размеромqueue.- Если канал заполнен,
AddTaskвозвращает ошибку. workers— количество горутин-обработчиков.- Каждая горутина читает задачи из канала и выполняет
Do(). Closeостанавливает обработчики и ждёт завершения всех задач.
Для реализации статусов задач можно расширить Task структурой с ID и статусом, а также добавить мапу для хранения статусов с синхронизацией.