golang이 지연된 작업을 구현하는 방법에 대한 간략한 분석-Golang-php.cn

golang에서 지연된 작업을 구현하는 방법은 무엇입니까? 다음 글은 golang을 기반으로 한 일련의 지연된 작업 솔루션을 여러분과 공유할 것입니다. 이것이 여러분에게 도움이 되기를 바랍니다!

golang이 지연된 작업을 구현하는 방법에 대한 간략한 분석

실제 비즈니스 시나리오에서는 때때로 지연 요구 사항이 발생할 수 있습니다. 예를 들어 전자 상거래 플랫폼에서 운영자가 관리 백그라운드에 제품을 추가한 후 프런트 데스크에 즉시 표시할 필요가 없습니다. , 그러나 특정 시점에 표시됩니다.

물론 우리는 이 문제를 해결하기 위한 많은 아이디어를 가지고 있습니다. 예를 들어, 출시할 제품 정보를 db에 추가한 후 예약된 작업을 통해 데이터 테이블을 폴링하여 현재 시점에 출시된 제품을 쿼리하는 또 다른 예는 모든 제품 정보를 redis에 추가하고 이를 완료하는 것입니다. SortSet 속성을 통해 작동합니다. 최종 선택은 당사의 비즈니스 시나리오와 운영 환경에 따라 다릅니다.

여기서는 golang을 기반으로 한 지연 작업 솔루션 세트를 여러분과 공유하고 싶습니다.

lGolang 파이프라인의 유연한 적용을 수확할 수 있습니다.

마인드 맵 구현에 Golang 슬라이싱 요소
모두에게 일반적인 인상을 주기 위해 개요를 나열하겠습니다. 아래 텍스트.

구현 아이디어

우리 모두는 모든 종류의 대기열이 실제로 생산자와 소비자라는 두 부분으로 구성되어 있다는 것을 알고 있습니다. 지연된 작업에는 일반 대기열에 비해 추가 지연 기능이 있습니다.

1. 생산자

golang이 지연된 작업을 구현하는 방법에 대한 간략한 분석 생산자의 관점에서 사용자가 작업을 푸시하면 지연된 실행 시간 값이 전달됩니다. 이 작업이 예정된 시간에 실행되도록 하려면 이 작업을 일정 시간 동안 메모리에 저장해야 하며 시간은 1차원적이고 증가합니다. 그렇다면 이를 저장하기 위해 어떤 데이터 구조를 사용해야 할까요?

(1) 지도 중 하나를 선택하세요. 맵이 무질서하고 실행 시간에 따라 정렬할 수 없기 때문에 수행된 작업이 현재 시점에서 실행되어야 하는지 여부를 보장할 수 없으므로 이 옵션은 제외됩니다. (2) 선택 2: 채널. 실제로 채널은 때때로 대기열로 간주될 수 있습니다. 그러나 채널의 출력과 입력은 "선입선출" 원칙을 엄격하게 따릅니다. 불행하게도 고급 작업이 먼저 실행되지 않을 수 있으므로 채널은 적합하지 않습니다. (3) 선택 3:

slice

. 슬라이스 요소는 순서가 지정되어 있기 때문에 슬라이싱이 가능해 보입니다. 따라서 모든 슬라이스 요소를 실행 시간 순서대로 배열할 수 있다면 매번 슬라이스의 헤드 요소(아마도 테일 요소)만 읽으면 됩니다. 우리가 원하는 작업.

2. 소비자

소비자 입장에서 보면 각 작업을 특정 시점에 어떻게 소비하게 만들느냐가 가장 큰 어려움입니다. 그렇다면 각 작업에 대해 실행하기 전에 일정 시간 동안 기다리게 하려면 어떻게 해야 할까요?

네, 타이머입니다.

결론적으로 "

슬라이싱 + 타이머" 조합이 목적을 달성할 수 있어야 합니다.

단계별

1. 데이터 흐름

(1) 사용자는

InitDelayQueue()를 호출하여 지연된 작업 개체를 초기화합니다.

(2) 코루틴을 열고 작업 작업 파이프라인(추가/삭제 신호)과 실행 시간 파이프라인(timer.C 신호)을 수신합니다.

(3) 사용자가 추가/삭제 신호를 보냅니다. (4) (2)의 코루틴은 (3)의 신호를 캡처하고 작업 목록을 변경합니다.

