効率的な分散タスクのスケジューリングと実行を実現する Go と RabbitMQ のベストプラクティス

2年 ago
芽依, 雨夜
2 minutes

GolangとRabbitMQの組み合わせを利用すれば、分散タスクのスケジューリングと実行という高効率なソリューションを実現できます。以下に、GolangとRabbitMQを使ったベストプラクティスの例を示します。

  1. タスク定義モデル:まず、タスクIDやタスクパラメータなどのフィールドを含むタスク構造を定義し、タスクのデータ構造をJSON形式でシリアル化する。
type Task struct {
ID       string                 `json:"id"`
Params   map[string]interface{} `json:"params"`
// 添加其他任务字段
}
  1. 生産者: RabbitMQのタスクキューにタスクを送信するためのProducerプログラムを記述します。
func main() {
// 连接RabbitMQ服务器
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("Failed to connect to RabbitMQ: %v", err)
}
defer conn.Close()
// 创建一个任务通道
ch, err := conn.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %v", err)
}
defer ch.Close()
// 声明一个任务队列
queue, err := ch.QueueDeclare("task_queue", true, false, false, false, nil)
if err != nil {
log.Fatalf("Failed to declare a queue: %v", err)
}
// 发布任务
task := Task{
ID:     "1",
Params: map[string]interface{}{"key": "value"},
}
message, err := json.Marshal(task)
if err != nil {
log.Fatalf("Failed to marshal task: %v", err)
}
err = ch.Publish("", queue.Name, false, false, amqp.Publishing{
DeliveryMode: amqp.Persistent,
ContentType:  "application/json",
Body:         message,
})
if err != nil {
log.Fatalf("Failed to publish task: %v", err)
}
log.Println("Task sent")
}
  1. 消費者: RabbitMQのタスクキューからタスク受信し、対応する処理ロジックを実行するためのコンシューマプログラムを作成します。
func main() {
// 连接RabbitMQ服务器
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("Failed to connect to RabbitMQ: %v", err)
}
defer conn.Close()
// 创建一个任务通道
ch, err := conn.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %v", err)
}
defer ch.Close()
// 声明一个任务队列
queue, err := ch.QueueDeclare("task_queue", true, false, false, false, nil)
if err != nil {
log.Fatalf("Failed to declare a queue: %v", err)
}
// 设置每次只接收一条未确认的任务
err = ch.Qos(1, 0, false)
if err != nil {
log.Fatalf("Failed to set QoS: %v", err)
}
// 注册一个任务消费者
msgs, err := ch.Consume(queue.Name, "", false, false, false, false, nil)
if err != nil {
log.Fatalf("Failed to register a consumer: %v", err)
}
// 处理任务
go func() {
for msg := range msgs {
var task Task
err := json.Unmarshal(msg.Body, &task)
if err != nil {
log.Printf("Failed to unmarshal task: %v", err)
msg.Ack(false)
continue
}
// 执行任务逻辑
// ...
msg.Ack(false)
log.Printf("Task %s executed", task.ID)
}
}()
log.Println("Waiting for tasks...")
select {}
}

上記の手順により、Go言語とRabbitMQを利用して、高効率な分散タスクのスケジューリングおよび実行システムを実現できます。プロデューサーはタスクをRabbitMQのタスクキューに送信します。コンシューマーはキューからタスクを受信し、対応する処理ロジックを実行します。また、未確認のタスクを一度に1つだけ受信するように設定することで、タスクの負荷分散と順次実行を実現します。

なお、これらの事例はあくまで説明のためのものであり、実際には

bannerAds