Use the SDK for PHP to connect to ApsaraMQ for Kafka to send and receive messages - ApsaraMQ for Kafka

Connectez une application PHP à ApsaraMQ for Kafka pour envoyer et recevoir des messages. Les exemples de ce guide utilisent l'extension php-rdkafka, un wrapper PHP pour la bibliothèque C librdkafka.

Prérequis

Avant de commencer, assurez-vous que les logiciels suivants sont installés sur votre serveur Linux :

GNU Compiler Collection (GCC) : Requis pour compiler les extensions natives. Pour les instructions d'installation, consultez Installing GCC.
PHP : Pour les instructions de téléchargement et d'installation, consultez la page de téléchargement de PHP.
PHP Extension Community Library (PECL) : Permet d'installer les extensions PHP. Pour plus de détails, consultez Downloading PECL extensions.

Vous avez également besoin des éléments suivants :

Une instance ApsaraMQ for Kafka avec un topic et un groupe de consommateurs créés. Récupérez l'endpoint, le nom du topic et l'ID de groupe depuis la console ApsaraMQ for Kafka.
Si vous vous connectez via l'endpoint SSL : le nom d'utilisateur et le mot de passe Simple Authentication and Security Layer (SASL) de votre instance.

Étape 1 : Installer librdkafka

L'extension php-rdkafka dépend de la bibliothèque C librdkafka. Installez-la à partir du dépôt de packages Confluent.

Accédez au répertoire du dépôt yum :
```
   cd /etc/yum.repos.d/
```

Créez un fichier nommé confluent.repo avec le contenu suivant :

   [Confluent.dist]
   name=Confluent repository (dist)
   baseurl=https://packages.confluent.io/rpm/5.1/7
   gpgcheck=1
   gpgkey=https://packages.confluent.io/rpm/5.1/archive.key
   enabled=1

   [Confluent]
   name=Confluent repository
   baseurl=https://packages.confluent.io/rpm/5.1
   gpgcheck=1
   gpgkey=https://packages.confluent.io/rpm/5.1/archive.key
   enabled=1

Installez le package de développement librdkafka :
```
   yum install librdkafka-devel
```

Étape 2 : Installer l'extension php-rdkafka

Installez l'extension via PECL :
```
   pecl install rdkafka
```
Ajoutez la ligne suivante à votre fichier php.ini pour activer l'extension :
```
   extension=rdkafka.so
```

Étape 3 : Configurer les paramètres de connexion

(Facultatif) Si vous vous connectez via l'endpoint SSL, téléchargez le certificat racine SSL.
Téléchargez le projet de démonstration depuis aliware-kafka-demos et extrayez l'archive.

Dans le package extrait, accédez au dossier kafka-php-demo. Ouvrez le sous-dossier correspondant à votre type d'endpoint (SSL ou par défaut), puis modifiez le fichier setting.php :

   <?php

   return [
       'sasl_plain_username' => '<YOUR_SASL_USERNAME>',
       'sasl_plain_password' => '<YOUR_SASL_PASSWORD>',
       'bootstrap_servers'   => '<HOST1>:<PORT1>,<HOST2>:<PORT2>',
       'topic_name'          => '<YOUR_TOPIC_NAME>',
       'consumer_id'         => '<YOUR_CONSUMER_GROUP_ID>',
   ];

Remplacez les espaces réservés par vos valeurs réelles :