(5) 작업 실행 시점이 도래하면(timer.C 파이프라인에서 출력되는 요소가 있는 경우) 작업을 실행합니다.

2. 데이터 구조

(1) 지연된 작업 개체

// 延时任务对象
type DelayQueue struct {
   tasks                 []*task             // 存储任务列表的切片
   add                   chan *task          // 用户添加任务的管道信号
   remove                chan string         // 用户删除任务的管道信号
   waitRemoveTaskMapping map[string]struct{} // 等待删除的任务id列表
}

로그인 후 복사

여기에

waitRemoveTaskMapping

필드가 있다는 점에 유의해야 합니다. 삭제할 작업이 아직 추가 파이프라인에 있고 작업 필드에서 적시에 업데이트되지 않았을 수 있으므로 고객이 삭제하려는 작업의 ID를 임시로 기록해야 합니다.

(2) 태스크 객체

// 任务对象
type task struct {
   id       string    // 任务id
   execTime time.Time // 执行时间
   f        func()    // 执行函数
}

로그인 후 복사

golang이 지연된 작업을 구현하는 방법에 대한 간략한 분석 3. 초기화 지연 태스크 객체

// 初始化延时任务对象
func InitDelayQueue() *DelayQueue {
   q := &DelayQueue{
      add:                   make(chan *task, 10000),
      remove:                make(chan string, 100),
      waitRemoveTaskMapping: make(map[string]struct{}),
   }

   return q
}

로그인 후 복사

이 과정에서 태스크에 대한 사용자의 조작 신호와 태스크의 실행 시간 신호를 모니터링해야 합니다.

func (q *DelayQueue) start() {
   for {
      // to do something...

      select {
      case now := <-timer.C:
         // 任务执行时间信号
         // to do something...
      case t := <-q.add:
         // 任务推送信号
         // to do something...
      case id := <-q.remove:
         // 任务删除信号
         // to do something...
      }
   }
}

로그인 후 복사

초기화 방법 개선:

// 初始化延时任务对象
func InitDelayQueue() *DelayQueue {
   q := &DelayQueue{
      add:                   make(chan *task, 10000),
      remove:                make(chan string, 100),
      waitRemoveTaskMapping: make(map[string]struct{}),
   }

   // 开启协程，监听任务相关信号
   go q.start()
   return q
}

로그인 후 복사

4. 생산자가 작업을 푸시할 때 추가 파이프라인에 작업을 추가하기만 하면 됩니다. 여기서는 작업 ID를 생성하고 사용자에게 반환합니다. .

// 用户推送任务
func (q *DelayQueue) Push(timeInterval time.Duration, f func()) string {
   // 生成一个任务id，方便删除使用
   id := genTaskId()
   t := &task{
      id:       id,
      execTime: time.Now().Add(timeInterval),
      f:        f,
   }

   // 将任务推到add管道中
   q.add <- t
   return id
}

로그인 후 복사

5、任务推送信号的处理

在这里，我们要将用户推送的任务放到延时任务的tasks字段中。由于，我们需要将任务按照执行时间顺序排序，所以，我们需要找到新增任务在切片中的插入位置。又因为，插入之前的任务列表已经是有序的，所以，我们可以采用二分法处理。

// 使用二分法判断新增任务的插入位置
func (q *DelayQueue) getTaskInsertIndex(t *task, leftIndex, rightIndex int) (index int) {
   if len(q.tasks) == 0 {
      return
   }

   length := rightIndex - leftIndex
   if q.tasks[leftIndex].execTime.Sub(t.execTime) >= 0 {
      // 如果当前切片中最小的元素都超过了插入的优先级，则插入位置应该是最左边
      return leftIndex
   }

   if q.tasks[rightIndex].execTime.Sub(t.execTime) <= 0 {
      // 如果当前切片中最大的元素都没超过插入的优先级，则插入位置应该是最右边
      return rightIndex + 1
   }

   if length == 1 && q.tasks[leftIndex].execTime.Before(t.execTime) && q.tasks[rightIndex].execTime.Sub(t.execTime) >= 0 {
      // 如果插入的优先级刚好在仅有的两个优先级之间，则中间的位置就是插入位置
      return leftIndex + 1
   }

   middleVal := q.tasks[leftIndex+length/2].execTime

   // 这里用二分法递归的方式，一直寻找正确的插入位置
   if t.execTime.Sub(middleVal) <= 0 {
      return q.getTaskInsertIndex(t, leftIndex, leftIndex+length/2)
   } else {
      return q.getTaskInsertIndex(t, leftIndex+length/2, rightIndex)
   }
}

