L'algorithme Snowflake, développé par Twitter, a pour objectif de générer un entier de 64 bits unique dans un système distribué. Voici sa structure typique :

• 1 bit : Généralement réservé au signe, non utilisé dans la pratique
• 41 bits : Horodatage, permettant de représsembler environ 69 années. Si on définit une date de départ (ex: 2023), cela couvre jusqu'en 2092
• 10 bits : Identifiant de machine, permettant d'adresser jusqu'à 1024 machines. Souvent divisé en 5 bits pour le centre de données et 5 bits pour l'identifiant machine au sein du centre
• 12 bits : Compteur séquentiel, permettant de générer jusqu'à 4095 ID distincts par milliseconde par machine. Au-delà, il faut attendre la milliseconde suivante

Cette répartition des bits peut être adaptée en fonction des besoins spécifiques d'une application. Prenons un exemple concret :

Scénario d'application

Une application actuelle avec un QPS de 100 000, prévue évoluer à plusieurs millions dans les prochaines années. L'infrastructure est déployée dans trois centres de données (Paris, Lyon, Marseille) avec environ 15 machines actuellement, prévu passer à une centaine.

Pour ce cas, on pourrait répartir différemment les 62 bits utilisés (en excluant le bit de signe) :

• 10 bits pour le compteur séquentiel (pour gérer jusqu'à 1024 requêtes par milliseconde)
• 3 bits pour identifier les centres de données (8 possibilités)
• 7 bits pour l'identifiant machine (128 possibilités)
• 41 bits pour l'horodatage
• 1 bit réservé pour d'éventuelles extensions futures

Problème du retour en arrière de l'horloge

En raison de problèmes système, l'horloge d'une machine peut revenir en arrière. Comme l'algorithme Snowflake dépend fortement du temps, un tel événement pourrait générer des ID dupliqués. Notre implémentation vérifie si l'heure actuelle est inférieure à l'enregistrement précédent, et dans ce cas, lève une expection.

Implémentation Python

# Implémentation de l'algorithme Snowflake pour la génération d'ID distribués
# Basé sur le concept original de Twitter

import time
import logging

from .exceptions import HorlogeInvalide


# Configuration des bits pour l'ID de 64 bits
BITS_MACHINE = 5
BITS_CENTRE = 5
BITS_SEQUENCE = 12

# Calcul des valeurs maximales
VALEUR_MAX_MACHINE = -1 ^ (-1 << BITS_MACHINE)
VALEUR_MAX_CENTRE = -1 ^ (-1 << BITS_CENTRE)

# Décalages pour le positionnement des bits
DECALAGE_MACHINE = BITS_SEQUENCE
DECALAGE_CENTRE = BITS_SEQUENCE + BITS_MACHINE
DECALAGE_TEMPS = BITS_SEQUENCE + BITS_MACHINE + BITS_CENTRE

# Masque de séquence
MASQUE_SEQUENCE = -1 ^ (-1 << BITS_SEQUENCE)

# Époque de référence (janvier 2010)
EPOQUE_REFERENCE = 1288834974657


logger = logging.getLogger('generateur_id')


class GenerateurID(object):
    """
    Classe pour générer des ID uniques distribués
    """

    def __init__(self, id_centre, id_machine, sequence=0):
        """
        Initialisation du générateur
        :param id_centre: Identifiant du centre de données
        :param id_machine: Identifiant de la machine
        :param sequence: Valeur initiale du compteur séquentiel
        """
        # Vérifications des paramètres
        if id_machine > VALEUR_MAX_MACHINE or id_machine < 0:
            raise ValueError('Identifiant machine hors limites')

        if id_centre > VALEUR_MAX_CENTRE or id_centre < 0:
            raise ValueError('Identifiant centre hors limites')

        self.id_machine = id_machine
        self.id_centre = id_centre
        self.sequence = sequence

        self.dernier_temps = -1  # Dernier horodatage enregistré

    def _obtenir_temps(self):
        """
        Génère un horodatage en millisecondes
        :return: Horodatage entier
        """
        return int(time.time() * 1000)

    def generer_id(self):
        """
        Génère un nouvel ID unique
        :return: ID unique
        """
        temps_actuel = self._obtenir_temps()

        # Vérification du retour en arrière de l'horloge
        if temps_actuel < self.dernier_temps:
            logger.error('L\'horloge recule. Rejet des requêtes jusqu\'au {}'.format(self.dernier_temps))
            raise HorlogeInvalide

        if temps_actuel == self.dernier_temps:
            self.sequence = (self.sequence + 1) & MASQUE_SEQUENCE
            if self.sequence == 0:
                temps_actuel = self._attendre_prochaine_ms(self.dernier_temps)
        else:
            self.sequence = 0

        self.dernier_temps = temps_actuel

        nouvel_id = ((temps_actuel - EPOQUE_REFERENCE) << DECALAGE_TEMPS) | (self.id_centre << DECALAGE_CENTRE) | \
                 (self.id_machine << DECALAGE_MACHINE) | self.sequence
        return nouvel_id

    def _attendre_prochaine_ms(self, dernier_temps):
        """
        Attend la prochaine milliseconde si nécessaire
        """
        temps = self._obtenir_temps()
        while temps <= dernier_temps:
            temps = self._obtenir_temps()
        return temps


if __name__ == '__main__':
    generateur = GenerateurID(1, 2, 0)
    print(generateur.generer_id())

Dans le même dossier, créez un fichier exceptions.py :

class HorlogeInvalide(Exception):
    """
    Exception levée en cas de retour en arrière de l'horloge système
    """
    pass

Enfin, ajoutez les paramètres de configuration dans votre fichier de configuration :

# Paramètres du générateur d'ID Snowflake
ID_CENTRE_DONNEES = 0
ID_MACHINE = 0
VALEUR_SEQUENCE = 0