Стратегия очереди задач Golang

Go
Стратегия очереди задач Golang — читайте «ОЧЕРЕДИ РАБОТЫ В GO» — artong0416 — Blog Park

artong0416

Стратегия очереди задач Golang — читайте «ОЧЕРЕДИ ЗАДАНИЙ В GO»

Golang отлично справляется с асинхронной обработкой. Потому что горутины и каналы — очень простые в использовании и эффективные средства асинхронной обработки. Давайте посмотрим на простую очередь задач Golang.

Очередь задач «не очередь задач»

Иногда нам нужно выполнить асинхронную обработку, но не нужна пара задач.Мы можем использовать Golang для очень простой реализации такой задачи. следующим образом:

go process(job)скопировать код

Это действительно отличный выбор во многих сценариях, таких как обработка HTTP-запроса и ожидание результата. Однако в некоторых относительно сложных сценариях с высоким уровнем параллелизма вы не можете просто использовать этот метод для реализации асинхронной обработки. В это время вам нужна очередь для управления задачами, которые необходимо обработать, и обрабатывать эти задачи в определенном порядке.

Простейшая очередь задач

Далее рассмотрим простейшую очередь задач и рабочую модель.

func worker(jobChan <-chan Job) {
    for job := range jobChan {
        process(job)
    }
}

// make a channel with a capacity of 100.
jobChan := make(chan Job, 100)

// start the worker
go worker(jobChan)

// enqueue a job
jobChan <- jobскопировать код

Код создает канал объектов Job емкостью 100. Затем запустите рабочую сопрограмму, чтобы удалить задачу из канала и выполнить ее. Операция постановки задачи в очередь состоит в том, чтобы поместить объект Job в канал задачи.

Хотя выше всего несколько коротких строк кода, проделана большая работа. Мы реализовали простую потокобезопасную параллельную надежную очередь задач.

Ограничение

В приведенном выше примере мы инициализировали канал задач емкостью 100.

// make a channel with a capacity of 100.
jobChan := make(chan Job, 100)скопировать код

Это означает, что операция постановки задачи в очередь очень проста:

// enqueue a job
jobChan <- jobскопировать код

这样一来，当 job channel 中已经放入 100 个任务的时候，入队操作将会阻塞，直至有任务被工作者处理完成。这通常不是一个好的现象，因为我们通常不希望程序出现阻塞等待。这时候，我们通常希望有一个超时机制来告诉服务调用方，当前服务忙，稍后重试。我之前的博文--我读《通过Go来处理每分钟达百万的数据请求》介绍过类似的限流策略。这里方法类似，就是当队列满的时候，返回503，告诉调用方服务忙。 код показывает, как показано ниже:

// TryEnqueue tries to enqueue a job to the given job channel. Returns true if
// the operation was successful, and false if enqueuing would not have been
// possible without blocking. Job is not enqueued in the latter case.
func TryEnqueue(job Job, jobChan <-chan Job) bool {
    select {
    case jobChan <- job:
        return true
    default:
        return false
    }
}скопировать код

Таким образом, когда мы пытаемся присоединиться к очереди, если ему не удается присоединиться к очереди, возвращаем false , чтобы мы могли обработать это возвращаемое значение следующим образом:

if !TryEnqueue(job, chan) {
    http.Error(w, "max capacity reached", 503)
    return
}скопировать код

Таким образом, операция ограничения тока реализуется просто. Когда jobChan заполнен, программа вернется к умолчанию и вернет false, чтобы проинформировать вызывающую сторону о текущей ситуации на сервере.

близкий работник

Вышеуказанными шагами можно решить текущий лимит, поэтому давайте рассмотрим, как изящно закрыть воркера? Допустим, мы решили не вставлять задачи в очередь задач, а хотим завершить выполнение всех поставленных в очередь задач, сделать это можно очень просто:

close(jobChan)скопировать код

Да, с помощью этой строки кода мы можем запретить очереди задач получать новые задачи (вы все еще можете читать задания из канала).Если мы хотим выполнить существующие задачи в очереди, нам нужно только:

for job := range jobChan {...}скопировать код

Все задания, которые были поставлены в очередь, будут забраны вокером и выполнены в обычном режиме. Однако на самом деле на этом пути есть проблема, то есть основная ассоциация не будет ждать, пока воркер закончит работу, а затем выйдет. Неизвестно, когда рабочий союз сможет завершить вышеуказанные задачи. Рабочий пример выглядит следующим образом:

