Kafka'ya Giriş

giriiş

Apache Kafka, Java dilinde yazılmış ve gerçek zamanlı veri akışlarını işlemek için tasarlanmış, açık kaynaklı, dağıtılmış bir olay ve akış işleme platformudur. Yüksek verimlilik ve kullanılabilirlik ile doğası gereği ölçeklenebilirdir. Apache Yazılım Vakfı tarafından geliştirilen Kafka, güvenilirliği, kullanım kolaylığı ve hata toleransı nedeniyle yaygın olarak benimsenmiştir. Dünyanın en büyük kuruluşları tarafından büyük miktarda veriyi dağıtık ve verimli bir şekilde yönetmek için kullanılmaktadır.

Bu eğitimde, Apache Kafka'yı indirip kuracaksınız. Verilen betikleri kullanarak konu oluşturmayı ve silmeyi, ayrıca etkinlik göndermeyi ve almayı öğreneceksiniz. Ayrıca, aynı amaca sahip benzer projeler hakkında bilgi edinecek ve Kafka'nın diğer projelerle nasıl karşılaştırıldığını öğreneceksiniz.

Ön koşullar

• En az 4 GB RAM ve 2 CPU'ya sahip bir cihaz.
• Droplet veya yerel makinenize Java 8 veya üzeri yüklü olmalıdır.

Adım 1 – Apache Kafka'yı İndirin ve Yapılandırın

Bu bölümde, Apache Kafka'yı bilgisayarınıza indirip çıkaracaksınız. Daha fazla güvenlik için, kendi kullanıcı hesabınız altında kuracaksınız. Ardından, KRaft kullanarak yapılandırıp çalıştıracaksınız.

Öncelikle Kafka'nın çalışacağı ayrı bir kullanıcı oluşturun. Aşağıdaki komutu çalıştırarak, şu adla bir kullanıcı oluşturacaksınız: Kafka Yaratmak:

sudo adduser kafka

Hesap şifreniz istenecektir. Güçlü bir şifre girin ve Enter'a basın. GİRMEK Her alan için ek bilgi girmeyi atlayın.

Son olarak, belirli Kafka kullanıcısına geçin:

su kafka

Ardından, resmi indirme sayfasından Kafka sürüm paketini indireceksiniz. Yazım sırasında en son sürüm 3.6.1'dir. macOS veya Linux kullanıyorsanız, Kafka'yı curl ile indirebilirsiniz.

Kafka'yı indirmek ve kurmak için bu komutu kullanın /geçici Yer:

curl -o /tmp/kafka.tgz https://dlcdn.apache.org/kafka/3.6.1/kafka_2.13-3.6.1.tgz

Aşağıdaki sürüme sahipsiniz ~/kafka, ana dizine kaydedeceksiniz. Şunu çalıştırarak oluşturun:

mkdir ~/kafka

Daha sonra çalıştırarak, ~/kafka Çıkarmak:

tar -xzf /tmp/kafka.tgz -C ~/kafka --strip-components=1

İndirdiğiniz arşiv, Kafka sürümüyle aynı ada sahip bir kök klasör içerdiğinden, –strip-components=1 bunu atlayacak ve içindeki her şeyi çıkaracaktır.

