【Go】タスクキューとクーロンを使った並列処理の実装

package clocky_be

import (
	"log"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/jmoiron/sqlx"
)

// Message はキューに追加されるメッセージを表します。
type Message struct {
	UserID string // ユーザーID
	Type   string // メッセージのタイプ
}

// ScheduleTask はスケジュールされたタスクを実行し、メッセージをキューに追加します。
func ScheduleTask(db *sqlx.DB, messagesChan chan Message) {
	log.Println("ScheduleTask started")
	// 現在の時刻を取得
	currentTime := time.Now()
	jst, err := time.LoadLocation("Asia/Tokyo")
	if err != nil {
		log.Fatalf("Failed to load the 'Asia/Tokyo' time zone: %v", err)
	}
	currentTimeJST := currentTime.In(jst)
	log.Printf("Current time in JST: %v", currentTimeJST)

	// 現在時刻から1時間後の時刻を計算
	oneHourLater := currentTimeJST.Add(time.Hour)
	log.Printf("One hour later in JST: %v", oneHourLater)

	// SQLクエリを実行して、条件に合致するユーザーを取得
	var users []User
	query := `SELECT * FROM users WHERE DATE_FORMAT(weather_announcement_time, '%H:%i') = DATE_FORMAT(?, '%H:%i');`
	err = db.Select(&users, query, oneHourLater.String())

	if err != nil {
		log.Printf("Error fetching users: %v", err)
		return
	}
	// 合致したユーザーがいるかどうかを確認
	if len(users) > 0 {
		// 取得したユーザーからメッセージを生成し、チャネルに送信
		for _, user := range users {
			log.Printf("Scheduling task for user: %s", user.UID)
			message := Message{
				UserID: user.UID,
				Type:   "weather",
			}
			messagesChan <- message
		}
		log.Println("ScheduleTask completed")
	} else {
		// 合致したユーザーがいない場合のログ
		log.Println("ScheduleTask skipped: No matching users found")
	}
}

// processMessages はキューからメッセージを取り出し、処理します。
func ProcessMessages(messagesChan chan Message) {
	for message := range messagesChan {
		log.Printf("Processing message for user: %s, type: %s", message.UserID, message.Type)
		if message.Type == "weather" {
			// 天気情報のテキストを取得する。一旦直書きで
			// 天気情報取得のAPIリクエストを廃止する回収の時に、合わせて回収する。
			weatherText := "東京の本日の天気は晴れ時々曇り、最高気温は25度です。"

			// AWSセッションの作成
			config := NewConfig()
			awsSession, err := session.NewSession(&aws.Config{
				Region:      aws.String(config.AWSRegion),
				Credentials: credentials.NewStaticCredentials(config.AWSAccessKey, config.AWSSecretKey, ""),
			})
			if err != nil {
				log.Fatalf("Failed to create AWS session: %v", err)
			}

			// 天気情報を音声ファイルに変換して、AWS S3にアップロードする
			if err := SynthesizeSpeechAndUploadToS3(awsSession, config.AWSS3Bucket, weatherText, message.UserID, message.Type); err != nil {
				log.Printf("Failed to synthesize and upload speech for user %s: %v", message.UserID, err)
				continue
			}
		}
	}
}

$ go get github.com/robfig/cron/v3@v3.0.0

// Seconds field, required
cron.New(cron.WithSeconds())

// Seconds field, optional
cron.New(cron.WithParser(cron.NewParser(
	cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
)))

package main

import (
	"log"

	clocky "github.com/EarEEG-dev/clocky_be"
	_ "github.com/go-sql-driver/mysql"
	"github.com/jmoiron/sqlx"
	"github.com/robfig/cron/v3"
)

func main() {
	config := clocky.NewConfig()

	db, err := sqlx.Open("mysql", config.DSN)
	if err != nil {
		log.Fatalf("unable to connect to database: %v", err)
	}

	if err := db.Ping(); err != nil {
		log.Fatalf("unable to ping database: %v", err)
	}
	log.Printf("connected to database")

	// メッセージチャネルとCronジョブの設定
	messagesChan := make(chan clocky.Message)

	// 秒単位をサポートしてcronを初期化
	c := cron.New(cron.WithSeconds())

	// 毎分0秒時点で起動
	c.AddFunc("0 * * * * *", func() {
		log.Println("Cron job triggered")
		clocky.ScheduleTask(db, messagesChan)
	})
	go c.Start()

	// 処理関数の起動数を増やして並列に処理させることも可能
	go clocky.ProcessMessages(messagesChan)

	// gRPCサーバー
	grpcServer := clocky.NewGRPCServer(config, db)
	go grpcServer.Start(":50051")

	// Rest API
	server := clocky.NewServer(config, db)
	server.Start()

}