로그인 후 복사

找到正确的插入位置后，我们才能将任务准确插入：

// 将任务添加到任务切片列表中
func (q *DelayQueue) addTask(t *task) {
   // 寻找新增任务的插入位置
   insertIndex := q.getTaskInsertIndex(t, 0, len(q.tasks)-1)
   // 找到了插入位置，更新任务列表
   q.tasks = append(q.tasks, &task{})
   copy(q.tasks[insertIndex+1:], q.tasks[insertIndex:])
   q.tasks[insertIndex] = t
}

로그인 후 복사

那么，在监听add管道的时候，我们直接调用上述addTask() 即可。

func (q *DelayQueue) start() {
   for {
      // to do something...

      select {
      case now := <-timer.C:
         // 任务执行时间信号
         // to do something...
      case t := <-q.add:
         // 任务推送信号
         q.addTask(t)
      case id := <-q.remove:
         // 任务删除信号
         // to do something...
      }
   }
}

로그인 후 복사

6、生产者删除任务

// 用户删除任务
func (q *DelayQueue) Delete(id string) {
   q.remove <- id
}

로그인 후 복사

7、任务删除信号的处理

在这里，我们可以遍历任务列表，根据删除任务的id找到其在切片中的对应index。

// 删除指定任务
func (q *DelayQueue) deleteTask(id string) {
   deleteIndex := -1
   for index, t := range q.tasks {
      if t.id == id {
         // 找到了在切片中需要删除的所以呢
         deleteIndex = index
         break
      }
   }

   if deleteIndex == -1 {
      // 如果没有找到删除的任务，说明任务还在add管道中，来不及更新到tasks中，这里我们就将这个删除id临时记录下来
      // 注意，这里暂时不考虑，任务id非法的特殊情况
      q.waitRemoveTaskMapping[id] = struct{}{}
      return
   }

   if len(q.tasks) == 1 {
      // 删除后，任务列表就没有任务了
      q.tasks = []*task{}
      return
   }

   if deleteIndex == len(q.tasks)-1 {
      // 如果删除的是，任务列表的最后一个元素，则执行下列代码
      q.tasks = q.tasks[:len(q.tasks)-1]
      return
   }

   // 如果删除的是，任务列表的其他元素，则需要将deleteIndex之后的元素，全部向前挪动一位
   copy(q.tasks[deleteIndex:len(q.tasks)-1], q.tasks[deleteIndex+1:len(q.tasks)-1])
   q.tasks = q.tasks[:len(q.tasks)-1]
   return
}

로그인 후 복사

然后，我们可以完善start()方法了。

func (q *DelayQueue) start() {
   for {
      // to do something...

      select {
      case now := <-timer.C:
         // 任务执行时间信号
         // to do something...
      case t := <-q.add:
         // 任务推送信号
         q.addTask(t)
      case id := <-q.remove:
         // 任务删除信号
         q.deleteTask(id)
      }
   }
}

로그인 후 복사

8、任务执行信号的处理

start()执行的时候，分成两种情况：任务列表为空，只需要监听add管道即可；任务列表不为空的时候，需要监听所有管道。任务执行信号，主要是依靠timer来实现，属于第二种情况。

func (q *DelayQueue) start() {
   for {
      if len(q.tasks) == 0 {
           // 任务列表为空的时候，只需要监听add管道
           select {
           case t := <-q.add:
              //添加任务
              q.addTask(t)
           }
        
           continue
      }

      // 任务列表不为空的时候，需要监听所有管道

      // 任务的等待时间=任务的执行时间-当前的时间
      currentTask := q.tasks[0]
      timer := time.NewTimer(currentTask.execTime.Sub(time.Now()))

      select {
      case now := <-timer.C:
         // 任务执行信号
         timer.Stop()
        if _, isRemove := q.waitRemoveTaskMapping[currentTask.id]; isRemove {
           // 之前客户已经发出过该任务的删除信号，因此需要结束任务，刷新任务列表
           q.endTask()
           delete(q.waitRemoveTaskMapping, currentTask.id)
           continue
        }
        
        // 开启协程，异步执行任务
        go q.execTask(currentTask, now)
        // 任务结束，刷新任务列表
        q.endTask()
      case t := <-q.add:
         // 任务推送信号
         timer.Stop()
         q.addTask(t)
      case id := <-q.remove:
         // 任务删除信号
         timer.Stop()
         q.deleteTask(id)
      }
   }
}

