Geekly articles weekly

Mulai Cepat: Buka + Apache Kafka + Redis

Baru-baru ini, karena kebutuhan, saya melihat melalui semua lowongan untuk pengembang Go, dan setengah dari mereka (setidaknya) menyebutkan platform pemrosesan pesan Apache Kafka dan database Redis NoSQL. Yah, semua orang, tentu saja, ingin kandidat mengenal Docker dan orang lain yang menyukainya. Semua persyaratan ini bagi kami, yang telah melihat pandangan para insinyur sistem, tampaknya entah bagaimana sepele atau apa. Sebenarnya, bagaimana perbedaan satu baris dengan yang lainnya? Situasi dengan database NoSQL, tentu saja, lebih beragam, tetapi masih tampak lebih sederhana daripada MS SQL Server. Semua ini, tentu saja, adalah pribadi saya, efek Dunning - Kruger , yang disebutkan berulang kali tentang Habré.
Jadi, karena semua pengusaha menuntut, perlu mempelajari teknologi ini. Tetapi mulai dengan membaca semua dokumentasi dari awal hingga akhir tidak terlalu menarik. Menurut pendapat saya, lebih produktif untuk membaca pendahuluan, membuat prototipe yang berfungsi, memperbaiki kesalahan, mengalami masalah, menyelesaikannya. Dan setelah semua ini, dengan pengertian, baca dokumentasi, atau bahkan buku yang terpisah.



Mereka yang tertarik dalam waktu singkat untuk berkenalan dengan kemampuan dasar dari produk ini, silakan baca terus.

Program pelatihan akan mempertimbangkan jumlah. Ini akan terdiri dari generator angka besar, prosesor angka, antrian, penyimpanan kolom dan server web.

Selama pengembangan, pola desain berikut akan diterapkan:


Arsitektur sistem akan terlihat seperti ini:



Dalam gambar, oval menunjukkan pola desain konveyor. Saya akan membahasnya lebih terinci.

Templat "conveyor" mengasumsikan bahwa informasi tersebut datang dalam bentuk aliran dan diproses secara bertahap. Biasanya ada beberapa generator (sumber informasi) dan satu atau lebih prosesor (pengolah informasi). Dalam hal ini, generator akan menjadi program on Go yang mengantri dalam jumlah besar secara acak. Dan prosesor (satu-satunya) akan menjadi program yang mengambil data dari antrian dan melakukan faktorisasi. Pada Go murni, pola ini cukup mudah diimplementasikan menggunakan saluran (chan). Di atas ada tautan ke github saya dengan contoh. Di sini, antrian pesan akan memainkan peran saluran.

Fan-In - Templat Fan-Out biasanya digunakan bersama dan, sebagaimana diterapkan pada Go, berarti paralelisasi perhitungan menggunakan goroutine, diikuti dengan merangkum hasil dan mentransfernya, misalnya, ke bawah pipa. Tautan ke contoh juga diberikan di atas. Sekali lagi, saluran diganti oleh antrian, goroutine tetap di tempatnya.

Sekarang beberapa kata tentang Apache Kafka. Kafka adalah sistem manajemen pesan yang memiliki alat pengelompokan yang sangat baik, menggunakan log transaksi (persis seperti dalam RDBMS) untuk menyimpan pesan, dan mendukung model antrian dan model penerbit / pelanggan. Yang terakhir dicapai melalui grup penerima pesan. Setiap pesan hanya menerima satu anggota grup (pemrosesan paralel), tetapi pesan tersebut akan dikirimkan satu kali ke setiap grup. Mungkin ada banyak kelompok seperti itu, serta penerima dalam setiap kelompok.

Untuk bekerja dengan Kafka saya akan menggunakan paket "github.com/segmentio/kafka-go".
Redis, di sisi lain, adalah basis data kolom nilai kunci dalam memori yang mendukung kemampuan untuk menyimpan data secara permanen. Tipe data utama untuk kunci dan nilai adalah string, tetapi ada beberapa yang lain. Redis dianggap sebagai salah satu basis data tercepat (atau sebagian besar) di kelasnya. Adalah baik untuk menyimpan segala macam statistik, metrik, aliran pesan, dll.
Untuk bekerja dengan Redis saya akan menggunakan paket "github.com/go-redis/redis".

