Schreiben eines einfachen Balancers auf Go

Load Balancer spielen eine Schlüsselrolle in der Webarchitektur. Mit ihnen können Sie die Last auf mehrere Backends verteilen und so die Skalierbarkeit verbessern. Und da wir mehrere Backends konfiguriert haben, wird der Service hoch verfügbar, da der Balancer im Falle eines Ausfalls auf einem Server einen anderen funktionierenden Server auswählen kann.

Nachdem ich mit professionellen Balancern wie NGINX gespielt hatte, versuchte ich, aus Spaß einen einfachen Balancer zu erstellen. Ich habe es auf Go geschrieben, es ist eine moderne Sprache, die volle Parallelität unterstützt. Die Standardbibliothek in Go bietet viele Funktionen und ermöglicht das Schreiben von Hochleistungsanwendungen mit weniger Code. Darüber hinaus wird zur Vereinfachung der Verteilung eine einzelne statisch verknüpfte Binärdatei generiert.

Wie unser Balancer funktioniert

Es werden verschiedene Algorithmen verwendet, um die Last auf die Backends zu verteilen. Zum Beispiel:

• Round Robin - Die Last wird unter Berücksichtigung der gleichen Rechenleistung der Server gleichmäßig verteilt.
• Weighted Round Robin - Je nach Rechenleistung können Servern unterschiedliche Gewichte zugewiesen werden.
• Geringste Verbindungen - Die Last wird auf Server mit der geringsten Anzahl aktiver Verbindungen verteilt.

In unserem Balancer implementieren wir den einfachsten Algorithmus - Round Robin.

Auswahl bei Round Robin

Der Round Robin-Algorithmus ist einfach. Es gibt allen Darstellern die gleiche Möglichkeit, Aufgaben zu erledigen.


Wählen Sie Server in Round Robin aus, um eingehende Anforderungen zu verarbeiten.

Wie in der Abbildung gezeigt, wählt der Algorithmus die Server zyklisch in einem Kreis aus. Aber wir können sie nicht direkt auswählen, oder?

Und wenn der Server lügt? Wir müssen wahrscheinlich keinen Datenverkehr dorthin senden. Das heißt, der Server kann nicht direkt verwendet werden, bis wir ihn in den gewünschten Zustand versetzt haben. Es ist erforderlich, den Datenverkehr nur an die Server zu leiten, die in Betrieb sind.

Definieren Sie die Struktur

Wir müssen alle Details im Zusammenhang mit dem Backend nachverfolgen. Sie müssen wissen, ob er lebt, und die URL verfolgen. Dazu können wir folgende Struktur definieren:

type Backend struct { URL *url.URL Alive bool mux sync.RWMutex ReverseProxy *httputil.ReverseProxy } 

Keine Sorge, ich erkläre die Bedeutung der Felder im Backend.

Jetzt müssen Sie im Balancer irgendwie alle Backends verfolgen. Dazu können Sie Slice und einen variablen Zähler verwenden. Definiere es in ServerPool:

 type ServerPool struct { backends []*Backend current uint64 } 

Verwenden von ReverseProxy

Wie wir bereits festgestellt haben, besteht das Wesen des Balancers darin, den Datenverkehr auf verschiedene Server zu verteilen und die Ergebnisse an den Client zurückzugeben. In der Go-Dokumentation heißt es:

ReverseProxy ist ein HTTP-Handler, der eingehende Anforderungen entgegennimmt und an einen anderen Server sendet und die Antworten an den Client zurücksendet.

Genau das, was wir brauchen. Das Rad muss nicht neu erfunden werden. Sie können unsere Anfragen einfach über ReverseProxy .

 u, _ := url.Parse("http://localhost:8080") rp := httputil.NewSingleHostReverseProxy(u) // initialize your server and add this as handler http.HandlerFunc(rp.ServeHTTP) 

Mit httputil.NewSingleHostReverseProxy(url) Sie ReverseProxy initialisieren, das Anforderungen an die übergebene url sendet. Im obigen Beispiel wurden alle Anforderungen an localhost: 8080 gesendet und die Ergebnisse an den Client gesendet.