package main

import (
    "fmt"
)

var jobChan chan int

func worker(jobChan <- chan int)  {
    for job := range jobChan{
        fmt.Printf("执行任务 %d \n", job)
    }
}

func main() {
    jobChan = make(chan int, 100)
    //入队
    for i := 1; i <= 10; i++{
        jobChan <- i
    }
    
    close(jobChan)
    go worker(jobChan)

}скопировать код

Обнаружено, что рабочий процесс не может гарантировать, что он завершится после выполнения задания в канале. Так как же решить эту проблему?

Подождите, пока рабочий закончит выполнение

Используйте sysc.WaitGroup:

package main

import (
    "fmt"
    "sync"
)

var jobChan chan int
var wg sync.WaitGroup

func worker(jobChan <- chan int)  {
    defer wg.Done()
    for job := range jobChan{
        fmt.Printf("执行任务 %d \n", job)
    }
}

func main() {
    jobChan = make(chan int, 100)
    //入队
    for i := 1; i <= 10; i++{
        jobChan <- i
    }

    wg.Add(1)
    close(jobChan)

    go worker(jobChan)
    wg.Wait()
}скопировать код

Используя этот метод синхронизации между сопрограммами, сопрограмма будет ждать, пока рабочий закончит выполнение задания, прежде чем выйти. результат операции:

执行任务 1 
执行任务 2 
执行任务 3 
执行任务 4 
执行任务 5 
执行任务 6 
执行任务 7 
执行任务 8 
执行任务 9 
执行任务 10 

Process finished with exit code 0скопировать код

Это идеально? При разработке функции, чтобы предотвратить приостановку анимации сопрограммы, мы должны установить тайм-аут для сопрограммы.

установка времени ожидания

В приведенном выше примере wg.Wait() будет ждать, пока не будет вызвана wg.Done(). Но если эта операция приостановлена ​​и не может быть вызвана, она будет ждать вечно. Это то, что мы не хотим видеть, поэтому мы можем установить для этого тайм-аут. Методы, как показано ниже:

package main

import (
    "fmt"
    "sync"
    "time"
)

var jobChan chan int
var wg sync.WaitGroup

func worker(jobChan <-chan int) {
    defer wg.Done()
    for job := range jobChan {
        fmt.Printf("执行任务 %d \n", job)
        time.Sleep(1 * time.Second)
    }
}

func main() {
    jobChan = make(chan int, 100)
    //入队
    for i := 1; i <= 10; i++ {
        jobChan <- i
    }

    wg.Add(1)
    close(jobChan)

    go worker(jobChan)
    res := WaitTimeout(&wg, 5*time.Second)
    if res {
        fmt.Println("执行完成退出")
    } else {
        fmt.Println("执行超时退出")
    }
}

//超时机制
func WaitTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
    ch := make(chan struct{})
    go func() {
        wg.Wait()
        close(ch)
    }()
    select {
    case <-ch:
        return true
    case <-time.After(timeout):
        return false
    }
}скопировать код

Результат выполнения следующий:

执行任务 1 
执行任务 2 
执行任务 3 
执行任务 4 
执行任务 5 
执行超时退出

Process finished with exit code 0скопировать код

Таким образом, вступает в силу тайм-аут 5 с, и хотя не все задачи выполняются, они также завершатся из-за тайм-аута.

Иногда мы хотим, чтобы воркер сбрасывал выполняемую работу, то есть операцию отмены, как с этим быть?

Cancel Worker

Мы можем сделать это с помощью context.Context. следующим образом:

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

var jobChan chan int
var ctx context.Context
var cancel context.CancelFunc

func worker(jobChan <-chan int, ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return
        case job := <-jobChan:
            fmt.Printf("执行任务 %d \n", job)
            time.Sleep(1 * time.Second)
        }
    }
}

func main() {
    jobChan = make(chan int, 100)
    //带有取消功能的 contex
    ctx, cancel = context.WithCancel(context.Background())
    //入队
    for i := 1; i <= 10; i++ {
        jobChan <- i
    }

    close(jobChan)

    go worker(jobChan, ctx)
    time.Sleep(2 * time.Second)
    //調用cancel
    cancel()
}скопировать код

Результат выглядит следующим образом:

执行任务 1 
执行任务 2 

Process finished with exit code 0скопировать код

Видно, что после того, как мы подождали 2 с, мы активно вызвали операцию отмены, и сопрограмма waker активно вышла.