로그인 후 복사

执行任务：

// 执行任务
func (q *DelayQueue) execTask(task *task, currentTime time.Time) {
   if task.execTime.After(currentTime) {
      // 如果当前任务的执行时间落后于当前时间，则不执行
      return
   }

   // 执行任务
   task.f()
   return
}

로그인 후 복사

结束任务，刷新任务列表：

// 一个任务去执行了，刷新任务列表
func (q *DelayQueue) endTask() {
   if len(q.tasks) == 1 {
      q.tasks = []*task{}
      return
   }

   q.tasks = q.tasks[1:]
}

로그인 후 복사

9、完整代码

delay_queue.go

package delay_queue

import (
   "go.mongodb.org/mongo-driver/bson/primitive"
   "time"
)

// 延时任务对象
type DelayQueue struct {
   tasks                 []*task             // 存储任务列表的切片
   add                   chan *task          // 用户添加任务的管道信号
   remove                chan string         // 用户删除任务的管道信号
   waitRemoveTaskMapping map[string]struct{} // 等待删除的任务id列表
}

// 任务对象
type task struct {
   id       string    // 任务id
   execTime time.Time // 执行时间
   f        func()    // 执行函数
}

// 初始化延时任务对象
func InitDelayQueue() *DelayQueue {
   q := &DelayQueue{
      add:                   make(chan *task, 10000),
      remove:                make(chan string, 100),
      waitRemoveTaskMapping: make(map[string]struct{}),
   }

   // 开启协程，监听任务相关信号
   go q.start()
   return q
}

// 用户删除任务
func (q *DelayQueue) Delete(id string) {
   q.remove <- id
}

// 用户推送任务
func (q *DelayQueue) Push(timeInterval time.Duration, f func()) string {
   // 生成一个任务id，方便删除使用
   id := genTaskId()
   t := &task{
      id:       id,
      execTime: time.Now().Add(timeInterval),
      f:        f,
   }

   // 将任务推到add管道中
   q.add <- t
   return id
}

// 监听各种任务相关信号
func (q *DelayQueue) start() {
   for {
      if len(q.tasks) == 0 {
         // 任务列表为空的时候，只需要监听add管道
         select {
         case t := <-q.add:
            //添加任务
            q.addTask(t)
         }

         continue
      }

      // 任务列表不为空的时候，需要监听所有管道

      // 任务的等待时间=任务的执行时间-当前的时间
      currentTask := q.tasks[0]
      timer := time.NewTimer(currentTask.execTime.Sub(time.Now()))

      select {
      case now := <-timer.C:
         timer.Stop()
         if _, isRemove := q.waitRemoveTaskMapping[currentTask.id]; isRemove {
            // 之前客户已经发出过该任务的删除信号，因此需要结束任务，刷新任务列表
            q.endTask()
            delete(q.waitRemoveTaskMapping, currentTask.id)
            continue
         }

         // 开启协程，异步执行任务
         go q.execTask(currentTask, now)
         // 任务结束，刷新任务列表
         q.endTask()
      case t := <-q.add:
         // 添加任务
         timer.Stop()
         q.addTask(t)
      case id := <-q.remove:
         // 删除任务
         timer.Stop()
         q.deleteTask(id)
      }
   }
}

// 执行任务
func (q *DelayQueue) execTask(task *task, currentTime time.Time) {
   if task.execTime.After(currentTime) {
      // 如果当前任务的执行时间落后于当前时间，则不执行
      return
   }

   // 执行任务
   task.f()
   return
}

// 一个任务去执行了，刷新任务列表
func (q *DelayQueue) endTask() {
   if len(q.tasks) == 1 {
      q.tasks = []*task{}
      return
   }

   q.tasks = q.tasks[1:]
}

// 将任务添加到任务切片列表中
func (q *DelayQueue) addTask(t *task) {
   // 寻找新增任务的插入位置
   insertIndex := q.getTaskInsertIndex(t, 0, len(q.tasks)-1)
   // 找到了插入位置，更新任务列表
   q.tasks = append(q.tasks, &task{})
   copy(q.tasks[insertIndex+1:], q.tasks[insertIndex:])
   q.tasks[insertIndex] = t
}

