Kafka Inicio rápido

A menudo puede desear configurar un par de conectores: uno para enviar y otro para recibir mensajes de/a un destino. Esta guía explica cómo hacer esto utilizando Kafka.

Para ejemplos más completos sobre el uso de conectores para consumir y producir mensajes en general, consulte Solicitud-Respuesta Asincrónica.

Paso 1: Agregue connector-kafka dependencia

La dependencia que debe agregar a su pom.xml es:

<dependency>
    <groupId>com.iconsolutions.ipf.core.connector</groupId>
    <artifactId>connector-kafka</artifactId>
</dependency>

Si está importando el BOM desde Icon, o utilizando el BOM de Icon como padre, no es necesario proporcionar una versión separada.

Paso 2: Config

Configuración del conector-en general-es altamente impulsado por la configuración. La configuración nos permite especificar:

  • Nombres de temas

  • Configuración del consumidor/productor de Kafka

  • Reiniciar configuraciones (en caso de fallo)

Aquí tiene un ejemplo de un bloque de configuración para un par de conectores de envío y recepción que conectaremos a nuestro `ConnectorTransport`s, para el sistema de reservas de un banco ficticio que presenta un tema de solicitud y respuesta:

mybank.booking { (1)
  kafka.producer {
    topics {
      request-topic = "mybank.booking.request" (2)
    }
    kafka-clients {
      compression.type = lz4 (4)
    }
  }
  kafka.consumer {
    topics {
      response-topic = "mybank.booking.response" (3)
    }
    kafka-clients {
      group.id = ipf (4)
    }
  }
}
1 Esto se conoce como la ruta raíz de configuración y se hará referencia en el código. Indica dónde en la aplicación configuración para buscar la configuración de transporte de este conector
2 El request-topic la clave también se referenciará en el código. Esta capacidad existe si usted desea referirse a múltiples temas dentro de un común Kafka bloque de configuración. Alternativamente, puede utilizar simplemente topic = "mybank.booking.request" en lugar de un discreto topics bloque de configuración.
3 El response-topic, igual que <2> <4>`kafka-clients` le permite especificar cualquier estándar Kafka configuraciones de productor o consumidor según lo documentado aquí(productor) y aquí(consumidor)

Paso 2.1: Configuración Común

Puede haber notado que no especificamos el bootstrap.servers propiedad. Esto se debe a que-por defecto-el Kafka la configuración que usted especifique aquí volverá a la akka.kafka.consumer y akka.kafka.producer ajustes.

Lo que esto significa es que usted puede simplemente especificar dos configuraciones para configurar globalmente todos Kafka Servidores Bootstrap para todos Kafkaconsumidores y productores. Por ejemplo:

akka.kafka.consumer.kafka-clients.bootstrap.servers="kafka:9092"
akka.kafka.producer.kafka-clients.bootstrap.servers="kafka:9092"

Esto es lo mismo para todas las demás configuraciones de consumidor y productor, por ejemplo, TLS, compresión, tamaño del búfer.

Si-por ejemplo-uno específico Kafka el productor o consumidor consume/produce de/a un tema diferente, puede ser configurado como tal:

akka.kafka.consumer.kafka-clients.bootstrap.servers="kafka:9092"
akka.kafka.producer.kafka-clients.bootstrap.servers="kafka:9092"
mybank.booking.kafka.producer.kafka-clients.bootstrap.servers="someotherkafka:9094"

La configuración anterior significa que todos Kafka consumidores y productores-aparte del productor de reservas-irá a kafka:9092, y el productor de reservas irá a someotherkafka:9094.

Paso 3: Crear Send Connector Transporte

La mejor manera de crear un SendingConnectorTransport para Kafka que utiliza cadenas como tipos de clave y valor es a través del stringBuilder:

var kafkaConnectorTransport = KafkaConnectorTransport.stringBuilder(
        "accounts-booking-send-transport", (1)
        actorSystem, (2)
        "mybank.booking" (3)
).build();

Como puede ver, este generador requiere solo tres cosas:

1 El ConnectorTransport nombre; utilizar nombres significativos para los conectores facilitará la resolución de problemas
2 El ActorSystem
3 La Ruta Raíz de Configuración; utilizando el mybank.booking clave de configuración definida en Step 2 
También puede crear el transporte utilizando el constructor normal, que es la forma recomendada en caso de que desee utilizar tipos que no sean cadenas.
var producerConfig = AlpakkaConfigProvider.getProducerConfig(classicActorSystemProvider.classicSystem(), "mybank.booking");(1)
var config = new StringSerializerProducerConfig(producerConfig);

var kafkaConnectorTransport = KafkaConnectorTransport.<String, String>builder()
        .withName("accounts-booking-send-transport") (2)
        .withTopicName(config.topic())
        .withProducerSettings(config.producerSettings()) (3)
        .withRestartSettings(config.restartSettings()) (4)
        .withHealthCheckSettings(config.healthCheckSettings()) (4)
        .withProducerRecordKeyValueProvider(new StringProducerRecordKeyValueProvider()) (5)
        .withActorSystem(actorSystem)
        .build();
1 Usando el mybank.booking clave de configuración definida en Step 2
2 Asigne un nombre significativo al transporte
3 Estos son los ProducerSettings de la kafka-clients bloque
4 Prefiera la verificación de salud y reinicie la configuración desde la configuración a menos que haya una razón muy convincente para especificar la suya propia.
5 Es a través de la ProducerRecordKeyValueProvider instancia en la que podrá elegir tipos de clave-valor para su Kafka mensajes

Paso 4: Crear Receive Connector Transporte

La mejor manera de crear un ReceivingConnectorTransport para Kafka que utiliza cadenas como tipos de clave y valor es a través de la stringBuilder:

var kafkaAckReceiveConnectorTransport = KafkaAckReceiveConnectorTransport.stringBuilder(
        "accounts-booking-receive-transport", (1)
        actorSystem, (2)
        "mybank.booking" (3)
).build();

Como el constructor que se encuentra en KafkaConnectorTransport, este constructor acepta los mismos tres parámetros:

1 El nombre del ReceivingConnectorTransport; utilizar nombres significativos para los conectores facilitará la resolución de problemas.
2 El ActorSystem
3 La Ruta Raíz de Configuración; utilizando el mybank.booking clave de configuración definida en Step 2 
También puede crear el transporte utilizando el constructor normal, que es la forma recomendada en caso de que desee utilizar tipos que no sean cadenas.
var consumerConfig = AlpakkaConfigProvider.getConsumerConfig(classicActorSystemProvider.classicSystem(), "mybank.booking");(1)
var config = new StringDeserializerConsumerConfig(consumerConfig);

var kafkaAckReceiveConnectorTransport = KafkaAckReceiveConnectorTransport.<String, String>builder()
                        .withName("accounts-booking-receive-transport") (2)
                        .withActorSystem(classicActorSystemProvider)
                        .withConsumerSettings(config.consumerSettings()) (3)
                        .withTopics(config.topic("response-topic")) (4)
                        .withRestartSettings(config.restartSettings()) (5)
                        .withMetadataClientSettings(config.healthCheckSettings()) (5)
                        .build();
1 Usando el mybank.booking clave de configuración definida en Step 2
2 Asigne un nombre significativo al transporte
3 Estos son los ConsumerSettings de la kafka-clients bloque
4 Capturando response-topic desde el topics bloque. Si solo utiliza un tema, puede reemplazar esto con solo una llamada a topic()
5 Prefiera la verificación de salud y reinicie la configuración desde la configuración a menos que haya una razón muy convincente para especificar la suya propia.