Wenn Sie sich die Signatur der ServeHTTP-Methode ansehen, finden Sie die Signatur des HTTP-Handlers darin. Daher können Sie es in http an HandlerFunc .

Weitere Beispiele finden Sie in der Dokumentation .

Für unseren Balancer können Sie ReverseProxy mit der zugehörigen URL im Backend initiieren, sodass ReverseProxy Anforderungen an die URL weiterleitet.

Serverauswahlprozess

Bei der nächsten Serverauswahl müssen wir die zugrunde liegenden Server überspringen. Aber Sie müssen die Zählung organisieren.

Zahlreiche Clients stellen eine Verbindung zum Balancer her, und wenn jeder von ihnen den nächsten Knoten auffordert, Datenverkehr zu übertragen, kann ein Race Condition auftreten. Um dies zu verhindern, können wir ServerPool mit mutex blockieren. Aber es wird redundant sein, außerdem wollen wir ServerPool nicht blockieren. Wir müssen nur den Zähler um eins erhöhen.

Die beste Lösung, um diese Anforderungen zu erfüllen, wäre die atomare Inkrementierung. Go unterstützt es mit dem atomic Package.

 func (s *ServerPool) NextIndex() int { return int(atomic.AddUint64(&s.current, uint64(1)) % uint64(len(s.backends))) } 

Wir erhöhen den aktuellen Wert atomar um eins und geben den Index zurück, indem wir die Länge des Arrays ändern. Dies bedeutet, dass der Wert immer im Bereich von 0 bis zur Länge des Arrays liegen sollte. Letztendlich interessieren wir uns für einen bestimmten Index, nicht für den gesamten Zähler.

Einen Live-Server auswählen

Wir wissen bereits, dass unsere Anfragen zyklisch über alle Server hinweg rotiert werden. Und wir müssen nur den Leerlauf überspringen.

GetNext() immer einen Wert zwischen 0 und der Länge des Arrays zurück. Wir können jederzeit den nächsten Knoten abrufen, und wenn dieser inaktiv ist, müssen wir das Array als Teil der Schleife weiter durchsuchen.


Wir durchlaufen das Array.

Wie in der Abbildung gezeigt, möchten wir vom nächsten Knoten zum Ende der Liste gehen. Dies kann mit next + length . Um jedoch einen Index auszuwählen, müssen Sie ihn auf die Länge des Arrays beschränken. Dies kann einfach mit der Änderungsoperation durchgeführt werden.

Nachdem wir während der Suche einen funktionierenden Server gefunden haben, sollte dieser als aktuell markiert sein:

 // GetNextPeer returns next active peer to take a connection func (s *ServerPool) GetNextPeer() *Backend { // loop entire backends to find out an Alive backend next := s.NextIndex() l := len(s.backends) + next // start from next and move a full cycle for i := next; i < l; i++ { idx := i % len(s.backends) // take an index by modding with length // if we have an alive backend, use it and store if its not the original one if s.backends[idx].IsAlive() { if i != next { atomic.StoreUint64(&s.current, uint64(idx)) // mark the current one } return s.backends[idx] } } return nil } 

Vermeidung der Racebedingung in der Backend-Struktur

Hier müssen Sie sich an ein wichtiges Thema erinnern. Die Backend Struktur enthält eine Variable, die von mehreren Goroutinen gleichzeitig geändert oder abgefragt werden kann.

Wir wissen, dass Goroutinen die Variable mehr lesen als in sie schreiben. Aus diesem RWMutex wir uns für RWMutex entschieden, um den Zugriff auf Alive zu serialisieren.

 // SetAlive for this backend func (b *Backend) SetAlive(alive bool) { b.mux.Lock() b.Alive = alive b.mux.Unlock() } // IsAlive returns true when backend is alive func (b *Backend) IsAlive() (alive bool) { b.mux.RLock() alive = b.Alive b.mux.RUnlock() return } 

Ausgleichsanforderungen

