ApsaraMQ for Kafka:Send and receive messages through the default endpoint

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

• Une instance ApsaraMQ for Kafka avec un topic et un groupe de consommateurs déjà créés. Consultez la rubrique Étape 3 : Créer des ressources

• JDK 1.8 ou version ultérieure installé. Consultez la page Java Downloads

• Maven 2.5 ou version ultérieure installé. Consultez la page Downloading Apache Maven

• Un IDE ou un éditeur de texte (ce tutoriel utilise IntelliJ IDEA)

Configurer le projet de démonstration

1. Téléchargez le projet aliware-kafka-demos depuis GitHub et décompressez l’archive.

2. Ouvrez le dossier kafka-java-demo dans IntelliJ IDEA ou dans votre IDE préféré. Le projet inclut des classes utilitaires telles que JavaKafkaConfigurer, qui charge les propriétés de connexion depuis le fichier kafka.properties.

Ajouter les dépendances Java

Vérifiez que les dépendances suivantes sont présentes dans le fichier pom.xml. Si ce n’est pas le cas, ajoutez-les :

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.6.0</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.6</version>
</dependency>

Configurer les propriétés de connexion

Ouvrez le fichier kafka.properties du projet de démonstration et définissez les paramètres suivants :

bootstrap.servers=<your-default-endpoint>
topic=<your-topic-name>
group.id=<your-group-id>

Toutes ces valeurs sont disponibles dans la console ApsaraMQ for Kafka.

Le paramètre group.id est requis uniquement pour les consommateurs. Les producteurs n’en ont pas besoin.

Envoyer des messages

Compilez et exécutez le fichier KafkaProducerDemo.java pour vous connecter à l’endpoint par défaut et envoyer des messages vers le topic spécifié.

L’extrait de code suivant envoie 100 messages :

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class KafkaProducerDemo {

    public static void main(String[] args) {
        // Load kafka.properties from the demo project.
        Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();

        // Default endpoint (bootstrap.servers from kafka.properties).
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                kafkaProperties.getProperty("bootstrap.servers"));

        // Serialize both keys and values as strings.
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        // Maximum wait time for a request: 30 seconds.
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);

        // Retry up to 5 times on transient failures, with a 3-second backoff.
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);

        // Create a thread-safe producer.
        // One producer per process is sufficient. For higher throughput, use up to five.
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        String topic = kafkaProperties.getProperty("topic");

        try {
            List<Future<RecordMetadata>> futures = new ArrayList<>(128);

            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> record =
                        new ProducerRecord<>(topic, "message value: " + i);
                futures.add(producer.send(record));
            }

            producer.flush();

            for (Future<RecordMetadata> future : futures) {
                try {
                    RecordMetadata metadata = future.get();
                    System.out.println("Produce ok: " + metadata.toString());
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } catch (Exception e) {
            // Troubleshoot if messages still fail after retries.
            System.out.println("Failed to send messages.");
            e.printStackTrace();
        }
    }
}

Produce ok: <your-topic>-0@15

Recevoir des messages

Le projet de démonstration inclut des exemples de consommateurs qui s’abonnent à un topic via l’endpoint par défaut. Deux approches d’abonnement sont disponibles :

Le projet de démonstration fournit des exemples de consommateurs pour les deux approches d’abonnement. Consultez les fichiers source dans le dossier kafka-java-demo pour obtenir le code complet des consommateurs.

Pour des exemples de code de consommateur dans d’autres langages, consultez la rubrique SDK overview.

Dépannage

Si des erreurs surviennent lors de la connexion ou de la livraison des messages, consultez la rubrique Que faire si une erreur est signalée sur un client lors de l’utilisation d’ApsaraMQ for Kafka ?