Bu yazının yazıldığı tarihte, Kafka 3, meta veri yönetimi için iki sistemi destekleyen son büyük sürümdü: Apache ZooKeeper ve Kafka KRaft (Kafka Raft'ın kısaltması). ZooKeeper, uygulamalar için dağıtılmış verileri koordine etmenin standart bir yolunu sağlayan ve Apache Yazılım Vakfı tarafından geliştirilen açık kaynaklı bir projedir.

Ancak Kafka 3.3 sürümünden itibaren KRaft desteği de sunuldu. KRaft, yalnızca Kafka örneklerini koordine etmek için özel olarak tasarlanmış bir sistemdir, kurulum sürecini basitleştirir ve çok daha fazla ölçeklenebilirlik sağlar. KRaft ile Kafka, yönetimsel meta verileri harici olarak tutmak yerine verilerin tüm sorumluluğunu üstlenir.

ZooKeeper desteğinin Kafka 4 ve sonrasında kaldırılması bekleniyor. Bu eğitimde, KRaft kullanarak Kafka'yı kuracaksınız.

Yeni Kafka kümeniz için benzersiz bir tanımlayıcı oluşturmanız gerekiyor. Şimdilik yalnızca bir düğümden oluşacak. Kafka'nın şu anda bulunduğu dizine gidin:

cd ~/kafka

Kafka, KRaft ile kendini yapılandırır config/kraft/server.properties kaydederken, yapılandırma dosyası ZooKeeper yapılandırması/sunucu.özellikleri Öyledir.

İlk kez çalıştırmadan önce, varsayılan ayarlardan bazılarını geçersiz kılmanız gerekir. Dosyayı düzenlemek için şu komutu çalıştırarak açın:

nano config/kraft/server.properties

...
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs
...

Ayarlar log.dirs Kafka'nın günlük dosyalarını nerede sakladığını belirtir. Varsayılan olarak, bunları şurada depolar: /tmp/kafka-günlükleri geçici de olsa yazılabilir oldukları garanti edildiğinden, bunları kaydeder. Değeri belirtilen yol ile değiştirin:

...
############################# Log Basics #############################
# A comma separated list of directories under which to store log files
log.dirs=/home/kafka/kafka-logs
...

Kafka için ayrı bir kullanıcı oluşturduğunuz için, günlük dizini yolunu kullanıcının ana dizininin altına yerleştireceksiniz. Eğer mevcut değilse, Kafka bunu oluşturacaktır. İşiniz bittiğinde dosyayı kaydedip kapatın.

Artık Kafka'yı yapılandırdığınıza göre, rastgele bir küme kimliği oluşturmak için aşağıdaki komutu çalıştırın:

KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

Daha sonra aşağıdaki komutu çalıştırıp ID'yi girerek günlük dosyaları için depolama alanı oluşturun:

bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

Çıktı şu şekilde olacaktır:

Output
Formatting /home/kafka/kafka-logs with metadata.version 3.6-IV2.

Son olarak Kafka sunucusunu ilk kez başlatabilirsiniz:

bin/kafka-server-start.sh config/kraft/server.properties

Son çıktı şuna benzer olacaktır:

Output
...
[2024-02-26 10:38:26,889] INFO Awaiting socket connections on 0.0.0.0:9092. (kafka.network.DataPlaneAcceptor)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Waiting for all of the authorizer futures to be completed (kafka.server.BrokerServer)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Finished waiting for all of the authorizer futures to be completed (kafka.server.BrokerServer)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Waiting for all of the SocketServer Acceptors to be started (kafka.server.BrokerServer)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Finished waiting for all of the SocketServer Acceptors to be started (kafka.server.BrokerServer)
[2024-02-26 10:38:26,890] INFO [BrokerServer id=1] Transition from STARTING to STARTED (kafka.server.BrokerServer)
[2024-02-26 10:38:26,891] INFO Kafka version: 3.6.1 (org.apache.kafka.common.utils.AppInfoParser)
[2024-02-26 10:38:26,891] INFO Kafka commitId: 5e3c2b738d253ff5 (org.apache.kafka.common.utils.AppInfoParser)
[2024-02-26 10:38:26,891] INFO Kafka startTimeMs: 1708943906890 (org.apache.kafka.common.utils.AppInfoParser)
[2024-02-26 10:38:26,892] INFO [KafkaRaftServer nodeId=1] Kafka Server started (kafka.server.KafkaRaftServer)

Çıktı, Kafka'nın KRaft kullanarak başarıyla başlatıldığını ve bağlantılar oluşturduğunu gösteriyor. 0.0.0.0:9092 Kabul eder.

Ne zaman CTRL + C tuşuna bastığınızda işlem sonlandırılacaktır. Kafka'yı oturum açıkken çalıştırmak tercih edilmediğinden, bir sonraki adımda Kafka'yı arka planda çalıştıracak bir servis oluşturacaksınız.

Adım 2 – Kafka için bir systemd hizmeti oluşturun

Bu bölümde, Kafka'yı her zaman arka planda çalıştıracak bir systemd hizmeti oluşturacaksınız. Systemd hizmetleri sürekli olarak başlatılabilir, durdurulabilir ve yeniden başlatılabilir.

Hizmet yapılandırmasını şu adlı bir dosyaya koyun: kod-sunucusu.hizmeti Listede /lib/systemd/sistem Sen nerede tasarruf edersen sistemd Hizmetlerinizi depolar. Metin düzenleyicinizi kullanarak oluşturun:

sudo nano /etc/systemd/system/kafka.service

Aşağıdaki satırları ekleyin:

[Unit]
Description=kafka-server
[Service]
Type=simple
User=kafka
ExecStart=/bin/sh -c '/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/kraft/server.properties > /home/kafka/kafka/kafka.log 2>&1'
ExecStop=/home/kafka/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal
[Install]
WantedBy=multi-user.target

Burada önce hizmet açıklamasını belirtin. Daha sonra [Hizmet] Hizmet türünü tanımlarsınız (basit, komutun basit bir şekilde çalıştırılması gerektiği anlamına gelir) ve çalıştırılacak komutu sağlarsınız. Ayrıca, çalıştırılacak kullanıcıyı da belirtirsiniz. Kafka ve Kafka'dan çıkıldığında servisin otomatik olarak yeniden başlatılması gerekir.

Bölüm [düzenlemek] sunucunuza giriş yapmak mümkün olduğunda sisteme bu hizmeti başlatmasını söyler. İşlem tamamlandığında dosyayı kaydedip kapatın.

Aşağıdaki komutu çalıştırarak Kafka servisini başlatın:

sudo systemctl start kafka

Doğru şekilde başlatıldığını durumunu görüntüleyerek kontrol edin:

sudo systemctl status kafka

Aşağıdakine benzer bir çıktı göreceksiniz:

Output
● kafka.service - kafka-server
Loaded: loaded (/etc/systemd/system/kafka.service; disabled; preset: enabled)
Active: active (running) since Mon 2024-02-26 11:17:30 UTC; 2min 40s ago
Main PID: 1061 (sh)
Tasks: 94 (limit: 4646)
Memory: 409.2M
CPU: 10.491s
CGroup: /system.slice/kafka.service
├─1061 /bin/sh -c "/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/kraft/server.properties > /home/kafka/kafka/kafka.log 2>&1"
└─1062 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true "-Xlog:gc*:file=/home/kafka/kafka/bin/../logs/kaf>
Feb 26 11:17:30 kafka-test1 systemd[1]: Started kafka.service - kafka-server.

Sunucu yeniden başlatıldıktan sonra Kafka'nın otomatik olarak başlaması için aşağıdaki komutu çalıştırarak hizmetini etkinleştirin:

sudo systemctl enable kafka

Bu noktada, Kafka için bir systemd hizmeti oluşturup etkinleştirdiniz, böylece her sunucu açılışında başlatılacak. Şimdi, Kafka'da konu oluşturmayı ve silmeyi, ayrıca mevcut betikleri kullanarak metin mesajlarını nasıl oluşturup kullanacağınızı öğreneceksiniz.

Adım 3 – Güncel mesajların üretilmesi ve tüketilmesi

Artık bir Kafka sunucusu kurduğunuza göre, konular ve sağlanan betikleri kullanarak bunları nasıl yöneteceğiniz hakkında bilgi sahibi olacaksınız. Ayrıca bir konudan nasıl mesaj gönderip alacağınızı da öğreneceksiniz. Olay Akışı makalesinde açıklandığı gibi, mesaj yayınlamak ve almak konularla ilgilidir. Bir konu, mesajın ait olduğu bir kategoriyle ilişkilendirilebilir.

Sağlanan komut dosyasından kafka-topics.sh Kafka'da konuları şu şekilde yönetebilirsiniz: CLI adlı bir konu oluşturmak için kullanılır ilk konu Aşağıdaki komutu çalıştırın:

bin/kafka-topics.sh --create --topic first-topic --bootstrap-server localhost:9092

Sağlanan tüm Kafka betikleri, sunucu adresinin belirtilmesini gerektirir --önyükleme sunucusu Belirtin.

Çıktı şu şekilde olacaktır:

Output
Created topic first-topic.

Mevcut tüm konuları listelemek yerine --yaratmak İçinde --liste Göndermek:

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

Oluşturduğunuz konuyu görüyorsunuz:

Output
first-topic

Konuyla ilgili detaylı bilgi ve istatistiklere şuradan ulaşabilirsiniz: --betimlemek Elde etmek:

bin/kafka-topics.sh --describe --topic first-topic --bootstrap-server localhost:9092

Çıktı şu şekilde görünecektir:

Output
Topic: first-topic TopicId: VtjiMIUtRUulwzxJL5qVjg PartitionCount: 1 ReplicationFactor: 1 Configs: segment.bytes=1073741824
Topic: first-topic Partition: 0 Leader: 1 Replicas: 1 Isr: 1

İlk satır, konu adını, kimliğini ve tekrarlama faktörünü belirtir; konu yalnızca geçerli bilgisayarda mevcut olduğundan 1'dir. İkinci satır özellikle girintilidir ve konunun ilk (ve tek) bölümü hakkında bilgi gösterir. Kafka, konuyu bölümlere ayırmanıza olanak tanır; bu da bir konunun farklı bölümlerinin farklı sunuculara dağıtılabileceği ve böylece ölçeklenebilirliğin artırılabileceği anlamına gelir. Burada yalnızca bir bölüm vardır.

Artık bir konu oluşturduğunuza göre, kafka-console-producer.sh betiğini kullanarak bu konu için mesajlar üreteceksiniz. Üreticiyi başlatmak için aşağıdaki komutu çalıştırın:

bin/kafka-console-producer.sh --topic first-topic --bootstrap-server localhost:9092

Boş bir bildirim göreceksiniz:

>

Yapımcı SMS'inizi bekliyor. Teste girin ve GİRMEK 'e basın. Bildirim şu şekilde görünecektir:

>test
>

Üretici şu anda bir sonraki mesajı bekliyor, yani önceki mesaj Kafka'ya başarıyla iletildi. Test için istediğiniz sayıda mesaj girebilirsiniz. Üreticiden çıkmak için: CTRL+C Basmak .

Bir konudan gelen mesajları almak için bir tüketiciye ihtiyacınız vardır. Kafka, şu biçimde basit bir tüketici sağlar: kafka-konsol-tüketici.sh Sunar. Çalıştırmak için şunu çalıştırın:

bin/kafka-console-consumer.sh --topic first-topic --bootstrap-server localhost:9092

Ancak herhangi bir çıktı alınmayacaktır. Bunun nedeni, tüketicinin konudan veri akışı yapması ve şu anda hiçbir şey üretilmemesi veya gönderilmemesidir. Tüketici başlamadan önce ürettiğiniz mesajları tüketmek için, konuyu baştan itibaren şu komutu çalıştırarak okumanız gerekir:

bin/kafka-console-consumer.sh --topic first-topic --from-beginning --bootstrap-server localhost:9092

Tüketici tüm konu olaylarını tekrar oynatır ve mesajları getirir:

Outputtest
...

İnşaatçı gibi çıkmak için, CTRL+C Basmak .

Tüketicinin gerçekten veri akışı sağladığını doğrulamak için, ayrı bir terminal oturumunda açın. İkincil bir SSH oturumu açın ve tüketiciyi varsayılan yapılandırmada çalıştırın:

bin/kafka-console-consumer.sh --topic first-topic --bootstrap-server localhost:9092

İlk oturumda, oluşturucuyu çalıştırın:

bin/kafka-console-producer.sh --topic first-topic --bootstrap-server localhost:9092

Ardından istediğiniz mesajları girin:

>second test
>third test
>

Bunların tüketici tarafından hemen alındığını göreceksiniz:

Output
second test
third test

Test tamamlandıktan sonra hem üreticiyi hem de tüketiciyi sonlandırın.

İlk konuyu silmek için, --silmek ile kafka-topics.sh Transfer:

bin/kafka-topics.sh --delete --topic first-topic --bootstrap-server localhost:9092

Çıktı alınmayacak. Konuların gerçekten silindiğini doğrulamak için listeleyebilirsiniz:

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

Çıktı şu şekilde olacaktır:

Output
__consumer_offsets

__Consumer_Equivalent, bir tüketicinin bir konuyu ne kadar süreyle okuduğunu kaydeden dahili bir Kafka konusudur. Bu noktada, bir Kafka konusu oluşturup içinde mesajlar oluşturdunuz. Ardından, verilen betiği kullanarak mesajları tükettiniz ve sonunda gerçek zamanlı olarak aldınız. Bir sonraki adımda, Kafka'nın diğer olay aracıları ve benzer yazılımlarla nasıl karşılaştırıldığını öğreneceksiniz.

Benzer mimarilerle karşılaştırma

Apache Kafka, olay akışı kullanım durumları için gerçek bir çözüm olarak kabul edilir. Bununla birlikte, Apache Pulsar ve RabbitMQ da yaygın olarak kullanılır ve yaklaşımlarında farklılıklar olsa da çok yönlü seçenekler olarak öne çıkar. Mesaj kuyrukları ve olay akışları arasındaki temel fark, birincisinin temel görevinin, mesajları istemcilere, sıralarından bağımsız olarak mümkün olan en hızlı şekilde iletmek olmasıdır. Bu tür sistemler genellikle mesajları, tüketiciler tarafından onaylanana kadar bellekte saklar. Mesaj filtreleme ve yönlendirme, tüketiciler belirli veri kategorilerine ilgi gösterebildikleri için önemli hususlardır. RabbitMQ, birden fazla tüketicinin bir konuya abone olabileceği ve bir mesajın birden fazla kopyasını alabileceği geleneksel bir mesajlaşma sisteminin güçlü bir örneğidir. Diğer yandan olay akışı, kalıcılığa odaklanır. Olaylar arşivlenmeli, düzenli bir şekilde saklanmalı ve bir kez işlenmelidir. Tüm tüketicilerin olayları aynı şekilde işlemesi fikri nedeniyle, belirli tüketicilere yönlendirilmeleri önemli değildir. Apache Pulsar, Apache Yazılım Vakfı tarafından geliştirilen ve olay akışını destekleyen açık kaynaklı bir mesajlaşma sistemidir. Pulsar, sıfırdan inşa edildiği Kafka'nın aksine, başlangıçta geleneksel bir mesaj kuyruğu çözümü olarak ortaya çıkmış ve daha sonra olay akışı yetenekleri kazanmıştır. Bu nedenle Pulsar, ayrı uygulamalar dağıtmaya gerek kalmadan her iki yöntemin bir kombinasyonuna ihtiyaç duyulduğunda kullanışlıdır.

Sonuç

Artık sunucunuzun arka planında güvenli bir şekilde çalışan ve bir sistem hizmeti olarak yapılandırılmış Apache Kafka'ya sahipsiniz. Ayrıca, komut satırından konuları nasıl yöneteceğinizi, mesajları nasıl oluşturup kullanacağınızı da öğrendiniz. Ancak, Kafka'nın asıl cazibesi, uygulamalarınıza entegre edebileceğiniz geniş istemci yelpazesidir.