Nun können wir eine einfache Methode formulieren, um unsere Anforderungen auszugleichen. Es wird nur fehlschlagen, wenn alle Server ausfallen.

 // lb load balances the incoming request func lb(w http.ResponseWriter, r *http.Request) { peer := serverPool.GetNextPeer() if peer != nil { peer.ReverseProxy.ServeHTTP(w, r) return } http.Error(w, "Service not available", http.StatusServiceUnavailable) } 

Diese Methode kann einfach als HandlerFunc an den HTTP-Server HandlerFunc .

 server := http.Server{ Addr: fmt.Sprintf(":%d", port), Handler: http.HandlerFunc(lb), } 

Wir leiten den Datenverkehr nur an aktive Server weiter

Unser Balancer hat ein ernstes Problem. Wir wissen nicht, ob der Server läuft. Um dies herauszufinden, müssen Sie den Server überprüfen. Hierfür gibt es zwei Möglichkeiten:

• Aktiv: Beim Ausführen der aktuellen Anforderung stellen wir fest, dass der ausgewählte Server nicht antwortet, und markieren ihn als inaktiv.
• Passiv: Sie können Server in bestimmten Abständen anpingen und den Status überprüfen.

Aktive Überprüfung laufender Server

Wenn ein Fehler ReverseProxy initiiert ErrorHandler die ErrorHandler Rückruffunktion. Dies kann verwendet werden, um Fehler zu erkennen:

 proxy.ErrorHandler = func(writer http.ResponseWriter, request *http.Request, e error) { log.Printf("[%s] %s\n", serverUrl.Host, e.Error()) retries := GetRetryFromContext(request) if retries < 3 { select { case <-time.After(10 * time.Millisecond): ctx := context.WithValue(request.Context(), Retry, retries+1) proxy.ServeHTTP(writer, request.WithContext(ctx)) } return } // after 3 retries, mark this backend as down serverPool.MarkBackendStatus(serverUrl, false) // if the same request routing for few attempts with different backends, increase the count attempts := GetAttemptsFromContext(request) log.Printf("%s(%s) Attempting retry %d\n", request.RemoteAddr, request.URL.Path, attempts) ctx := context.WithValue(request.Context(), Attempts, attempts+1) lb(writer, request.WithContext(ctx)) } 

Bei der Entwicklung dieses Fehlerhandlers haben wir die Funktionen von Closures verwendet. Dies ermöglicht es uns, externe Variablen wie Server-URLs in unserer Methode zu erfassen. Der Handler überprüft den Wiederholungszähler. Wenn er kleiner als 3 ist, senden wir dieselbe Anforderung erneut an denselben Server. Dies liegt daran, dass der Server aufgrund vorübergehender Fehler möglicherweise unsere Anforderungen verwirft, diese jedoch bald verfügbar sind (der Server verfügt möglicherweise nicht über freie Sockets für neue Clients). Sie müssen also den Verzögerungstimer für einen neuen Versuch nach ca. 10 ms einstellen. Mit jeder Anfrage erhöhen wir die Anzahl der Versuche.

Nach jedem fehlgeschlagenen Versuch markieren wir den Server als inaktiv.

Jetzt müssen Sie einen neuen Server für dieselbe Anforderung zuweisen. Wir werden dies mit dem Versuchszähler unter Verwendung des context tun. Nachdem wir die Anzahl der Versuche erhöht haben, übergeben wir sie an lb , um einen neuen Server für die Verarbeitung der Anforderung auszuwählen.

Wir können dies nicht auf unbestimmte Zeit tun, daher werden wir in lb prüfen, ob die maximale Anzahl von Versuchen erreicht wurde, bevor wir mit der Verarbeitung der Anforderung fortfahren.

Sie können einfach den Versuchszähler aus der Anfrage abrufen. Wenn er das Maximum erreicht, unterbrechen wir die Anfrage.

 // lb load balances the incoming request func lb(w http.ResponseWriter, r *http.Request) { attempts := GetAttemptsFromContext(r) if attempts > 3 { log.Printf("%s(%s) Max attempts reached, terminating\n", r.RemoteAddr, r.URL.Path) http.Error(w, "Service not available", http.StatusServiceUnavailable) return } peer := serverPool.GetNextPeer() if peer != nil { peer.ReverseProxy.ServeHTTP(w, r) return } http.Error(w, "Service not available", http.StatusServiceUnavailable) } 

Dies ist eine rekursive Implementierung.

Verwenden des Kontextpakets

Mit dem context können Sie nützliche Daten in HTTP-Anforderungen speichern. Wir werden dies aktiv nutzen, um Daten zu verfolgen, die sich auf Anfragen beziehen - Attempt und Retry .

Zunächst müssen Sie die Schlüssel für den Kontext festlegen. Es wird empfohlen, keine Zeichenfolgen, sondern eindeutige numerische Werte zu verwenden. Go verfügt über ein iota Schlüsselwort für die inkrementelle Implementierung von Konstanten, von denen jede einen eindeutigen Wert enthält. Dies ist eine großartige Lösung zum Definieren von Zifferntasten.

 const ( Attempts int = iota Retry ) 

Sie können den Wert dann extrahieren, wie wir es normalerweise mit der HashMap tun. Der Standardwert kann von der aktuellen Situation abhängen.

 // GetAttemptsFromContext returns the attempts for request func GetRetryFromContext(r *http.Request) int { if retry, ok := r.Context().Value(Retry).(int); ok { return retry } return 0 } 

Passive Serverüberprüfung

Passive Überprüfungen identifizieren und beheben Serverausfälle. Wir rufen sie in einem bestimmten Intervall an, um ihren Status zu bestimmen.

Versuchen Sie zum Pingen, eine TCP-Verbindung herzustellen. Wenn der Server antwortet, wird markiert, dass er funktioniert. Diese Methode kann angepasst werden, um bestimmte Endpunkte wie /status aufzurufen. Stellen Sie sicher, dass die Verbindung nach dem Erstellen geschlossen wird, um die zusätzliche Belastung des Servers zu verringern. Andernfalls wird er versuchen, diese Verbindung aufrechtzuerhalten und schließlich seine Ressourcen zu erschöpfen.

 // isAlive checks whether a backend is Alive by establishing a TCP connection func isBackendAlive(u *url.URL) bool { timeout := 2 * time.Second conn, err := net.DialTimeout("tcp", u.Host, timeout) if err != nil { log.Println("Site unreachable, error: ", err) return false } _ = conn.Close() // close it, we dont need to maintain this connection return true } 

Jetzt können Sie die Server durchlaufen und deren Status markieren:

 // HealthCheck pings the backends and update the status func (s *ServerPool) HealthCheck() { for _, b := range s.backends { status := "up" alive := isBackendAlive(b.URL) b.SetAlive(alive) if !alive { status = "down" } log.Printf("%s [%s]\n", b.URL, status) } } 

Um diesen Code regelmäßig auszuführen, können Sie den Timer in Go ausführen. Hiermit können Sie Ereignisse im Kanal anhören.

 // healthCheck runs a routine for check status of the backends every 2 mins func healthCheck() { t := time.NewTicker(time.Second * 20) for { select { case <-tC: log.Println("Starting health check...") serverPool.HealthCheck() log.Println("Health check completed") } } } 

In diesem Code gibt der <-tC Kanal alle 20 Sekunden einen Wert zurück. select können Sie dieses Ereignis definieren. Wenn keine default vorliegt, wird gewartet, bis mindestens ein Fall ausgeführt werden kann.

Führen Sie nun den Code in einer separaten Goroutine aus:

 go healthCheck() 

Fazit

In diesem Artikel haben wir viele Fragen untersucht:

• Round Robin Algorithmus
• ReverseProxy aus der Standardbibliothek
• Mutexe
• Atomare Operationen
• Kurzschlüsse
• Rückrufe
• Auswahloperation

Es gibt viele weitere Möglichkeiten, unseren Balancer zu verbessern. Zum Beispiel:

• Verwenden Sie Heap, um Live-Server zu sortieren und den Suchbereich zu verringern.
• Statistiken sammeln.
• Implementieren Sie den gewichteten Round-Robin-Algorithmus mit der geringsten Anzahl von Verbindungen.
• Unterstützung für Konfigurationsdateien hinzufügen.

Usw.

Der Quellcode ist hier .