Consumo de datos de Kafka
¿Cómo sabe el consumidor qué partición consumir? ¿Cómo cambia la asignación de partición cuando cambia el número de consumidores en el grupo de consumidores?
Divida el número total de consumidores y el número total de particiones para obtener un intervalo, y luego distribuya las particiones uniformemente según el intervalo para garantizar que las particiones se distribuyan a todos los consumidores de la manera más uniforme posible. Para cada tema, esta estrategia clasificará a todos los consumidores del grupo de consumidores según el orden lexicográfico del nombre y luego le dará a cada consumidor un rango de partición fijo. Si no se distribuye uniformemente, al consumidor con el orden lexicográfico más alto se le asignará una partición extra.
Suponiendo que n = número de particiones/número de consumidores, m = número de particiones y consumidores, entonces a los primeros m consumidores se les asignan n 1 particiones y a cada consumidor posterior se le asignan n particiones.
Como se muestra en la figura, hay 7 particiones en el tema. En este momento, solo hay un consumidor C0 en el grupo de consumidores y C0 está suscrito a 7 particiones.
A medida que los consumidores se involucran más en el grupo de consumidores, las particiones se asignan gradualmente de C0 a C1 ~ C6. Cuando se une el último consumidor C7, siempre hay 8 consumidores pero solo 7 particiones, por lo que C7 no puede consumir ningún mensaje porque no puede asignar particiones.
Cuantos más consumidores, mejor. El número de consumidores debe ser menor o igual que el número de particiones; de lo contrario, se desperdiciarán recursos.
Desventajas:
Cuando un grupo de consumidores se suscribe a dos temas y tiene cuatro particiones, los resultados de la asignación de particiones son los siguientes, que son relativamente uniformes.
Sin embargo, cuando dos preguntas tienen tres particiones cada una, ocurrirán los siguientes problemas. Si situaciones similares se amplían, puede haber un problema de sobrecarga del consumidor.
Las particiones de todos los consumidores en el grupo de consumidores y todos los temas suscritos por el consumidor se clasifican en orden lexicográfico y luego las particiones se asignan a cada consumidor por turno mediante sondeo. Si la información de suscripción de los consumidores del grupo de consumidores es la misma, la distribución de la partición será más uniforme. Por ejemplo, dos consumidores de un grupo de consumidores se suscriben a los temas de dos particiones utilizando 3, como se muestra en la figura.
Pero cuando la información de suscripción de los consumidores en el grupo de consumidores es diferente, ocurrirá el problema de la distribución desigual. Como se muestra en la figura, suponga que el grupo de consumidores tiene tres consumidores y que los temas 1/2/3 tienen 1/2/3 particiones respectivamente. C0 se suscribe al tema 1, C1 se suscribe a los temas 1 y 2 y C2 se suscribe al tema 1/2/3. Los resultados de la partición son los siguientes.
Esta estrategia se presentó más adelante y su objetivo principal es:
Supongamos que tres consumidores se suscriben a cuatro temas y cada tema tiene dos particiones, entonces los resultados iniciales de la asignación de particiones son los siguientes:
A primera vista, el resultado es el mismo que el de la estrategia de asignación RoundRobin, pero en este momento, si C1 se desconecta, el grupo de consumidores realizará una operación de reequilibrio y redistribuirá las particiones de mensajes. Si se trata de una estrategia de asignación circular, el resultado de la asignación es el siguiente:
Si se trata de una estrategia de asignación fija, el resultado es el siguiente:
StickyAssignor conserva los resultados finales de la asignación de C0 y C2, y asigna la partición de C1 a C0 y C2 para equilibrarla.
Si se produce una redistribución de partición, para la misma partición, el consumidor anterior y el consumidor recién designado pueden ser diferentes, y el procesamiento del consumidor anterior se repetirá en el consumidor recién designado, lo que resultará en un consumo repetido. . La estrategia de asignación StickyAssignor, al igual que su nombre "Sticky", hace que la estrategia de asignación sea "fija" y hace que las asignaciones anteriores y posteriores sean lo más iguales posible, reduciendo así la pérdida de recursos del sistema y otras situaciones anormales.
Echemos un vistazo a la situación en la que la información de suscripción del consumidor es diferente, tomando el ejemplo en RoundRobinAssignor como ejemplo.
Supongamos que el grupo de consumidores tiene tres consumidores y que los temas 1/2/3 tienen 1/2/3 particiones respectivamente.
C0 se suscribe al tema 1, C1 se suscribe a los temas 1 y 2 y C2 se suscribe al tema 1/2/3. El resultado de la partición de RoundRobinAssignor será
Cuando se utiliza StickyAssignor, el resultado de la asignación de la partición es el siguiente. siguiente:
Si C0 está fuera de línea en este momento, los resultados de la reasignación de RoundRobinAssignor son los siguientes:
Los resultados de la reasignación de StickyAssignor son los siguientes:
En resumen:
Estrategia de asignación de StickyAssignor La ventaja es que puede hacer que la reasignación de particiones sea "fija" y reducir los movimientos de partición innecesarios (una partición se elimina de un consumidor antes de asignarse a otro nuevo consumidor).
El consumo de mensajes de Kafka se basa en el modelo pull.
Kafka extrae un conjunto de mensajes a la vez, y cada mensaje tiene el siguiente formato:
Cada vez que se recupera un método, se devuelve un conjunto de mensajes no utilizados. Para implementar esta función, necesita conocer el desplazamiento del consumo durante el último consumo. El desplazamiento del consumo debe persistir en el tema __consumer_offsets.
El desplazamiento máximo del mensaje de extracción actual es x. El desplazamiento del envío de finalización del consumo del consumidor es en realidad x 1, que representa la posición inicial del siguiente mensaje de extracción.
Envío automático
El envío automático se adopta de forma predeterminada y el desplazamiento máximo de mensajes de cada partición se envía cada 5 segundos de forma predeterminada. La acción de envío real se completa en la lógica de extraer el mensaje. Antes de extraer cada mensaje, se juzgará si se puede enviar el desplazamiento y, de ser así, se enviará el último desplazamiento. Aquí surgirán dos problemas, como se muestra en la siguiente figura.
Consumo repetido: Actualmente tirando mensajes X 2 y X 7, el consumo actual llega a X 5. Antes de enviar el desplazamiento del consumo, el consumidor está hacia abajo; el nuevo consumidor aún extraerá mensajes de X 2, lo que resultará en un consumo repetido.
Pérdida de mensajes: actualmente extrayendo mensajes X 2 y X 7, actualmente consumiendo X 5. En el siguiente tirón, el desplazamiento consumido se ha comprometido a X 8. Si el consumidor deja de funcionar en este momento, el nuevo consumidor comenzará a consumir desde X 8, lo que provocará que los mensajes de X 5 a X 7 no se consuman, lo que provocará la pérdida de mensajes.
Envío manual
Envío sincrónico y envío asincrónico.
El valor predeterminado para el envío sincrónico es el desplazamiento máximo del mensaje de partición de extracción. Por ejemplo, para los mensajes de extracción de X 2 y X 7, la posición de envío predeterminada para el envío sincrónico es. se consume una vez. Debido a que el envío en sí es una operación sincrónica, consumirá una cierta cantidad de rendimiento.
El envío sincrónico también puede provocar problemas de consumo repetido, como el cierre del consumidor después de que se completa el consumo y antes del envío.
El subproceso del consumidor de envío asincrónico no se bloqueará y mejorará el rendimiento. Sin embargo, los reintentos de envíos asincrónicos fallidos pueden causar que se sobrescriba el desplazamiento de envío. Por ejemplo, si el envío asincrónico offset=X falla esta vez. , el siguiente envío asincrónico no se bloqueará. La confirmación offset=X y es exitosa en este punto, se volverá a intentar la confirmación y se volverá a confirmar offset=x; Si no hay un reintento de verificación en el negocio, la compensación se sobrescribirá, lo que eventualmente provocará un consumo repetido.
Cuando se establece un nuevo grupo de consumidores, el consumidor se suscribe a un nuevo tema o la información de desplazamiento enviada anteriormente se elimina debido a su vencimiento, el desplazamiento de consumo registrado no se puede encontrar en este momento. Kafka se puede configurar para comenzar a consumir desde la versión más reciente o más temprana.
Kafka también admite consumos a partir de una cilindrada concreta y puede realizar consumos retroactivos. Kafka proporciona internamente el método Seek() para restablecer el desplazamiento del consumo.
Cuando necesite rastrear mensajes después de un tiempo específico, primero puede usar el método offsetsForTimes para encontrar el desplazamiento del primer mensaje después del tiempo especificado y luego usar buscar para restablecer el desplazamiento.
La propiedad de la partición se transfiere de un consumidor a otro, asegurando la alta disponibilidad y escalabilidad del grupo de consumidores, permitiéndonos eliminar o agregar consumidores de forma cómoda y segura.
Kfaka proporciona un Coordinador de grupo y un Coordinador de consumidores. El primero es responsable de gestionar los grupos de consumidores y el segundo es responsable de interactuar con los primeros. La responsabilidad más importante de ambos es la de encargarse de las operaciones de reequilibrio.
Por ejemplo, cuando un consumidor se une a un grupo de consumidores, el consumidor, el grupo de consumidores y el coordinador del grupo generalmente pasan por las siguientes etapas.
Fase 1 (encontrar el coordinador)
Los consumidores deben determinar el agente donde se encuentra el Coordinador de Grupo correspondiente al grupo de consumidores al que pertenecen y crear una conexión de red para comunicarse con el agente.
El consumidor enviará una FindCoordinatorRequest a un nodo del clúster para encontrar el coordinador del grupo correspondiente.
Kafka calcula el número de partición en __consumer_offsets en función del valor hash de coordinador_key (es decir, groupld) en la solicitud, como se muestra en la siguiente figura. Después de encontrar la partición correspondiente, busque el nodo intermediario donde se encuentra la copia líder de la partición, que es el nodo coordinador del grupo donde se encuentra el grupo de consumidores actual.
El plan de asignación de partición final del grupo de consumidores y la información de reemplazo de consumo enviada por los consumidores del grupo se enviarán al nodo del agente. El nodo intermediario no solo desempeña la función de GroupCoordinato, sino que también desempeña la función de guardar el plan de asignación de particiones y el reemplazo de consumidores dentro del grupo, lo que puede ahorrar muchos gastos generales innecesarios causados por la rotación intermedia.
La segunda etapa (unirse al grupo)
Después de encontrar con éxito el Coordinador de grupo correspondiente al grupo de consumidores, el consumidor ingresará a la etapa de unirse al grupo de consumidores. En esta etapa, el consumidor enviará una solicitud JoinGroupRequest al Coordinador de grupo y procesará la respuesta.
Las siguientes cosas se hacen principalmente dentro del coordinador del grupo:
* * * *Elegir al líder del grupo de consumidores.
Si el grupo actual no tiene un líder de grupo, la primera persona en unirse al grupo de consumidores será el líder del grupo. Si el líder cuelga, el coordinador del grupo seleccionará la primera clave del HashMap mantenido internamente (información del consumidor, la clave es member_id) como el nuevo líder.
Elegir la estrategia de asignación de particiones
Como se mencionó anteriormente, cada consumidor puede informar múltiples estrategias de asignación de particiones. El proceso de selección es el siguiente:
La tercera fase (sincronización). Grupo)
El consumidor líder implementa la asignación de particiones de acuerdo con la política de asignación de particiones obtenida en la segunda fase y luego sincroniza los resultados de la asignación con el coordinador del grupo. Cada consumidor enviará una solicitud SyncGroupRequest al coordinador del grupo para sincronizar el plan de asignación.
La estructura de la solicitud es como se muestra en la figura. La solicitud enviada por el líder tendrá group_assignment.
Contiene el plan de asignación específico correspondiente a cada consumidor, donde member_id representa el identificador único del consumidor y member_assignment es el plan de asignación correspondiente al consumidor, como se muestra en la figura.
Después de recibir el plan de asignación de partición específico, el consumidor iniciará la tarea de latido y enviará solicitudes de latido al coordinador del grupo con regularidad para asegurarse de que esté en línea.
La cuarta etapa (latido del corazón)
Antes del consumo formal, el consumidor también necesita determinar la posición inicial del mensaje. Suponiendo que el último desplazamiento de consumo se haya enviado con éxito antes, el consumidor solicitará obtener el último desplazamiento de consumo enviado y continuar consumiendo desde aquí.
El hilo de latido es un hilo independiente que se puede enviar entre mensajes de sondeo. Si un consumidor deja de enviar latidos durante un tiempo suficiente, el coordinador del grupo considerará que el consumidor está muerto y desencadenará un comportamiento de reequilibrio.