Redis क्लस्टर

रेडिस क्लस्टर का उपयोग करने के फायदे

रेडिस क्लस्टर का उपयोग करके, आप निम्नलिखित फायदे प्राप्त कर सकते हैं:

• आसानी से डेटा को कई रेडिस नोड्स पर विभाजित करें
• किसी नोड के विफल होने के मामले में उपलब्धता बनाए रखें
• स्वचालित रूप से फेलओवर करें

अवलोकन

Asynq कतारों पर आधारित डेटा को शार्ड करता है। उपरोक्त आरेख में, हमारे पास 6 इंस्टेंस (3 मास्टर और 3 स्लेव्स) वाला एक रेडिस क्लस्टर है और 4 कतार हैं (क्यू1, क्यू2, क्यू3, क्यू4)।

• मास्टर1 (और उसका प्रतिरूप, स्लेव1) क्यू1 और क्यू2 को होस्ट करते हैं।
• मास्टर2 (और उसका प्रतिरूप, स्लेव2) क्यू3 को होस्ट करते हैं।
• मास्टर3 (और उसका प्रतिरूप, स्लेव3) क्यू4 को होस्ट करते हैं।

जब आप asynq.Client का उपयोग करके tasks को enqueue करते हैं, आप Queue विकल्प का उपयोग करके कतार को निर्दिष्ट कर सकते हैं। इन कार्यों को enqueued किया जाएगा, जो की asynq.Server(s) द्वारा कटारों से tasks को पुल करने वाले dwara consumers द्वारा सेवन किया जाएगा।

ट्यूटोरियल

इस खंड में, हम यह बताएँगे कि कैसे Asynq के लिए मैसेज ब्रोकर के रूप में रेडिस क्लस्टर का उपयोग करें। हम मान लेते हैं कि आपके पास 7000-7005 पोर्ट्स पर चल रहे 6 रेडिस इंस्टेंस का क्लस्टर है। नीचे एक redis.conf फ़ाइल का एक उदाहरण है:

port 7000
cluster-enabled yes
cluster-config-file nodes.conf
cluster-node-timeout 5000
appendonly yes

अगले, हम दो बाइनरी फ़ाइलें बनाएँगे: क्लाइंट और वर्कर।

go mod init asynq-redis-cluster-quickstart
mkdir client worker
touch client/client.go worker/worker.go

client.go में, हम एक नया asynq.Client बनाएँगे और RedisClusterClientOpt को द्वारा रेडिस क्लस्टर से कनेक्ट करने के तरीके को निर्दिष्ट करेंगे।

client := asynq.NewClient(asynq.RedisClusterClientOpt{
    Addrs: []string{":7000", ":7001", ":7002", ":7003", ":7004", ":7005"},
})

जैसे ही हमारे पास क्लाइंट होता है, हम टास्क बनाएँगे और उन्हें तीन अलग-अलग क्यू में डालेंगे:

• नोटिफिकेशन्स
• वेबहुक्स
• इमेजेस

// client.go

package main

import (
    "fmt"
    "log"

    "github.com/hibiken/asynq"
)

// क्यू नाम
const (
    QueueNotifications = "notifications"
    QueueWebhooks      = "webhooks"
    QueueImages        = "images"
)

func main() {
    client := asynq.NewClient(asynq.RedisClusterClientOpt{
        Addrs: []string{":7000", ":7001", ":7002", ":7003", ":7004", ":7005"},
    })
    defer client.Close()

    // "notifications:email" नामक एक टास्क बनाएँगे और उसे "notifications" क्यू में डालेंगे।
    task := asynq.NewTask("notifications:email", map[string]interface{}{"to": 123, "from": 456})
    res, err := client.Enqueue(task, asynq.Queue(QueueNotifications))
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("सफलतापूर्वक इनक्यू किया गया: %+v\n", res)

    // "webhooks:sync" नामक एक टास्क बनाएँगे और उसे "webhooks" क्यू में डालेंगे।
    task = asynq.NewTask("webhooks:sync", map[string]interface{}{"data": 123})
    res, err = client.Enqueue(task, asynq.Queue(QueueWebhooks))
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("सफलतापूर्वक इनक्यू किया गया: %+v\n", res)

    // "images:resize" नामक एक टास्क बनाएँगे और उसे "images" क्यू में डालेंगे।
    task = asynq.NewTask("images:resize", map[string]interface{}{"src": "some/path/to/image"})
    res, err = client.Enqueue(task, asynq.Queue(QueueImages))
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("सफलतापूर्वक इनक्यू किया गया: %+v\n", res)
}