Это реализация операции отмены с помощью пакета контекста, которая по сути является операцией прослушивания канала, поэтому можем ли мы реализовать операцию отмены без использования контекста?

Чтобы реализовать отмену без использования механизма тайм-аута контекста:

package main

import (
    "fmt"
    "time"
)

var jobChan chan int

func worker(jobChan <-chan int, cancelChan <-chan struct{}) {
    for {
        select {
        case <-cancelChan:
            return
        case job := <-jobChan:
            fmt.Printf("执行任务 %d \n", job)
            time.Sleep(1 * time.Second)
        }
    }
}

func main() {
    jobChan = make(chan int, 100)
    //通过chan 取消操作
    cancelChan := make(chan struct{})
    //入队
    for i := 1; i <= 10; i++ {
        jobChan <- i
    }

    close(jobChan)

    go worker(jobChan, cancelChan)
    time.Sleep(2 * time.Second)
    //关闭chan
    close(cancelChan)
}скопировать код

Таким образом реализуем операцию отмены с сигналом, закрывающим чан. Причина в том, что чтение небуферизованного чана будет блокироваться, при его закрытии он может быть прочитан пустым, поэтому будет выполнен возврат в select.

Суммировать

Как обычно, в этой статье представлены некоторые методы синхронизации и связи между эквилатерациями Голанга, простейшая реализация очереди задач. О реализации пула воркеров я также писал в других постах в блоге, а здесь не писал. Еще в этой статье есть инструментальный код, и его можно заимствовать при написании, например тайм-аут, отмена, операция Чана и тому подобное.

posted on 2017-11-23 09:58 artong0416Читать(107) Комментарии(2)редактировать собирать

Комментарий

#1-й этаж 2017-11-23 10:26 lanfuyi9774  

хорошая статьяПоддержка(0)против (0)   

#2-ой этаж[Арендодатель] 38481122017/11/23 10:28:33 2017-11-23 10:28artong0416  

@ lanfuyi9774
спасибо за поддержкуПоддержка(0)против (0) http://pic.cnblogs.com/face/550061/20170824150610.png    обновить комментарийобновить страницуBack to topЗарегистрированные пользователи могут оставлять комментарии только после авторизации.Авторизоватьсяилирегистр,доступдомашняя страница сайта.[Рекомендуется] 500 000 строк исходного кода VC++: крупномасштабное промышленное управление конфигурацией, моделирование энергопотребления САПР и библиотека исходного кода ГИС
【Новости】Платформа H3 BPM полностью запущена
[Рекомендуется] Быстрый старт Vue.js 2.x, множество эффективных практических примеров
[Рекомендация] Создайте апплет WeChat, начиная с 3 юаней для нового пользовательского интерфейса.
葡萄城1114 Последние ИТ-новости:
· Почему некоторые жители Гонконга не понимают Alipay?
· Онлайн-шрифты Microsoft Store: будущая Windows 10 сможет легко переключать шрифты
· Новый раунд финансирования Lyft увеличился до 1,5 млрд долларов, оценка выросла до 11,5 млрд долларов.
· Медицинская страховка WeChat оплачивает 100% и получает красный конверт наличными: до 200 юаней
· Масаёси Сон инвестирует в индийское программное обеспечение для такси Венчурный капитал: он хочет контролировать мировой рынок такси
» больше новостей... 阿里云1113 Последние статьи базы знаний:
· Посмотрите внутри и снаружи двери, чтобы увидеть набор
· Дорога к простоте
· Что касается программирования, эффективны ли ваши упражнения?
· 3+10 привычек, улучшающих качество жизни программиста
· 10 принципов кодирования НАСА
» Другие статьи базы знаний...

навигация

< ноябрь 2017 г. >
день один два три Четыре Пятерки шесть
29 30 31 1 2 3 4
5 6 7 8 9 10 11
12 13 14 15 16 17 18
19 20 21 22 23 24 25
26 27 28 29 30 1 2
3 4 5 6 7 8 9

объявление

Никнейм:artong0416
Возраст сада:4 года и 4 месяца
вентилятор:3
Сфокусируйся на:0 +Добавить подписку

статистика

  • Эссе - 15
  • Статья - 2
  • Комментарии - 18

поиск

 

Самая используемая ссылка

Мои теги

классификация эссе

Архив эссе

последний комментарий

Читать таблицу лидеров

Обзор лидеров

Рекомендуемая таблица лидеров

Powered by:
Блог Парк
Copyright © artong0416
Категории