ApsaraMQ for Kafka:Best practices for producers

Envoyer un message

Toute interaction avec un producteur commence par producer.send(), qui accepte un ProducerRecord contenant le topic, la partition, l'horodatage, la clé et la valeur.

Future<RecordMetadata> metadataFuture = producer.send(new ProducerRecord<String, String>(
        topic,   // The message topic.
        null,   // The partition number. Set this to null to let the producer assign a partition.
        System.currentTimeMillis(),   // The timestamp.
        String.valueOf(value.hashCode()),   // The message key.
        value   // The message value.
));

La méthode send() est asynchrone. Pour bloquer l'exécution jusqu'à l'obtention du résultat, appelez :

RecordMetadata metadata = metadataFuture.get(timeout, TimeUnit.MILLISECONDS);

Pour des exemples complets de SDK, consultez Présentation du SDK.

Clé et valeur

Les messages dans ApsaraMQ for Kafka 0.10.2 comportent deux champs :

• Key : identifiant du message. Définissez une clé unique par message afin de suivre son cycle de vie via les journaux d'envoi et de consommation.

• Value : contenu du message.

Pour les envois à haut volume ne nécessitant pas de routage basé sur une clé, omettez la clé et utilisez la stratégie de partitionnement sticky. Cela améliore l'efficacité du regroupement par lots.

Sécurité des threads

Une instance de producteur est thread-safe et peut envoyer des messages vers n'importe quel topic. Utilisez un seul producteur par application.

Configurer les nouvelles tentatives

Dans un environnement distribué, les envois peuvent échouer en raison de problèmes réseau. L'échec peut se produire à deux niveaux : le message a atteint le broker, mais l'accusé de réception (ACK) a été perdu, ou le message n'a jamais atteint le broker.

ApsaraMQ for Kafka utilise une architecture réseau d'adresse IP virtuelle (VIP) qui ferme automatiquement les connexions inactives. Les clients inactifs reçoivent souvent une erreur connection reset by peer. Les nouvelles tentatives permettent de gérer ces échecs temporaires.

Accusés de réception

Le paramètre acks contrôle le nombre de réplicas qui doivent confirmer une écriture avant que le broker ne réponde. Choisissez un paramétrage en fonction de vos exigences en matière de durabilité et de performance.

Pour améliorer les performances d'envoi, définissez acks=1.

Optimiser le regroupement par lots

Le producteur regroupe les messages destinés à la même partition en lots avant l'envoi. Des lots plus grands réduisent le nombre de requêtes réseau, diminuent l'utilisation du CPU et améliorent le débit et la latence. De petits lots provoquent une mise en file d'attente des requêtes côté client et serveur.

Deux paramètres contrôlent le regroupement par lots :

Un lot est envoyé dès que l'un ou l'autre des seuils est atteint, selon le premier événement survenant. Pour obtenir un bon équilibre entre débit et latence, définissez batch.size=16384 et linger.ms=1000.

Utilisez un client de version 2,4 ou ultérieure, qui active par défaut la stratégie de partitionnement sticky afin de réduire davantage les envois fragmentés.

Stratégie de partitionnement sticky

Seuls les messages envoyés vers la même partition sont regroupés dans un lot ; la stratégie de partitionnement affecte donc directement l'efficacité du regroupement par lots.

public class MyStickyPartitioner implements Partitioner {

    // Records the time of the last partition switch.
    private long lastPartitionChangeTimeMillis = 0L;
    // Records the current partition.
    private int currentPartition = -1;
    // The partition switch interval. Set the interval as needed.
    private long partitionChangeTimeGap = 100L;

    public void configure(Map<String, ?> configs) {}

    /**
     * Compute the partition for the given record.
     *
     * @param topic The topic name
     * @param key The key to partition on (or null if no key)
     * @param keyBytes serialized key to partition on (or null if no key)
     * @param value The value to partition on or null
     * @param valueBytes serialized value to partition on or null
     * @param cluster The current cluster metadata
     */
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {

        // Get all partition information.
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (keyBytes == null) {
            List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
            int availablePartitionSize = availablePartitions.size();

            // Check the current active partitions.
            if (availablePartitionSize > 0) {
                handlePartitionChange(availablePartitionSize);
                return availablePartitions.get(currentPartition).partition();
            } else {
                handlePartitionChange(numPartitions);
                return currentPartition;
            }
        } else {
            // For a message with a key, select a partition based on the key's hash value.
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }

    private void handlePartitionChange(int partitionNum) {
        long currentTimeMillis = System.currentTimeMillis();

        // If the time since the last switch exceeds the switch interval, switch to the next partition. Otherwise, use the current partition.
        if (currentTimeMillis - lastPartitionChangeTimeMillis >= partitionChangeTimeGap
            || currentPartition < 0 || currentPartition >= partitionNum) {
            lastPartitionChangeTimeMillis = currentTimeMillis;
            currentPartition = Utils.toPositive(ThreadLocalRandom.current().nextInt()) % partitionNum;
        }
    }

    public void close() {}

}

Prévenir les erreurs de mémoire insuffisante (OOM)

Le producteur met en cache les messages en mémoire avant d'envoyer les lots. Si le cache devient trop volumineux, une erreur de mémoire insuffisante (OOM) se produit.

Ordre des partitions

Au sein d'une seule partition, les messages sont stockés et consommés dans l'ordre où ils ont été envoyés.

Par défaut, ApsaraMQ for Kafka ne garantit pas un ordre strict au sein d'une partition. Lors d'une mise à niveau ou d'un basculement, un petit nombre de messages peut être désordonné lorsqu'ils sont redirigés vers une autre partition. Ce compromis améliore la disponibilité.

Pour imposer un ordre strict au sein d'une partition, sélectionnez local storage lors de la création du topic.