चलो, इस प्रोग्राम को चलाकर हम इन तीन टास्कों को क्यू में डालेंगे।

go run client/client.go

अब, वर्कर में जाकर हम इन तीन टास्कों को प्रोसेस करेंगे। worker.go में, हम एक asynq.Server बनाएँगे ताकि हम इन तीन क्यू से टास्क को कन्स्यूम कर सकें। उसी तरह, हम RedisClusterClientOpt का उपयोग करके हमारे रेडिस क्लस्टर से कनेक्ट करेंगे।

// worker.go

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/hibiken/asynq"
)

func main() {
	redisConnOpt := asynq.RedisClusterClientOpt{Addrs: []string{":7000", ":7001", ":7002", ":7003", ":7004", ":7005"}}
	srv := asynq.NewServer(redisConnOpt, asynq.Config{
		Concurrency: 20,
		// यहां प्रत्येक कतार के लिए समान प्राथमिकता सेट करें
		Queues: map[string]int{
			"notifications": 1,
			"webhooks":      1,
			"images":        1,
		},
	})
	mux := asynq.NewServeMux()
	mux.HandleFunc("notifications:email", handleEmailTask)
	mux.HandleFunc("webhooks:sync", handleWebhookSyncTask)
	mux.HandleFunc("images:resize", handleImageResizeTask)
	if err := srv.Run(mux); err != nil {
		log.Fatalf("सर्वर शुरू करने में विफल: %v", err)
	}
}

func handleEmailTask(ctx context.Context, t *asynq.Task) error {
	to, err := t.Payload.GetInt("to")
	if err != nil {
		return err
	}
	from, err := t.Payload.GetInt("from")
	if err != nil {
		return err
	}
	fmt.Printf("%d से %d को ईमेल भेज रहा हूँ\n", from, to)
	return nil
}

func handleWebhookSyncTask(ctx context.Context, t *asynq.Task) error {
	data, err := t.Payload.GetInt("data")
	if err != nil {
		return err
	}
	fmt.Printf("वेबहुक टास्क का संचालन कर रहा हूँ: %d\n", data)
	return nil
}

func handleImageResizeTask(ctx context.Context, t *asynq.Task) error {
	src, err := t.Payload.GetString("src")
	if err != nil {
		return err
	}
	fmt.Printf("छवि को रीसाइज कर रहा हूँ: %s\n", src)
	return nil
}

हमें इस कार्यकर्ता सर्वर को चलाने के लिए तीन कार्यों को हैंडल करने होंगे जो हमने पहले बनाए थे।

go run worker/worker.go

आपको हर हैंडलर से प्रिंट किए गए संदेश दिखाई देना चाहिए।

जैसा अवलोकन में कहा गया, Asynq कुयूंबी में डेटा को कतारों के आधार पर विभाजित करता है। सबको एक ही कतार में जोड़ी गई सभी कार्य एक ही रेडिस नोड पर होती हैं। तो, कौन सा रेडिस नोड कौन सी कतार होस्ट करता है?

हम इस सवाल का उत्तर देने के लिए CLI का उपयोग कर सकते हैं।

asynq queue ls --cluster

यह कमांड कतारों की सूची प्रिंट करेगा, साथ ही:

- क्लस्टर नोड जिसे कतार संबंधित है
- क्लस्टर हैश स्लॉट जिसे कतार को मैप किया गया है

नतीजा कुछ इस तरह से दिख सकता है:

कतार क्लस्टर की स्थानिक चाभी क्लस्टर नोड्स

images 9450 [{d54231bccd6c1765ea15caf95a41c67b10b91e58 127.0.0.1:7001} {70a7d4569eac28eed577ee91863703ffab98d2e0 127.0.0.1:7005}] webhooks 4418 [{d58959f6057ad0911d92d86d1d16dc2242e9ec48 127.0.0.1:7004} {e2fb9f1296a8d3a49818e0f9be3bfd74fdc052ea 127.0.0.1:7000}] notifications 16340 [{c738a8a98c5f5f9161e9563fa739f9c8191b7f1a 127.0.0.1:7002} {18cdaa0712191d74656f08017371df41eeaad5fa 127.0.0.1:7003}]