Espace réservé	Description	Où le trouver
`<YOUR_SASL_USERNAME>`	Nom d'utilisateur SASL. Non requis pour l'endpoint par défaut.	Instance Details > Configuration Information > Username dans la console ApsaraMQ for Kafka. Si la fonctionnalité ACL est activée, assurez-vous que l'utilisateur SASL est autorisé à envoyer et recevoir des messages. Pour plus d'informations, consultez Grant permissions to SASL users.
`<YOUR_SASL_PASSWORD>`	Mot de passe SASL. Non requis pour l'endpoint par défaut.	Même emplacement que le nom d'utilisateur.
`<HOST1>:<PORT1>,<HOST2>:<PORT2>`	Serveurs bootstrap (l'endpoint de votre instance ApsaraMQ for Kafka).	Instance Details > Endpoint Information dans la console ApsaraMQ for Kafka.
`<YOUR_TOPIC_NAME>`	Nom du topic.	Page Topics dans la console ApsaraMQ for Kafka.
`<YOUR_CONSUMER_GROUP_ID>`	ID du groupe de consommateurs.	Page Groups dans la console ApsaraMQ for Kafka.

Téléversez tous les fichiers du dossier vers le répertoire d'installation PHP sur votre serveur. Si vous utilisez l'endpoint SSL, assurez-vous d'inclure le fichier de certificat racine SSL (ca-cert.pem).

Étape 4 : Envoyer des messages

Exécutez kafka-producer.php pour envoyer un message :

php kafka-producer.php

Endpoint SSL

Utilisez ce code pour vous connecter via l'endpoint SSL avec authentification SASL :

<?php

$setting = require __DIR__ . '/setting.php';

$conf = new RdKafka\Conf();

// SASL authentication
$conf->set('sasl.mechanisms', 'PLAIN');
$conf->set('sasl.username', $setting['sasl_plain_username']);
$conf->set('sasl.password', $setting['sasl_plain_password']);

// SSL/TLS encryption
$conf->set('security.protocol', 'SASL_SSL');
$conf->set('ssl.ca.location', __DIR__ . '/ca-cert.pem');
$conf->set('ssl.endpoint.identification.algorithm', 'none');

// Producer settings
$conf->set('api.version.request', 'true');
$conf->set('message.send.max.retries', 5);

$producer = new RdKafka\Producer($conf);
$producer->setLogLevel(LOG_INFO);  // Set to LOG_DEBUG for troubleshooting
$producer->addBrokers($setting['bootstrap_servers']);

$topic = $producer->newTopic($setting['topic_name']);

// Send a message to an automatically assigned partition
$topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message hello kafka");
$producer->poll(0);

// Wait for all outstanding messages to be delivered
while ($producer->getOutQLen() > 0) {
    $producer->poll(50);
}

echo "send succ" . PHP_EOL;

Endpoint par défaut

Utilisez ce code pour vous connecter via l'endpoint par défaut sans authentification SASL :

<?php

$setting = require __DIR__ . '/setting.php';

$conf = new RdKafka\Conf();

// Producer settings
$conf->set('api.version.request', 'true');
$conf->set('message.send.max.retries', 5);

$producer = new RdKafka\Producer($conf);
$producer->setLogLevel(LOG_INFO);  // Set to LOG_DEBUG for troubleshooting
$producer->addBrokers($setting['bootstrap_servers']);

$topic = $producer->newTopic($setting['topic_name']);

// Send a message to an automatically assigned partition
$topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message hello kafka");
$producer->poll(0);

// Wait for all outstanding messages to be delivered
while ($producer->getOutQLen() > 0) {
    $producer->poll(50);
}

echo "send succ" . PHP_EOL;

Pour plus d'informations sur l'API producteur php-rdkafka, consultez php-rdkafka.

Étape 5 : S'abonner aux messages

Exécutez kafka-consumer.php pour commencer à consommer des messages :

php kafka-consumer.php

Endpoint SSL

Utilisez ce code pour vous connecter via l'endpoint SSL avec authentification SASL :

<?php

$setting = require __DIR__ . '/setting.php';

$conf = new RdKafka\Conf();

// SASL authentication
$conf->set('sasl.mechanisms', 'PLAIN');
$conf->set('sasl.username', $setting['sasl_plain_username']);
$conf->set('sasl.password', $setting['sasl_plain_password']);

// SSL/TLS encryption
$conf->set('security.protocol', 'SASL_SSL');
$conf->set('ssl.ca.location', __DIR__ . '/ca-cert.pem');
$conf->set('ssl.endpoint.identification.algorithm', 'none');

// Consumer settings
$conf->set('api.version.request', 'true');
$conf->set('group.id', $setting['consumer_id']);
$conf->set('session.timeout.ms', 10000);
$conf->set('request.timeout.ms', 305000);
$conf->set('metadata.broker.list', $setting['bootstrap_servers']);

$topicConf = new RdKafka\TopicConf();
$conf->setDefaultTopicConf($topicConf);

$consumer = new RdKafka\KafkaConsumer($conf);
$consumer->subscribe([$setting['topic_name']]);

echo "Waiting for partition assignment... (may take some time when" . PHP_EOL;
echo "quickly re-joining the group after leaving it.)" . PHP_EOL;

while (true) {
    $message = $consumer->consume(30 * 1000);  // 30-second poll timeout
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            // Successfully consumed a message
            var_dump($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "No more messages; will wait for more" . PHP_EOL;
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Timed out" . PHP_EOL;
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

?>

Endpoint par défaut

Utilisez ce code pour vous connecter via l'endpoint par défaut sans authentification SASL :

<?php

$setting = require __DIR__ . '/setting.php';

$conf = new RdKafka\Conf();

// Consumer settings
$conf->set('api.version.request', 'true');
$conf->set('group.id', $setting['consumer_id']);
$conf->set('session.timeout.ms', 10000);
$conf->set('request.timeout.ms', 305000);
$conf->set('metadata.broker.list', $setting['bootstrap_servers']);

$topicConf = new RdKafka\TopicConf();
$conf->setDefaultTopicConf($topicConf);

$consumer = new RdKafka\KafkaConsumer($conf);
$consumer->subscribe([$setting['topic_name']]);

echo "Waiting for partition assignment... (may take some time when" . PHP_EOL;
echo "quickly re-joining the group after leaving it.)" . PHP_EOL;

while (true) {
    $message = $consumer->consume(30 * 1000);  // 30-second poll timeout
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            // Successfully consumed a message
            var_dump($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "No more messages; will wait for more" . PHP_EOL;
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Timed out" . PHP_EOL;
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

?>

Pour plus d'informations sur l'API consommateur php-rdkafka, consultez php-rdkafka.

Étapes suivantes

Explorez les options de configuration et l'utilisation de l'API php-rdkafka sur le dépôt GitHub php-rdkafka.
Envoyez et recevez des messages avec des SDK pour d'autres langages. Consultez la référence développeur d'ApsaraMQ for Kafka.