Introduction à RabbitMQ dans une architecture distribuée

RabbitMQ est un agent de messages (message broker) open-source qui implémente le protocole AMQP. Dans le cadre d'un système de vente flash (Seckill), il joue un rôle crucial pour l'écrêtage des pics de charge, la gestion des files d'attente et la communication asynchrone entre les services.

Mise en œuvre de base

Pour intégrer RabbitMQ dans un projet Spring Boot, l'ajout de la dépendance suivante est nécessaire :

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

Voici un exemple de configuration pour définir une file d'attente simple :

@Configuration
public class ConfigurationRabbit {
    @Bean
    public Queue fileStandard() {
        return new Queue("file_donnees", true);
    }
}

Le service suivant permet l'envoi de messages vers l'échangeur par défaut :

@Service
@Slf4j
public class ProducteurService {
    @Autowired
    private RabbitTemplate template;

    public void envoyerMessage(Object contenu) {
        log.info("Envoi du message : {}", contenu);
        template.convertAndSend("file_donnees", contenu);
    }
}

La réception des messages s'effectue via un écouteur dédié :

@Service
@Slf4j
public class ConsommateurService {
    @RabbitListener(queues = "file_donnees")
    public void traiter(Object message) {
        log.info("Message reçu : {}", message);
    }
}

Types d'échangeurs RabbitMQ (Exchanges)

La flexibilité de RabbitMQ repose sur ses différents modes de routage :

• Mode Fanout : Diffuse le message à toutes les files reliées à l'échangeur, sans tenir compte des clés de routage. C'est le mode le plus rapide.
• Mode Direct : Le message est acheminé vers la file dont la clé de routage correspond exactement à celle fournie par le producteur.
• Mode Topic : Permet un routage basé sur des motifs (wildcards). Une file peut s'abonner à des thématiques spécifiques (ex: commande.#).
• Mode Headers : Utilise les attributs de l'en-tête du message plutôt que la clé de routage pour déterminer la destination.

Stratégies d'optimisation des API

L'objectif principal est de minimiser les accès directs à la base de données relationnelle pour éviter les goulots d'étranglement. Le flux optimisé se décline comme suit :

1. Initialisation : Chargement du stock des produits dans Redis au démarrage du système.
2. Pré-vérification : À la réception d'une requête, le stock est décrémenté atomiquement dans Redis. Si le stock est épuisé, la requête est rejetée immédiatement.
3. Mise en file d'attente : Si le stock est disponible, la demande est placée dans une file RabbitMQ et une réponse "En attente" est renvoyée au client.
4. Traitement asynchrone : Un consommateur récupère la demande, vérifie l'absence de doublons (une commande par utilisateur) et finalise la transaction en base de données.
5. Sondage (Polling) : Le client interroge périodiquement le serveur pour connaître le statut final de sa commande.

Implémentation de la vente flash asynchrone

Configuration de l'échangeur de type Topic pour les ventes flash :

@Configuration
public class ConfigVenteFlash {
    public static final String QUEUE_FLASH = "fileFlash";
    public static final String EXCHANGE_FLASH = "echangeurFlash";

    @Bean
    public Queue fileFlash() {
        return new Queue(QUEUE_FLASH);
    }

    @Bean
    public TopicExchange echangeurFlash() {
        return new TopicExchange(EXCHANGE_FLASH);
    }

    @Bean
    public Binding lierFile() {
        return BindingBuilder.bind(fileFlash()).to(echangeurFlash()).with("flash.commande.#");
    }
}

Logique du consommateur pour le traitement des commandes :

@Service
@Slf4j
public class ConsommateurFlash {
    @Autowired
    private InterfaceProduitService produitService;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private InterfaceCommandeService commandeService;

    @RabbitListener(queues = ConfigVenteFlash.QUEUE_FLASH)
    public void traiterCommandeFlash(String messageJson) {
        log.info("Traitement de la requête : {}", messageJson);
        MessageFlash message = TransformateurJson.versObjet(messageJson, MessageFlash.class);
        
        Long idProduit = message.getIdProduit();
        Utilisateur client = message.getUtilisateur();

        // Validation du stock résiduel en base
        ProduitVo produit = produitService.obtenirProduitParId(idProduit);
        if (produit.getQuantiteStock() < 1) {
            return;
        }

        // Vérification de l'existence d'une commande préalable (anti-doublon)
        String commandeExistante = (String) redisTemplate.opsForValue().get("commande:" + client.getId() + ":" + idProduit);
        if (commandeExistante != null) {
            return;
        }

        // Exécution de la transaction de vente
        commandeService.creerVenteFlash(client, produit);
    }
}