// 删除指定任务
func (q *DelayQueue) deleteTask(id string) {
   deleteIndex := -1
   for index, t := range q.tasks {
      if t.id == id {
         // 找到了在切片中需要删除的所以呢
         deleteIndex = index
         break
      }
   }

   if deleteIndex == -1 {
      // 如果没有找到删除的任务，说明任务还在add管道中，来不及更新到tasks中，这里我们就将这个删除id临时记录下来
      // 注意，这里暂时不考虑，任务id非法的特殊情况
      q.waitRemoveTaskMapping[id] = struct{}{}
      return
   }

   if len(q.tasks) == 1 {
      // 删除后，任务列表就没有任务了
      q.tasks = []*task{}
      return
   }

   if deleteIndex == len(q.tasks)-1 {
      // 如果删除的是，任务列表的最后一个元素，则执行下列代码
      q.tasks = q.tasks[:len(q.tasks)-1]
      return
   }

   // 如果删除的是，任务列表的其他元素，则需要将deleteIndex之后的元素，全部向前挪动一位
   copy(q.tasks[deleteIndex:len(q.tasks)-1], q.tasks[deleteIndex+1:len(q.tasks)-1])
   q.tasks = q.tasks[:len(q.tasks)-1]
   return
}

// 寻找任务的插入位置
func (q *DelayQueue) getTaskInsertIndex(t *task, leftIndex, rightIndex int) (index int) {
   // 使用二分法判断新增任务的插入位置
   if len(q.tasks) == 0 {
      return
   }

   length := rightIndex - leftIndex
   if q.tasks[leftIndex].execTime.Sub(t.execTime) >= 0 {
      // 如果当前切片中最小的元素都超过了插入的优先级，则插入位置应该是最左边
      return leftIndex
   }

   if q.tasks[rightIndex].execTime.Sub(t.execTime) <= 0 {
      // 如果当前切片中最大的元素都没超过插入的优先级，则插入位置应该是最右边
      return rightIndex + 1
   }

   if length == 1 && q.tasks[leftIndex].execTime.Before(t.execTime) && q.tasks[rightIndex].execTime.Sub(t.execTime) >= 0 {
      // 如果插入的优先级刚好在仅有的两个优先级之间，则中间的位置就是插入位置
      return leftIndex + 1
   }

   middleVal := q.tasks[leftIndex+length/2].execTime

   // 这里用二分法递归的方式，一直寻找正确的插入位置
   if t.execTime.Sub(middleVal) <= 0 {
      return q.getTaskInsertIndex(t, leftIndex, leftIndex+length/2)
   } else {
      return q.getTaskInsertIndex(t, leftIndex+length/2, rightIndex)
   }
}

func genTaskId() string {
   return primitive.NewObjectID().Hex()
}

로그인 후 복사

测试代码：delay_queue_test.go

package delay_queue

import (
   "fmt"
   "testing"
   "time"
)

func TestDelayQueue(t *testing.T) {
   q := InitDelayQueue()
   for i := 0; i < 100; i++ {
      go func(i int) {
         id := q.Push(time.Duration(i)*time.Second, func() {
            fmt.Printf("%d秒后执行...\n", i)
            return
         })

         if i%7 == 0 {
            q.Delete(id)
         }
      }(i)
   }

   time.Sleep(time.Hour)
}

로그인 후 복사

头脑风暴

上面的方案，的确实现了延时任务的效果，但是其中仍然有一些问题，仍然值得我们思考和优化。

1、按照上面的方案，如果大量延时任务的执行时间，集中在同一个时间点，会造成短时间内timer频繁地创建和销毁。

2、上述方案相比于time.AfterFunc()方法，我们需要在哪些场景下作出取舍。

3、如果服务崩溃或重启，如何去持久化队列中的任务。

小结

本文和大家讨论了延时任务在golang中的一种实现方案，在这个过程中，一次性定时器timer、切片、管道等golang特色，以及二分插入等常见算法都体现得淋漓尽致。

【相关推荐：Go视频教程、编程教学】
위 내용은 golang이 지연된 작업을 구현하는 방법에 대한 간략한 분석의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!