Karena artikel ini merupakan awal yang cepat, kami akan menggunakan kedua sistem menggunakan Docker menggunakan gambar yang sudah jadi dari DockerHub. Saya menggunakan docker-compose di Windows 10 dalam mode container di Linux VM (secara otomatis dibuat oleh Docker VM) dengan file docker-compose.yml ini seperti ini:

version: '2' services: zookeeper: image: wurstmeister/zookeeper ports: - "2181:2181" kafka: image: wurstmeister/kafka:latest ports: - "9092:9092" environment: KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_CREATE_TOPICS: "Generated:1:1,Solved:1:1,Unsolved:1:1" KAFKA_DELETE_TOPIC_ENABLE: "true" volumes: - /var/run/docker.sock:/var/run/docker.sock redis: image: redis ports: - "6379:6379"

Simpan file ini, pergi ke direktori dengan itu dan jalankan:

 docker-compose up -d

Tiga kontainer harus diunduh dan mulai: Kafka (antrian), Zookeeper (server konfigurasi untuk Kafka) dan (Redis).

Anda dapat memverifikasi bahwa kontainer berfungsi menggunakan perintah:

 docker-compose ps

Seharusnya sesuatu seperti:

 Name State Ports -------------------------------------------------------------------------------------- docker-compose_kafka_1 Up 0.0.0.0:9092->9092/tcp docker-compose_redis_1 Up 0.0.0.0:6379->6379/tcp docker-compose_zookeeper_1 Up 0.0.0.0:2181->2181/tcp, 22/tcp, 2888/tcp, 3888/tcp

Menurut file-yml, tiga antrian harus dibuat secara otomatis, Anda dapat melihatnya dengan perintah:

 docker exec kafka-container_kafka_1 /opt/kafka_2.12-2.1.0/bin/kafka-topics.sh --list --zookeeper zookeeper:2181

Harus ada antrian (topik - topik dalam hal Kafka) Dihasilkan, Dipecahkan dan Belum Terpecahkan.

Generator data nomor antrian tanpa batas dengan penundaan acak. Kodenya sangat sederhana. Anda dapat memverifikasi keberadaan pesan dalam antrean Generated menggunakan perintah:

 docker exec kafka-container_kafka_1 /opt/kafka_2.12-2.1.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic Generated --from-beginning

Berikutnya adalah prosesor - di sini Anda harus memperhatikan untuk memparalelkan pemrosesan nilai dari antrian dalam blok kode berikut:

  var wg sync.WaitGroup c := 0 //counter for { //       15     ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() //      //    -     m, err := r.ReadMessage(ctx) if err != nil { fmt.Println("3") fmt.Println(err) break } wg.Add(1) //       10      goCtx, goCcancel := context.WithTimeout(context.Background(), 10*time.Millisecond) defer goCcancel() //     () go process(goCtx, c, &wg, m) c++ } //     wg.Wait()

Karena membaca dari antrian pesan memblokir program, saya membuat objek context.Context dengan batas waktu 15 detik. Batas waktu ini akan menghentikan program jika antrian kosong untuk waktu yang lama.

Juga, untuk setiap gorutin yang memfaktorkan angka, waktu operasi maksimum juga ditetapkan. Saya ingin angka-angka yang dapat menjadi faktor untuk ditulis dalam satu database. Dan angka-angka yang tidak dapat difaktorkan dalam waktu yang ditentukan dipindahkan ke basis data lain.

Untuk menentukan perkiraan waktu, patokan digunakan:

 func BenchmarkFactorize(b *testing.B) { ch := make(chan []int) var factors []int for i := 1; i < bN; i++ { num := 2345678901234 go factorize(num, ch) factors = <-ch b.Logf("\n%d   %+v\n\n", num, factors) } }

Benchmark in Go adalah varietas tes dan ditempatkan dalam file dengan tes. Berdasarkan pengukuran ini, nomor maksimum untuk generator angka acak dipilih. Di komputer saya, sebagian dari nomor punya waktu untuk faktor keluar, dan sebagian - tidak.

