- \n
- Zookeeper nedir<\/li>\n
- Docker \u00fczerine zookeeper kurulumu<\/li>\n
- Docker \u00fczerine kafka kurulumu<\/li>\n
- C# client’\u0131 ile kafka \u00fczerinde \u00f6rnek<\/li>\n<\/ol>\n
![\"docker-logo\"](\"data:image\/svg+xml;base64,PHN2ZyB3aWR0aD0iMSIgaGVpZ2h0PSIxIiB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciPjwvc3ZnPg==\")
<\/a><\/p>\nKurulum i\u015flemine ba\u015flamak i\u00e7in\u00a0Docker Quickstart Terminal\u2019i a\u00e7al\u0131m ve a\u015fa\u011f\u0131daki komut sat\u0131r\u0131n\u0131 \u00e7al\u0131\u015ft\u0131ral\u0131m.<\/p>\n
docker search kafka<\/pre>\n
Komutunu \u00e7al\u0131\u015ft\u0131r\u0131lmas\u0131ndan sonra Docker Hub \u00fczerinde bir den \u00e7ok Kafka image’i oldu\u011funu g\u00f6rebilirsiniz. Ben a\u015fa\u011f\u0131daki image’lerden “ches\/kafka<\/em>” olan\u0131n\u0131 kullanaca\u011f\u0131m.<\/p>\n
<\/a><\/p>\nBu image’in detaylar\u0131na ise buradan<\/a>\u00a0ula\u015fabilirsiniz. Kendisi zookeeper ba\u011f\u0131ms\u0131z bir service olarak \u00e7al\u0131\u015fmaktad\u0131r ve configuration’lar\u0131 parameterized bir hale getirilmi\u015ftir.<\/p>\n
Dilerseniz biraz zookeeper’dan bahsedelim.<\/p>\n
1) Zookeeper Nedir?<\/h4>\n
Zookeeper ne yapar? K\u0131saca zookeeper, distributed<\/strong> uygulamalar geli\u015ftirmeye izin veren, distributed bir koordinasyon<\/strong> servisidir diyebiliriz. Bir ba\u015fka de\u011fi\u015fle, multiple instance’l\u0131 distributed sistemleri configure etmeye yarayan open-source bir projedir diyebiliriz. Bu proje hadoop alt\u0131nda geli\u015ftirilmeye ba\u015flay\u0131p, sonras\u0131nda ise \u00fcst seviye bir Apache projesi haline gelmi\u015ftir.<\/p><\/blockquote>\n
Zookeeper’\u0131n ne oldu\u011fundan biraz bahsetti\u011fimize g\u00f6re dilerseniz bunun Kafka ile ne alakas\u0131 var k\u0131sm\u0131na bir bakal\u0131m.<\/p>\n
Kafka’ya giri\u015f makalesinde de bahsetti\u011fimiz gibi kafka, tamamen distributed<\/strong> bir sistem \u00fczerinde ve \u00e7al\u0131\u015fmaktad\u0131r. Kafka ise burada zookeeper’\u0131 \u00e7e\u015fitli configuration bilgilerini store etmek i\u00e7in kullanmaktad\u0131r. Evet, bu c\u00fcmleden, kafka’n\u0131n zookeeper ile beraber \u00e7al\u0131\u015fmas\u0131n\u0131n zorunlu<\/strong> oldu\u011funu \u00e7\u0131kartabiliriz. Kafka Getting Started documentation’a bakt\u0131\u011f\u0131n\u0131zda ise zaten \u015fu c\u00fcmleye denk geleceksiniz:<\/p>\n
Step 2: Start the server<\/p>\n
Kafka uses zookeeper so you need to first start a zookeeper server<\/strong> if you don’t already have one. You can use the convenience script packaged with kafka to get a quick-and-dirty single-node zookeeper instance.<\/p><\/blockquote>\n
Kafka broker’lar\u0131 membership & failure detection, leader se\u00e7imi gibi konularda zookeeper’\u0131 kullanmaktad\u0131r. Kafka’n\u0131n zookeeper ile ne t\u00fcr bilgiler tuttu\u011funu merak ediyor iseniz,\u00a0buraya<\/a> bakabilirsiniz.<\/p>\n
2) Docker \u00dczerine Zookeeper Kurulumu<\/h4>\n
<\/a><\/p>\nZookeeper’s\u0131z bir kafka kullanamayaca\u011f\u0131m\u0131zdan bahsetti\u011fimize g\u00f6re, docker \u00fczerinden \u00f6ncelikle a\u015fa\u011f\u0131daki komut ile “jplock\/zookeeper” image’ini download edelim ve container’\u0131 aya\u011fa kald\u0131ral\u0131m.<\/p>\n
docker run -d --name zookeeper --publish 2181:2181 jplock\/zookeeper:latest<\/pre>\n
“-d” komutu ile container’\u0131n arkaplanda\u00a0\u00e7al\u0131\u015fmas\u0131 gerekti\u011fini, isminin zookeeper olaca\u011f\u0131n\u0131 ve “–publish” komutu ile de container’\u0131n “2181” port’unu\u00a0host’a publish ediyoruz.<\/p>\n
<\/a><\/p>\nKomutu \u00e7al\u0131\u015ft\u0131rd\u0131\u011f\u0131m\u0131zda “jplock\/zookeeper” image’ini bulamad\u0131\u011f\u0131ndan, latest versiyonu i\u00e7in docker hub \u00fczerinden pull i\u015flemini ger\u00e7ekle\u015ftirdi ve vermi\u015f oldu\u011fumuz configuration bilgileri ile container’\u0131 olu\u015fturdu. A\u015fa\u011f\u0131daki komutu \u00e7al\u0131\u015ft\u0131rarak aktif olan container’a bir bakal\u0131m.<\/p>\n
docker ps<\/pre>\n
\n<\/a><\/a><\/p>\n“f92916096ad7” id si ile “zookeeper” isminde container haz\u0131r durumdad\u0131r. \u015eimdi kafka kurulumuna ge\u00e7ebiliriz.<\/p>\n
3) Docker \u00dczerine Kafka Kurulumu<\/h4>\n
<\/a><\/p>\nMakale giri\u015finde “ches\/kafka<\/em>” image’ini kullanaca\u011f\u0131m\u0131zdan bahsetmi\u015ftik. Kurulum s\u0131ras\u0131nda zookeeper’da oldu\u011fu gibi ilgili port’u bind edip, kafka container’\u0131n\u0131 zookeeper’a link’leyece\u011fiz. Link’leme i\u015flemi sayesinde iki container birbirleri ile konu\u015fabilir hale geleceklerdir.<\/p>\n
Kurulum i\u015flemi i\u00e7in a\u015fa\u011f\u0131daki komut sat\u0131r\u0131n\u0131 terminal \u00fczerinden \u00e7al\u0131\u015ft\u0131ral\u0131m.<\/p>\n
docker run --name kafka -d -p \"9092:9092\" -e KAFKA_ADVERTISED_HOST_NAME=192.168.99.100 --link zookeeper:zookeeper ches\/kafka<\/pre>\n
Burada container’a kafka ismini verip, “-p” komutu ile kafka’n\u0131n default portu olan “9092” yi host’a bind ediyoruz. Ard\u0131ndan “-e” komutu ile\u00a0bir “KAFKA_ADVERTISED_HOST_NAME”\u00a0environment’\u0131 set ediyoruz. Bu de\u011feri docker’\u0131n ip’si olan “192.168.99.100” olarak belirliyoruz. Ip adresi olarak localhost yerine bunu belirtmemizin sebebi ise, kafka’y\u0131 multiple brokers \u00e7al\u0131\u015ft\u0131rabilmektir. Linkleme i\u015flemini ise “–link” key’i ile ger\u00e7ekle\u015ftiriyoruz.<\/p>\n
<\/a><\/p>\n“ches\/kafka” image’i daha \u00f6nce sistemde olmad\u0131\u011f\u0131 i\u00e7in yine docker hub \u00fczerinden pull i\u015flemleri ger\u00e7ekle\u015ferek, vermi\u015f oldu\u011fumuz configuration ayarlar\u0131 ile birlikte container haz\u0131r bir hale gelmi\u015ftir. Tekrardan a\u015fa\u011f\u0131daki komutu \u00e7al\u0131\u015ft\u0131ral\u0131m ve container’lar\u0131n son halini bir g\u00f6relim.<\/p>\n
docker ps<\/pre>\n
<\/a><\/p>\nYukar\u0131daki resimde g\u00f6rd\u00fc\u011f\u00fcm\u00fcz gibi “kafka” ve “zookeeper” isminde iki adet container aktif durumdad\u0131r. Docker \u00fczerindeki kurulum i\u015flemlerimizi tamamlad\u0131k ve art\u0131k kafka client’\u0131n\u0131 kullanarak, “192.168.99.100” numaral\u0131 ip si ve bind etmi\u015f oldu\u011fumuz “9092” portu \u00fczerinden kafka’ya ba\u011flanabilmeye haz\u0131r durumday\u0131z.<\/p>\n
4) C# ile Kafka Client Kullan\u0131m\u0131<\/h4>\n
Ger\u00e7ekle\u015ftirecek oldu\u011fumuz \u00f6rnekte Apache Kafka protocol’unu implemente eden,\u00a0kafka-net<\/strong> isimli k\u00fct\u00fcphaneyi kullanaca\u011f\u0131z. Bu k\u00fct\u00fcphane github \u00fczerinde Jroland<\/em>\u00a0isimli bir kullan\u0131c\u0131 taraf\u0131ndan geli\u015ftirilmektedir.<\/p>\n
“KafkaClientExample” isminde bo\u015f bir solution olu\u015fturup, i\u00e7erisine “KafkaProducerExample” isminde bir console application ekliyorum\u00a0ve ard\u0131ndan Nuget Package Manager’a girerek a\u015fa\u011f\u0131da bulunan\u00a0paketi projeye dahil ediyorum.<\/p>\n
<\/a><\/p>\n\u0130lgili\u00a0kafka client paketinin kurulumundan sonra ise “TestProducer” isminde yeni bir class ekleyelim\u00a0ve isminden de anla\u015f\u0131labilece\u011fi gibi Producer’\u0131 a\u015fa\u011f\u0131daki gibi kodlamaya ba\u015flayal\u0131m.<\/p>\n
using KafkaNet;\r\nusing KafkaNet.Protocol;\r\n\r\nnamespace KafkaProducerExample\r\n{\r\n public class TestProducer\r\n {\r\n private readonly IBrokerRouter _brokerRouter;\r\n\r\n public TestProducer(IBrokerRouter brokerRouter)\r\n {\r\n _brokerRouter = brokerRouter;\r\n }\r\n\r\n public void SendMessageAsync(string topic, string message)\r\n {\r\n var producer = new Producer(_brokerRouter);\r\n\r\n producer.SendMessageAsync(topic, new[] { new Message(message), }).Wait();\r\n }\r\n }\r\n}<\/pre>\nConstructor arac\u0131l\u0131\u011f\u0131 ile inject etmi\u015f oldu\u011fumuz “IBrokerRouter”, message’lar\u0131n do\u011fru kafka partition k\u0131sm\u0131na\u00a0gidebilmeleri i\u00e7in\u00a0metadata tabanl\u0131\u00a0routing i\u015flemi ger\u00e7ekle\u015ftirmektedir. “SendMessageAsync” method’u ise “topic” ve “message” parametrelerine sahiptir.\u00a0Kafka’ya giri\u015f makalesinden hat\u0131rlarsak\u00a0topic’ler kullan\u0131c\u0131 tan\u0131ml\u0131 category isimleri olup, yay\u0131nlanacak message’lar burada tutuluyordu. Bu nedenle “SendMessageAsync” method’u\u00a0i\u00e7erisinde, message’\u0131n hangi topic’e gidece\u011fini bildirmemiz gerekmektedir.<\/p>\n
Method i\u00e7eri\u011fine bakt\u0131\u011f\u0131m\u0131zda “_brokerRouter” \u0131 parametre olarak ge\u00e7ip yeni bir Producer instance’\u0131 olu\u015fturuyoruz. Bu instance \u00fczerinden ise “SendMessageAsync” method’unu \u00e7a\u011f\u0131r\u0131p, hangi topic’e gidece\u011fini ve ilgili message’\u0131 set edip, “Wait” method’u ile ilgili task’\u0131n execution’\u0131n\u0131n complete olmas\u0131n\u0131 bekliyoruz.<\/p>\n
“Program.cs” i ise a\u015fa\u011f\u0131daki gibi kodlayal\u0131m.<\/p>\n
using System;\r\nusing KafkaNet;\r\nusing KafkaNet.Model;\r\n\r\nnamespace KafkaProducerExample\r\n{\r\n class Program\r\n {\r\n static void Main(string[] args)\r\n {\r\n var kafkaOptions = new KafkaOptions(new Uri(\"http:\/\/192.168.99.100:9092\"));\r\n var brokerRouter = new BrokerRouter(kafkaOptions);\r\n var producer = new TestProducer(brokerRouter);\r\n\r\n Console.WriteLine(\"Send a Message to TestTopic:\");\r\n while (true)\r\n {\r\n producer.SendMessageAsync(\"TestTopic\", Console.ReadLine());\r\n }\r\n }\r\n }\r\n}<\/pre>\nBu k\u0131s\u0131m olduk\u00e7a straightforward. “KafkaOptions” class’\u0131na bir kafka server uri’\u0131 tan\u0131ml\u0131yoruz. Bu uri, kafka kurulumu s\u0131ras\u0131nda docker \u00fczerinden “9092” port adresine bind etti\u011fimiz docker ip’si. Sonras\u0131nda ise olu\u015fturmu\u015f\u00a0oldu\u011fumuz “KafkaOptions” instance’\u0131n\u0131 kullanarak, bir adet “BrokerRouter” instance’\u0131 yarat\u0131yoruz. “BrokerRouter” instance’\u0131n\u0131da kullanarak “TestProducer” \u0131 tan\u0131mlay\u0131p, “SendMessageAsync” method’u ile console \u00fczerinden gelen mesajlar\u0131 “TestTopic” e g\u00f6nderiyoruz.\u00a0Art\u0131k bir producer’e sahip oldu\u011fumuza g\u00f6re herhangi bir message g\u00f6nderildi\u011fi taktirde, bir adet topic ve bu topic’in x partition’\u0131n\u0131nda ve x offset’inde bir message’a sahip olaca\u011f\u0131z.\u00a0Art\u0131k bu topic’i\u00a0consume etmeye ba\u015flayabiliriz.<\/p>\n
Solution \u00fczerine “KafkaConsumerExample” isminde yeni bir console application daha ekleyelim ve producer k\u0131sm\u0131nda oldu\u011fu gibi Nuget Package Manager \u00fczerinden “kafka-net” k\u00fct\u00fcphanesini buraya da dahil edelim. Paket dahil etme i\u015fleminin hemen ard\u0131ndan “TestConsumer” isimli bir class ekleyelim ve a\u015fa\u011f\u0131daki gibi kodlamaya ba\u015flayal\u0131m.<\/p>\n
using System;\r\nusing System.Text;\r\nusing KafkaNet;\r\nusing KafkaNet.Model;\r\n\r\nnamespace KafkaConsumerExample\r\n{\r\n public class TestConsumer\r\n {\r\n private readonly IBrokerRouter _brokerRouter;\r\n\r\n public TestConsumer(IBrokerRouter brokerRouter)\r\n {\r\n _brokerRouter = brokerRouter;\r\n }\r\n\r\n public void StartConsume(string topic)\r\n {\r\n Console.WriteLine($\"Consuming {topic}\");\r\n\r\n var consumer = new Consumer(new ConsumerOptions(topic, _brokerRouter));\r\n\r\n foreach (var message in consumer.Consume())\r\n {\r\n Console.WriteLine(\"Response: PartitionId:{0}, Offset:{1} Message:{2}\",\r\n message.Meta.PartitionId, message.Meta.Offset, Encoding.UTF8.GetString(message.Value));\r\n }\r\n }\r\n }\r\n}<\/pre>\nBuradaki s\u00fcre\u00e7 de hemen hemen producer ile ayn\u0131. Consumer instance’\u0131 olu\u015ftururken, constructor \u00fczerinden extra olarak hangi topic’i consume etmesi gerekti\u011fini s\u00f6yl\u00fcyoruz. \u00a0Olu\u015fan consumer instance’\u0131 \u00fczerinden ise “Consume” method’unu \u00e7a\u011f\u0131rarak, “IEnumerable<Message>” tipinde bitmeyen bir stream elde ediyoruz. Yani s\u00fcrekli consume modunday\u0131z. Consume\u00a0edilen\u00a0message’\u0131n “Meta” property’si \u00fczerinden, hangi partition’da oldu\u011funu ve offset bilgisinin ne oldu\u011fu gibi spesifik bilgilerede\u00a0ula\u015fabilmek m\u00fcmk\u00fcnd\u00fcr.<\/p>\n
“Program.cs” k\u0131sm\u0131na ge\u00e7ti\u011fimizde ise consumer i\u00e7in:<\/p>\n
using System;\r\nusing KafkaNet;\r\nusing KafkaNet.Model;\r\n\r\nnamespace KafkaConsumerExample\r\n{\r\n class Program\r\n {\r\n static void Main(string[] args)\r\n {\r\n var kafkaOptions = new KafkaOptions(new Uri(\"http:\/\/192.168.99.100:9092\"));\r\n var brokerRouter = new BrokerRouter(kafkaOptions);\r\n var consumer = new TestConsumer(brokerRouter);\r\n\r\n consumer.StartConsume(\"TestTopic\");\r\n }\r\n }\r\n}<\/pre>\n“KafkaOptions” \u00fczerinden yine kafka server uri’\u0131n\u0131 verdikten sonra “TestConsumer” \u0131 initialize ediyoruz ve olu\u015fturmu\u015f oldu\u011fumuz “StartConsume” method’una, “TestTopic” ini consume etmek istedi\u011fimizi s\u00f6yl\u00fcyoruz. \u015eimdilik hepsi bu kadar.<\/p>\n
Dilerseniz iki console uygulamas\u0131n\u0131 da multiple startup olarak ba\u015flatal\u0131m ve producer \u00fczerinden bir ka\u00e7 mesaj g\u00f6nderelim.<\/p>\n
![\"docker-logo\"](\"\/wp-content\/uploads\/2016\/10\/docker-logo.png\"){kind=logo}
<\/a><\/p>\n![\"kafka-search\"](\"\/wp-content\/uploads\/2016\/10\/kafka-search.jpg\")
<\/a><\/p>\n![\"zookeeper-logo\"](\"\/wp-content\/uploads\/2016\/10\/zookeeper-logo.png\"){kind=logo}
<\/a><\/p>\n![\"docker-zookeeper-install\"](\"\/wp-content\/uploads\/2016\/10\/docker-zookeeper-install.jpg\")
<\/a><\/p>\n![\"docker-ps-zookeeper\"](\"\/wp-content\/uploads\/2016\/10\/docker-ps-zookeeper.jpg\")
<\/a><\/p>\n![\"apache_kafka\"](\"\/wp-content\/uploads\/2016\/10\/Apache_kafka.png\")
<\/a><\/p>\n![\"docker-kafka-install\"](\"\/wp-content\/uploads\/2016\/10\/docker-kafka-install.jpg\")
<\/a><\/p>\n![\"docker-ps\"](\"\/wp-content\/uploads\/2016\/10\/docker-ps.jpg\")
<\/a><\/p>\n![\"kafka-producer-net\"](\"\/wp-content\/uploads\/2016\/10\/kafka-producer-net.jpg\")
<\/a><\/p>\n![\"producer-consumer\"](\"\/wp-content\/uploads\/2016\/10\/producer-consumer.jpg\")
<\/a><\/p>\n