ApsaraMQ for Kafka:Gunakan SDK untuk PHP untuk mengirim dan menerima pesan

Persyaratan lingkungan

• GNU Compiler Collection (GCC) telah diinstal. Untuk informasi lebih lanjut, lihat Menginstal GCC.

• PHP telah diinstal. Untuk informasi lebih lanjut, lihat Unduhan.

• PHP Extension Community Library (PECL) telah diinstal. Untuk informasi lebih lanjut, lihat Mengunduh Ekstensi PECL.

Instal pustaka C++

1. Jalankan perintah berikut untuk beralih ke direktori repositori yum /etc/yum.repos.d/:

   cd /etc/yum.repos.d/

2. Buat file konfigurasi repositori yum bernama confluent.repo.

   [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

3. Jalankan perintah berikut untuk menginstal pustaka C++:

   yum install librdkafka-devel

Instal pustaka PHP

1. Jalankan perintah berikut untuk menginstal pustaka PHP:

   pecl install rdkafka

2. Di file inisialisasi PHP php.ini, tambahkan baris berikut untuk mengaktifkan ekstensi Kafka:

   extension=rdkafka.so

Siapkan file konfigurasi

1. (Opsional) Unduh sertifikat root Secure Sockets Layer (SSL). Jika Anda menggunakan titik akhir SSL untuk terhubung ke instance ApsaraMQ for Kafka Anda, Anda harus menginstal sertifikat tersebut.

2. Pergi ke halaman aliware-kafka-demos, klik untuk mengunduh proyek demo ke mesin lokal Anda, lalu dekompresi paket proyek demo tersebut.

3. Di dalam paket yang sudah didekompresi, masuk ke folder kafka-php-demo. Kemudian, buka folder yang sesuai berdasarkan titik akhir yang ingin Anda gunakan, dan konfigurasikan file setting.php di dalam folder tersebut.

   <?php
   
   return [
       'sasl_plain_username' => 'xxx',
       'sasl_plain_password' => 'xxx',
       'bootstrap_servers' => "xxx:xx,xxx:xx",
       'topic_name' => 'xxx',
       'consumer_id' => 'xxx'
   ];

   Parameter	Deskripsi
   sasl_plain_username	Nama pengguna Simple Authentication and Security Layer (SASL). Jika Anda menggunakan titik akhir default untuk terhubung ke instance ApsaraMQ for Kafka, parameter ini tidak tersedia. Catatan Jika fitur ACL tidak diaktifkan untuk instance ApsaraMQ for Kafka, Anda dapat memperoleh nama pengguna dan kata sandi pengguna SASL dari parameter Username dan Password di bagian Configuration Information pada halaman Instance Details di Konsol ApsaraMQ for Kafka. Jika fitur ACL diaktifkan untuk instance ApsaraMQ for Kafka, pastikan bahwa pengguna SASL memiliki izin untuk mengirim dan menerima pesan menggunakan instance tersebut. Untuk informasi lebih lanjut, lihat Berikan izin kepada pengguna SASL.
   sasl_plain_password	Kata sandi pengguna SASL. Jika Anda menggunakan titik akhir default untuk terhubung ke instance ApsaraMQ for Kafka, parameter ini tidak tersedia.
   bootstrap_servers	Titik akhir SSL dari instance ApsaraMQ for Kafka. Anda dapat memperoleh titik akhir di bagian Endpoint Information pada halaman Instance Details di Konsol ApsaraMQ for Kafka.
   topic_name	Nama topik. Anda dapat memperoleh nama topik di halaman Topics di Konsol ApsaraMQ for Kafka.
   consumer_id	ID grup. Anda dapat memperoleh ID grup di halaman Groups di Konsol ApsaraMQ for Kafka.

4. Setelah parameter yang diperlukan dikonfigurasi, unggah semua file di folder tempat file konfigurasi berada ke direktori instalasi PHP di server Anda. Folder yang sesuai dengan titik akhir SSL berisi file sertifikat root SSL.

Kirim pesan

Jalankan perintah berikut untuk menjalankan kafka-producer.php guna mengirim pesan:

php kafka-producer.php

<?php

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

$conf = new RdKafka\Conf();
$conf->set('sasl.mechanisms', 'PLAIN');
$conf->set('api.version.request', 'true');
$conf->set('sasl.username', $setting['sasl_plain_username']);
$conf->set('sasl.password', $setting['sasl_plain_password']);
$conf->set('security.protocol', 'SASL_SSL');
$conf->set('ssl.ca.location', __DIR__ . '/ca-cert.pem');
$conf->set('ssl.endpoint.identification.algorithm', 'none');
$conf->set('message.send.max.retries', 5);
$rk = new RdKafka\Producer($conf);
# jika ingin debug, atur level log ke LOG_DEBUG
$rk->setLogLevel(LOG_INFO);
$rk->addBrokers($setting['bootstrap_servers']);
$topic = $rk->newTopic($setting['topic_name']);
$a = $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Pesan halo kafka");
$rk->poll(0);
while ($rk->getOutQLen() > 0) {
    $rk->poll(50);
}
echo "pengiriman berhasil" . PHP_EOL;

Untuk informasi lebih lanjut tentang kode sampel, lihat php-rdkafka.

Berlangganan pesan

Jalankan perintah berikut untuk menjalankan kafka-consumer.php guna berlangganan pesan:

php kafka-consumer.php

<?php
$setting = require __DIR__ . '/setting.php';
$conf = new RdKafka\Conf();
$conf->set('sasl.mechanisms', 'PLAIN');
$conf->set('api.version.request', 'true');
$conf->set('sasl.username', $setting['sasl_plain_username']);
$conf->set('sasl.password', $setting['sasl_plain_password']);
$conf->set('security.protocol', 'SASL_SSL');
$conf->set('ssl.ca.location', __DIR__ . '/ca-cert.pem');

$conf->set('session.timeout.ms', 10000);

$conf->set('request.timeout.ms', 305000);

$conf->set('group.id', $setting['consumer_id']);

$conf->set('ssl.endpoint.identification.algorithm', 'none');

$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 "Menunggu penugasan partisi... (mungkin memerlukan waktu saat\n";
echo "bergabung kembali dengan grup setelah meninggalkannya.)\n";

while (true) {
    $message = $consumer->consume(30 * 1000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            var_dump($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "Tidak ada pesan lagi; akan menunggu lebih banyak\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "Waktu habis\n";
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

?>

Untuk informasi lebih lanjut tentang kode sampel, lihat php-rdkafka.