4 - быстро принимать входящие задания,

Условие

4 - быстро принимать входящие задания, 5 - отдавать результат задания по запросу 6 - если результата нет, возвращать соответствующий статус

/* Процессор - сервис, который выполняет долгую ресурсоемкую операцию. Он уже реализован Возвращаемые им ошибки связаны исключительно с некорректными входными данными, устойчивы и перезапрашивать их бесполезно */ type Processor interface { Process([]byte) ([]byte, error) }

/* Считаем, что у нас есть генератор уникальных идентификаторов (например, UUID). Генератор гарантирует, что коллизий не будет */ type ID string // func NewID() ID

/* Шедулер принимает от клиентов задачи, ставит в очередь и запускает в процессинг. Обеспечивает, что единовременно запущено не более threads методов process Дает возможность проверить статус задачи и получить результат не блокирует методы своего публичного интерфейса на процессинг */ type Scheduler struct { processor Processor }

func NewScheduler( prc Processor, threads int, ) *Scheduler { // todo initialization logic return &Scheduler{ processor: prc, } }

func (s *Scheduler) Queue(request []byte) ID { // todo implement return "" }

Ответ

Храните очередь и статусы под коротким mutex, а Process выполняйте только в фиксированном числе workers вне блокировки. Queue копирует вход, создаёт ID и быстро возвращает его.

Минимальный in-memory вариант с неограниченной очередью. Processor и NewID заданы условием; Processor должен допускать конкурентные вызовы.

type job struct { id ID; data []byte }
type Result struct { State string; Data []byte; Err error }
type Scheduler struct {
    processor Processor
    mu sync.Mutex
    cond *sync.Cond
    jobs []job
    results map[ID]Result
    closed bool
    wg sync.WaitGroup
}
func NewScheduler(p Processor, threads int) *Scheduler {
    if threads <= 0 { panic("threads must be positive") }
    s := &Scheduler{processor: p, results: make(map[ID]Result)}
    s.cond = sync.NewCond(&s.mu)
    s.wg.Add(threads)
    for i := 0; i < threads; i++ { go s.worker() }
    return s
}
func (s *Scheduler) Queue(request []byte) ID {
    data := append([]byte(nil), request...)
    s.mu.Lock()
    defer s.mu.Unlock()
    if s.closed { panic("scheduler closed") }
    id := NewID()
    s.results[id] = Result{State: "queued"}
    s.jobs = append(s.jobs, job{id, data})
    s.cond.Signal()
    return id
}
func (s *Scheduler) worker() {
    defer s.wg.Done()
    for {
        s.mu.Lock()
        for len(s.jobs) == 0 && !s.closed { s.cond.Wait() }
        if len(s.jobs) == 0 { s.mu.Unlock(); return }
        j := s.jobs[0]
        s.jobs[0] = job{}
        s.jobs = s.jobs[1:]
        if len(s.jobs) == 0 { s.jobs = nil }
        s.results[j.id] = Result{State: "running"}
        s.mu.Unlock()
        data, err := s.processor.Process(j.data)
        state := "done"
        if err != nil { state = "failed" }
        result := Result{state, append([]byte(nil), data...), err}
        s.mu.Lock()
        s.results[j.id] = result
        s.mu.Unlock()
    }
}
func (s *Scheduler) Status(id ID) (Result, bool) {
    s.mu.Lock()
    defer s.mu.Unlock()
    r, ok := s.results[id]
    r.Data = append([]byte(nil), r.Data...)
    return r, ok
}
func (s *Scheduler) Close() {
    s.mu.Lock()
    s.closed = true
    s.cond.Broadcast()
    s.mu.Unlock()
    s.wg.Wait()
}

Нужен импорт sync. После Close новые задачи запрещены, принятые дочитываются; ошибки Process не ретраятся. Вызовы Process обязаны завершаться без panic. Для production нужны лимит очереди, срок хранения результатов и durable-хранилище. Тогда Queue лучше изменить на (ID, error), чтобы явно сообщать переполнение и закрытие.