job-for-serverless/task.go

package main

import (
	"fmt"
	"job-for-serverless/db"
	"math"
	"net/http"
	"strconv"
	"time"
)

func getBeijingTZ() *time.Location {
	offset := int((8 * time.Hour).Seconds())
	return time.FixedZone("Beijing Time", offset)
}

func getBeijingTime(dt time.Time) time.Time {
	// offset := int((8 * time.Hour).Seconds())
	// _, offset2 := dt.Zone()
	// if offset == offset2 {
	// 	return dt
	// }

	// 北京时间
	beijingTZ := getBeijingTZ()
	return dt.In(beijingTZ)
}

func getFixDateTime(dt time.Time, fixedStr string) (time.Time, error) {
	y, m, d := dt.Date()
	dtStr := fmt.Sprintf("%d-%d-%d %s", y, m, d, fixedStr)
	return time.ParseInLocation("2006-01-02 15:04:05", dtStr, getBeijingTZ())
}

// GetTasks 获取指定的内容
func GetTasks(limit []int) []Task {
	jobs := make([]db.TaJobs, 0)
	if err := DB.Asc("job_id").Limit(limit[0], limit[1:]...).Find(&jobs); err != nil {
		logger.Error("Get jobs error: ", err)
	}

	// logger.Info("Get All jobs: ", jobs)

	tasks := make([]Task, 0)

	if len(jobs) == 0 {
		return tasks
	}

	for _, job := range jobs {
		now := getBeijingTime(time.Now())
		payload := job
		task := Task{
			ID:      strconv.Itoa(job.JobId),
			Payload: &payload,
		}

		// 下一次的执行时间
		var nextTime time.Time

		// 1、如果是指定时间
		if job.AtTime != "" && job.Interval == 0 {
			nextTime, _ = getFixDateTime(now, job.AtTime+":00")
			if math.Abs(nextTime.Sub(now).Minutes()) < 1 {
				tasks = append(tasks, task)
				continue
			}
		}

		// 2、如果是间隔执行
		if job.AtTime == "" && job.Interval > 0 {
			if job.LastTime.IsZero() {
				tasks = append(tasks, task)
				continue
			} else {
				nextTime = getBeijingTime(job.LastTime).Add(time.Minute * time.Duration(job.Interval))

				// 如果本次时间正确，或者前面有问题未执行的
				if math.Abs(nextTime.Sub(now).Minutes()) < 1 || nextTime.Before(now) {
					tasks = append(tasks, task)
					continue
				}
			}
		}

		// 3、如果是既指定了时间，又有间隔执行
		if job.AtTime != "" && job.Interval > 0 {
			// 如果没有执行过, 或者上次执行时非今天
			if job.LastTime.IsZero() || getBeijingTime(job.LastTime).Day() < now.Day() {
				nextTime, _ = getFixDateTime(now, job.AtTime+":00")
			} else {
				nextTime = getBeijingTime(job.LastTime).Add(time.Minute * time.Duration(job.Interval))
			}

			// 下次执行时间不是今天的, 不执行
			if nextTime.Day() != now.Day() {
				continue
			}

			if math.Abs(nextTime.Sub(now).Minutes()) < 1 {
				tasks = append(tasks, task)
				continue
			}
		}
	}

	return tasks
}

// Exec 具体的任务执行方式
var Exec = func(t Task) error {
	job := t.Payload.(*db.TaJobs)
	if job == nil || job.Content == "" {
		return nil
	}

	logger.Info("Run job: ", job.JobId)

	resp, err := http.Get(job.Content)
	if resp != nil && resp.StatusCode > 299 {
		err = fmt.Errorf("%d-%s", resp.StatusCode, resp.Status)
	}

	if err != nil {
		logger.Error("Job [", t.ID, "] ", err)

		// 再尝试一次
		_, err = http.Get(job.Content)
		if err != nil {
			logger.Error("Job [", t.ID, "] ", err)
		}
	}

	now := getBeijingTime(time.Now())
	job.LastTime = now
	job.LastResult = "success"
	if err != nil {
		job.LastResult = err.Error()
	}

	if _, e := DB.ID(job.JobId).Cols("last_time", "last_result").Update(job); e != nil {
		logger.Error("Update job [", t.ID, "] last time error ", err)
	}

	return err
}