ApsaraMQ for Kafka:Mejores prácticas para productores

Envío de un mensaje

Toda interacción del productor comienza con producer.send(), que acepta un objeto ProducerRecord con el topic, la partición, la marca de tiempo, la clave y el valor.

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.
));

El método send() es asíncrono. Para bloquear la ejecución hasta obtener el resultado, llame a:

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

Para consultar ejemplos completos del SDK, consulte SDK descripción general.

Clave y valor

Los mensajes en ApsaraMQ for Kafka 0.10.2 tienen dos campos:

• Clave: identificador del mensaje. Establezca una clave única por mensaje para rastrear su ciclo de vida en los registros de envío y consumo.

• Valor: contenido del mensaje.

Para envíos de alto volumen sin enrutamiento basado en claves, omita la clave y use la sticky partitioning strategy. Esto mejora la eficiencia del procesamiento por lotes.

Seguridad de hilos

Una instancia de productor es segura para hilos y puede enviar mensajes a cualquier topic. Use un solo productor por aplicación.

Configuración de reintentos

En un entorno distribuido, los envíos pueden fallar debido a problemas de red. El error puede ocurrir en dos puntos: el mensaje llegó al broker pero se perdió la confirmación (ACK), o bien el mensaje nunca llegó al broker.

ApsaraMQ for Kafka usa una arquitectura de red de dirección IP virtual (VIP) que cierra automáticamente las conexiones inactivas. Los clientes inactivos suelen recibir un error de connection reset by peer. Los reintentos permiten manejar estos fallos transitorios.

Confirmaciones

El parámetro acks controla cuántas réplicas deben confirmar una escritura antes de que el broker responda. Elija una configuración según sus requisitos de durabilidad y rendimiento.

Para mejorar el rendimiento de envío, establezca acks=1.

Optimización del procesamiento por lotes

El productor agrupa los mensajes destinados a la misma partición en lotes antes de enviarlos. Los lotes más grandes reducen las solicitudes de red, disminuyen el uso de CPU y mejoran el throughput y la latencia. Los lotes pequeños provocan colas de solicitudes tanto en el cliente como en el servidor.

Dos parámetros controlan el procesamiento por lotes:

Un lote se envía cuando se alcanza cualquiera de los umbrales, lo que ocurra primero. Para lograr un buen equilibrio entre throughput y latencia, establezca batch.size=16384 y linger.ms=1000.

Use un cliente de la versión 2.4 o posterior, que habilita la sticky partitioning strategy de forma predeterminada para reducir aún más los envíos fragmentados.

Estrategia de particionamiento adhesivo

Solo los mensajes enviados a la misma partición se agrupan en un lote, por lo que la estrategia de particionamiento afecta directamente la eficiencia del procesamiento por lotes.

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() {}

}

Prevención de errores de memoria insuficiente (OOM)

El productor almacena mensajes en caché en la memoria antes de enviar los lotes. Si la caché crece demasiado, se produce un error de memoria insuficiente (OOM).

Orden de las particiones

Dentro de una sola partición, los mensajes se almacenan y consumen en el orden en que se envían.

De forma predeterminada, ApsaraMQ for Kafka no garantiza un orden estricto dentro de una partición. Durante una actualización o un failover, una pequeña cantidad de mensajes puede desordenarse al redirigirse a otra partición. Esta compensación mejora la disponibilidad.

Para aplicar un orden estricto dentro de una partición, seleccione local storage al crear el topic.