Django, Sellerie, Redis, RabbitMQ: Angekettete Aufgaben für Fanout-On-Writes

Question

Ich habe Rick Bransons PyCon-Video angeschaut: Messaging at Scale at Instagram. Sie können sich das Video ansehen, um diese Frage zu beantworten. Rick Branson verwendet Sellerie, Redis und RabbitMQ. Um Sie auf dem Laufenden zu halten, verfügt jeder Benutzer über eine Wiedergabeliste für seinen Homefeed. Jede Liste enthält Medien-IDs von Fotos, die von den Personen gepostet werden, denen sie folgen.

Justin Bieber zum Beispiel hat 1,5 Millionen Follower. Wenn er ein Foto postet, muss die ID dieses Fotos in jede einzelne Redis-Liste für jeden seiner Follower eingefügt werden. Dies wird Fanout-On-Write-Ansatz genannt. Bei diesem Ansatz gibt es jedoch einige Zuverlässigkeitsprobleme. Es kann funktionieren, aber für jemanden wie Justin Bieber oder Lady Gaga, die Millionen von Anhängern haben, kann dies in der Web-Anfrage (wo Sie 0-500ms haben, um die Anfrage abzuschließen) ein Problem sein. Bis dahin wird die Anfrage Timeout.

Also entschied sich Rick Branson, Sellery zu verwenden, eine asynchrone Aufgabenwarteschlange/Jobwarteschlange, die auf verteilter Nachrichtenweiterleitung basiert. Jegliches schweres Heben, wie das Einfügen von Medien-IDs in Follower-Listen, kann asynchron außerhalb der Web-Anfrage erfolgen. Die Anfrage wird abgeschlossen und Sellerie wird weiterhin die IDs in alle Listen einfügen.

Dieser Ansatz wirkt Wunder. Aber noch einmal, Sie wollen Justins Anhänger nicht in einem riesigen Brocken zu Sellery bringen, weil es einen Sellerie-Arbeiter bindet. Warum arbeiten nicht mehrere Arbeiter gleichzeitig daran, damit es schneller fertig wird? Brilliante Idee! Sie möchten diesen Teil in kleinere Stücke zerlegen und verschiedene Arbeiter an jedem Stapel arbeiten lassen. Rick Branson macht einen Stapel von 10.000 Anhängern, und er benutzt etwas, das als Cursor bezeichnet wird, um Medien-IDs für alle Justin Bieber-Anhänger bis zur Fertigstellung einzufügen. In dem Video spricht er darüber in 3:56

Ich fragte mich, ob jemand das mehr erklären könnte und Beispiele zeigen, wie es gemacht werden kann. Ich versuche gerade, das gleiche Setup zu versuchen. Ich verwende Andy McCurdis Redispy-Python-Client-Bibliothek, um mit meinem Redis-Server zu kommunizieren. Für jeden Benutzer in meinem Service erstelle ich eine Redis-Follower-Liste.

So ein Benutzer mit einer ID von 343 würde auf der folgenden Taste wird eine Liste haben:

followers:343 

Ich schaffe auch eine homefeed Liste für jeden Benutzer. Jeder Benutzer hat seine eigene Liste. So ein Benutzer mit der ID 1990 würde auf der folgenden Taste wird eine Liste haben:

homefeed:1990 

In der „Follower: 343“ redis Liste enthält alle IDs der Menschen, die 343 Benutzer 343. Benutzer folgen hat 20.007 Anhänger. Im Folgenden lese ich alle IDs in der Liste ab, beginnend mit Index 0 bis zum Ende -1, um Ihnen zu zeigen, wie es aussieht.

>>> r_server.lrange("followers:343", 0, -1) 
['8', '7', '5', '3', '65', '342', '42', etc...] ---> for the sake of example, assume this list has another 20,000 IDs. 

Was Sie sehen, ist eine Liste aller IDs von Benutzern, die Benutzer folgen 343.

Hier ist meine proj/mydjangoapp/tasks.py die meine insert_into_homefeed Funktion enthält:

from __future__ import absolute_import 
from celery import shared_task 
import redis 
pool = redis.ConnectionPool(host='XX.XXX.XXX.X', port=6379, db=0, password='XXXXX') 

@shared_task 
def insert_into_homefeed(photo_id, user_id): 
    # Grab the list of all follower IDs from Redis for user_id. 
    r_server = redis.Redis(connection_pool=pool) 

    followers_list = r_server.lrange("followers:%s" % (user_id), 0, -1) 

    # Now for each follower_id in followers_list, find their homefeed key 
    # in Redis and insert the photo_id into that homefeed list. 

    for follower_id in followers_list: 
     homefeed_list = r_server.lpush("homefeed:%s" % (follower_id), photo_id) 
    return "Fan Out Completed for %s" % (user_id) 

Bei dieser Aufgabe werden beim Aufruf aus der Django-Ansicht alle IDs der Personen abgerufen, die dem Benutzer 343 folgen, und die Foto-ID wird dann in alle ihre Homefeed-Listen eingefügt.

Hier ist meine Upload-Ansicht in meiner proj/mydjangoapp/views.py.Ich nenne grundsätzlich Sellerie Verzögerung Methode und geben die notwendigen Variablen so, dass die Anforderung schnell endet:

# Import the Celery Task Here 
from mydjangoapp.tasks import insert_into_homefeed 


@csrf_exempt 
def Upload(request): 
    if request.method == 'POST': 
     data = json.loads(request.body) 
     newPhoto = Photo.objects.create(user_id = data['user_id'], description= data['description'], photo_url = data['photo_url']) 
     newPhoto_ID = newPhoto.pk 
     insert_into_homefeed.delay(newPhoto_ID, data['user_id']) 
     return HttpResponse("Request Completed") 

Wie kann ich dies tun, so dass es von 10.000 chargiert werden?

Answer 1

Der im Video beschriebene Ansatz ist Aufgabe "Verketten".

Um Ihre Task-Methode als Kette einzurichten und auszuführen, möchten Sie einen zusätzlichen Parameter hinzufügen, der den Index in der Liste der Follower darstellt. Anstatt an der vollständigen Liste der Follower zu arbeiten, arbeitet die Task nur mit einer festen Batchgröße, beginnend mit dem Indexargument, das übergeben wurde. Nach Beendigung sollte die Aufgabe eine neue Aufgabe erstellen und den neuen Index übergeben.

INSERT_INTO_HOMEFEED_BATCH = 10000 

@shared_task 
def insert_into_homefeed(photo_id, user_id, index=0): 
    # Grab the list of all follower IDs from Redis for user_id. 
    r_server = redis.Redis(connection_pool=pool) 

    range_limit = index + INSERT_INTO_HOMEFEED_BATCH - 1 # adjust for zero-index 

    followers_list_batch = r_server.lrange("followers:%s" % (user_id), index, range_limit) 

    if not followers_list_batch: 
     return # zero followers or no more batches 

    # Now for each follower_id in followers_list_batch, find their homefeed key 
    # in Redis and insert the photo_id into that homefeed list. 
    for follower_id in followers_list: 
     homefeed_list = r_server.lpush("homefeed:%s" % (follower_id), photo_id) 

    insert_into_homefeed.delay(photo_id, user_id, range_limit + 1) 

Das funktioniert gut, weil Redis lists are ordered und der lrange Befehl doesn't return an error on out-of-range inputs.