Angka-angka yang dapat diuraikan ditulis dalam DB No. 0, angka-angka yang tidak diuraikan dalam DB No. 1.
Di sini saya harus mengatakan bahwa dalam Redis tidak ada tabel dan tabel dalam arti klasik. Secara default, DBMS berisi 16 database yang tersedia untuk programmer. Basis-basis ini berbeda dalam jumlah mereka - dari 0 hingga 15.

Batas waktu untuk goroutine dalam prosesor disediakan dengan menggunakan konteks dan pernyataan pilih:

  //   go factorize(n, outChan) var item data select { case factors = <-outChan: { fmt.Printf("\ngoroutine #%d, input: %d, factors: %+v\n", counter, n, factors) item.Number = n item.Factors = factors err = storeSolved(item) if err != nil { fmt.Println("6") log.Fatal(err) } } case <-ctx.Done(): { fmt.Printf("\ngoroutine #%d, input: %d, exited on context timeout\n", counter, n) err = storeUnsolved(n) if err != nil { fmt.Println("7") log.Fatal(err) } return nil } }

Ini adalah tipuan pengembangan tipikal lainnya di Go. Artinya adalah bahwa pernyataan pilih berulang di saluran dan mengeksekusi kode yang sesuai dengan saluran aktif pertama. Dalam kasus ini, baik goroutine akan menampilkan hasilnya ke salurannya, atau saluran konteks dengan batas waktu akan ditutup. Alih-alih konteks, Anda dapat menggunakan saluran sewenang-wenang yang akan bertindak sebagai manajer dan memberikan penghentian paksa goroutine.

Subrutin untuk menulis ke database menjalankan perintah untuk memilih database yang diinginkan (0 atau 1) dan menulis pasangan bentuk (angka - faktor) untuk angka yang diuraikan atau (angka - angka) untuk angka yang tidak dikomposisi.

 func storeSolved(item data) (err error) { //    0 cmd := redis.NewStringCmd("select", 0) err = client.Process(cmd) b, err := json.Marshal(item.Factors) err = client.Set(strconv.Itoa(item.Number), string(b), 0).Err() return err }

Bagian terakhir akan menjadi server web , yang akan menampilkan daftar angka yang terdekomposisi dan undecomposed dalam bentuk json. Dia akan memiliki dua titik akhir:

  http.HandleFunc("/solved", solvedHandler) http.HandleFunc("/unsolved", unsolvedHandler)

Penangan permintaan http dengan menerima data dari Redis dan mengembalikannya sebagai json terlihat seperti ini:

 func solvedHandler(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.Header().Set("Access-Control-Allow-Origin", "*") w.Header().Set("Access-Control-Allow-Methods", "GET") w.Header().Set("Access-Control-Allow-Headers", "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization") //   №0 -   cmd := redis.NewStringCmd("select", 0) err := client.Process(cmd) if err != nil { w.WriteHeader(http.StatusInternalServerError) return } //      keys := client.Keys("*") var solved []data var item data //          for _, key := range keys.Val() { item.Key = key val, err := client.Get(key).Result() if err != nil { w.WriteHeader(http.StatusInternalServerError) return } item.Val = val solved = append(solved, item) } //    JSON err = json.NewEncoder(w).Encode(solved) if err != nil { w.WriteHeader(http.StatusInternalServerError) return } }

Hasil permintaan di: localhost / diselesaikan

 [{ "Key": "1604388558816", "Val": "[1,2,3,227]" }, { "Key": "545232916387", "Val": "[1,545232916387]" }, { "Key": "1786301239076", "Val": "[1,2]" }, { "Key": "698495534061", "Val": "[1,3,13,641,165331]" }]

Sekarang Anda dapat mempelajari dokumentasi dan literatur khusus. Semoga artikel ini bermanfaat.

Saya meminta para ahli untuk tidak terlalu malas dan menunjukkan kesalahan saya.

Source: https://habr.com/ru/post/id441250/

More articles:


All Articles