Kunci untuk menghasilkan dan memakan mesej Apache Kafka menggunakan Java adalah dengan betul mengkonfigurasi API pengeluar dan pengguna dan memahami proses asas mereka. 1. Pertama tambahkan kebergantungan klien Kafka untuk memastikan versi itu serasi dengan kluster; 2. Apabila menulis pengeluar, konfigurasikan bootstrap.servers, key.serializer dan value.serializer, dan buat contoh Kafkaproducer untuk menghantar mesej, perhatikan untuk menutup sumber dan pemprosesan panggilan balik pilihan; 3. Apabila menulis pengguna, konfigurasikan kumpulan.id, deserializer, dan lain -lain, gunakan Kafkaconsumer untuk melanggan topik dan tarik mesej dalam gelung, perhatikan strategi mengemukakan offset; 4. Masalah biasa termasuk konflik kumpulan.id, penyerahan offset yang tidak sesuai, dan penutupan sumber yang salah, dan lain-lain. Adalah disyorkan untuk menguji proses pengesahan dalam persekitaran yang berdiri sendiri terlebih dahulu.

Ia sebenarnya tidak sukar untuk menghasilkan dan menggunakan mesej Apache Kafka di Java. Inti terletak pada pemahaman konsep asas Kafka dan menguasai penggunaan API pengeluar dan pengguna. Selagi konfigurasi itu betul dan struktur kod jelas, penghantaran dan penerimaan mesej dapat dicapai dengan mudah.

Tambah Kafka Dependencies

Sebelum anda mula menulis kod, anda perlu memperkenalkan perpustakaan klien Kafka ke dalam projek anda. Jika anda menggunakan Maven, anda boleh menambah kebergantungan berikut dalam pom.xml :

 <geterlihat>
    <groupId> org.apache.kafka </groupId>
    <stifactid> Kafka-clients </artifactid>
    <versi> 3.6.0 </versi>
</ketergantungan>

Versi ini agak baru dan agak stabil. Sudah tentu, anda juga boleh memilih versi klien yang sepadan berdasarkan versi kluster Kafka anda.

Tulis pengeluar kafka

Tugas pengeluar adalah menghantar mesej ke topik Kafka. Langkah -langkah utama termasuk mengkonfigurasi sifat, mewujudkan contoh pengeluar, membina rekod mesej dan menghantar.

Item konfigurasi utama adalah:

• bootstrap.servers : Alamat Broker Kafka
• key.serializer dan value.serializer : Tentukan kaedah serialisasi kunci dan nilai. StringSerializer biasanya digunakan.

Kod sampel adalah seperti berikut:

 Sifat props = sifat baru ();
props.put ("bootstrap.servers", "localhost: 9092");
props.put ("key.serializer", "org.apache.kafka.common.serialization.stringserializer");
props.put ("value.serializer", "org.apache.kafka.common.serialization.stringserializer");

Pengeluar <string, string> pengeluar = kafkaproducer baru <> (props);
ProduceRrecord <string, string> record = produceRrecord baru <> ("my-topic", "kekunci", "nilai");

producer.send (rekod);
producer.close ();

Beberapa perkara yang perlu diperhatikan:

• Lebih baik memanggil close() setelah menghantar untuk mengelakkan kebocoran sumber
• Sekiranya anda ingin mengesahkan sama ada penghantaran berjaya, anda boleh menambah fungsi panggil balik.send .send(record, callback)
• Sekiranya ia hanya ujian, anda boleh menghilangkan kunci

Tulis pengguna Kafka

Pengguna bertanggungjawab membaca mesej dari topik Kafka. Berbanding dengan pengeluar, logik pengguna sedikit lebih rumit kerana ia memerlukan secara aktif menarik mesej, memproses offset, dll.

Konfigurasi utama termasuk:

• bootstrap.servers : Juga tentukan alamat broker
• group.id : ID Kumpulan Pengguna, mesti ditetapkan, jika tidak, ralat akan dilaporkan
• key.deserializer dan value.deserializer : deserializer, biasanya menggunakan StringDeserializer

Proses pengguna yang mudah adalah seperti berikut:

 Sifat props = sifat baru ();
props.put ("bootstrap.servers", "localhost: 9092");
props.put ("kumpulan.id", "kumpulan ujian");
props.put ("key.deserializer", "org.apache.kafka.common.serialization.stringdeserializer");
props.put ("value.deserializer", "org.apache.kafka.common.serialization.stringDeserializer");

KafKaconsumer <String, String> Consumer = New KafKaconsumer <> (props);
consumer.subscribe (Collections.SingLetOnList ("My-Topic"));

sementara (benar) {
    ConsumerRecords <string, string> records = consumer.poll (duration.ofmillis (100));
    untuk (consumerRecord <string, string> rekod: rekod) {
        System.out.println ("Diterima:" Record.Value ());
    }
}

Catatan:

• Kaedah poll() akan menarik sekumpulan data dan memprosesnya dalam gelung
• Jangan lupa untuk mengemukakan offset selepas penggunaan, anda boleh memilih untuk menghantar secara automatik atau secara manual
• Sekiranya anda mahu mengambil hanya sekali dan kemudian keluar, anda boleh memecahkan gelung selepas diproses

Soalan dan langkah berjaga -jaga yang sering ditanya

Kadang -kadang anda akan mendapati bahawa mesej itu belum diterima atau telah dimakan berulang kali, yang biasanya disebabkan oleh sebab -sebab berikut:

• Kumpulan Pengguna.ID Menetapkan Kesalahan atau Konflik
• Keras frekuensi offset secara automatik terlalu tinggi atau terlalu rendah
• Kegagalan untuk menutup pengeluar atau pengguna dengan betul menghasilkan status yang tidak konsisten
• Broker Kafka tidak dimulakan atau rangkaian tidak disambungkan

Adalah disyorkan untuk terlebih dahulu menjalankan persekitaran yang berdiri sendiri semasa pembangunan dan kemudian pergi ke kluster.

Pada dasarnya itu sahaja. Proses menyambung ke Kafka di Java tidak rumit, tetapi beberapa butiran mudah diabaikan, seperti strategi bersiri, kumpulan.id, dan penyerahan. Selagi anda memberi perhatian kepada konfigurasi dan proses, anda boleh berjalan dengan cepat.

Atas ialah kandungan terperinci Bagaimana untuk menghasilkan dan mengambil mesej dari Apache Kafka dengan Java?. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!