Latar mburi Tugas ing Faust, Part I: Pambuka

Kepiye carane aku bisa urip kaya ngono?

Ora suwe aku kudu nggarap backend proyek sing akeh dimuat, sing perlu kanggo ngatur eksekusi reguler saka akeh tugas latar mburi kanthi petungan rumit lan panjaluk layanan pihak katelu. Proyek kasebut ora sinkron lan sadurunge aku teka, ana mekanisme sing gampang kanggo tugas cron-launching: daur ulang mriksa wektu saiki lan ngluncurake grup coroutine liwat kumpul - pendekatan iki bisa ditampa nganti ana puluhan lan atusan coroutine kasebut. Nanging, nalika jumlahe ngluwihi rong ewu, aku kudu mikir babagan ngatur antrian tugas normal karo makelar, sawetara buruh, lan liya-liyane.

Pisanan aku mutusake kanggo nyoba Celery, sing wis digunakake sadurunge. Amarga sifat asynchronous saka project, Aku nyilem menyang pitakonan lan weruh artikel, ugi konsep, digawe dening penulis artikel.

Aku bakal ngomong iki, proyek kasebut menarik banget lan bisa digunakake kanthi sukses ing aplikasi liyane saka tim kita, lan penulis dhewe ujar manawa dheweke bisa nggawe produksi kanthi nggunakake blumbang asynchronous. Nanging, sayangé, ora cocog karo aku, kaya sing nyata masalahipun kanthi ngluncurake tugas klompok (pirsani. group). Ing wektu nulis Jeksa Agung bisa ngetokake wis ditutup, Nanging, karya wis arep kanggo sasi. Ing kasus apa wae, good luck kanggo penulis lan kabeh sing paling apik, amarga wis ana sing nggarap lib ... umume, titik kasebut ana ing aku lan alat kasebut dadi lembab kanggo aku. Kajaba iku, sawetara tugas duwe 2-3 panjalukan http kanggo layanan sing beda-beda, dadi sanajan ngoptimalake tugas, kita nggawe 4 ewu sambungan TCP, kira-kira saben 2 jam - ora apik banget ... Aku pengin nggawe sesi kanggo siji jinis tugas nalika miwiti buruh. A sethitik liyane babagan nomer akeh panjalukan liwat aiohttp kene.

Ing babagan iki, aku wiwit nggoleki alternatif lan ketemu! Pencipta celery, khusus, kaya sing dakmangerteni Takon Solem, digawe Faust, asline kanggo proyek kasebut robinhood. Faust diilhami dening Kafka Streams lan nggarap Kafka minangka broker, rocksdb uga digunakake kanggo nyimpen asil saka karya agen, lan sing paling penting yaiku perpustakaan ora sinkron.

Uga, sampeyan bisa ndeleng comparison singkat celery lan faust saka pangripta pungkasan: beda sing, beda antarane makelar, implementasine saka tugas dhasar. Kabeh iku cukup prasaja, Nanging, fitur becik ing faust nengsemake manungsa waé - diketik data kanggo transmisi kanggo topik.

Apa sing kita lakoni?

Dadi, ing seri artikel cekak, aku bakal nuduhake sampeyan carane ngumpulake data saka tugas latar mburi nggunakake Faust. Sumber kanggo proyek conto kita bakal, kaya jeneng kasebut, alphavantage.co. Aku bakal nduduhake carane nulis agen (sink, topik, partisi), carane nindakake eksekusi biasa (cron), perintah cli faust sing paling trep (pambungkus liwat klik), clustering prasaja, lan ing pungkasan kita bakal masang datadog ( apa metu saka kothak) lan nyoba kanggo soko kanggo ndeleng. Kanggo nyimpen data sing diklumpukake kita bakal nggunakake mongodb lan motor kanggo sambungan.

PS Dideleng saka kapercayan sing ditulis babagan pemantauan, aku mikir yen maca ing pungkasan artikel pungkasan isih bakal katon kaya iki:

Persyaratan proyek

Amarga kasunyatane aku wis janji, ayo nggawe dhaptar cilik babagan apa sing kudu ditindakake layanan kasebut:

1. Unggah sekuritas lan ringkesan kasebut (kalebu bathi lan kerugian, neraca, arus kas - kanggo taun kepungkur) - kanthi rutin
2. Unggah data historis (kanggo saben taun dagang, temokake nilai ekstrem saka rega penutupan dagang) - kanthi rutin
3. Unggah data dagang paling anyar - kanthi rutin
4. Unggah dhaptar indikator khusus kanggo saben keamanan - kanthi rutin

Kaya sing dikarepake, kita milih jeneng kanggo proyek kasebut saka awal: Horton

Kita lagi nyiapake infrastruktur

Judhul mesthi kuwat, Nanging, sampeyan mung kudu nulis konfigurasi cilik kanggo docker-compose karo kafka (lan zookeeper - ing siji wadhah), kafdrop (yen kita pengin ndeleng pesen ing topik), mongodb. We njaluk [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) saka wangun ing ngisor iki:

version: '3'

services:
  db:
    container_name: horton-mongodb-local
    image: mongo:4.2-bionic
    command: mongod --port 20017
    restart: always
    ports:
      - 20017:20017
    environment:
      - MONGO_INITDB_DATABASE=horton
      - MONGO_INITDB_ROOT_USERNAME=admin
      - MONGO_INITDB_ROOT_PASSWORD=admin_password

  kafka-service:
    container_name: horton-kafka-local
    image: obsidiandynamics/kafka
    restart: always
    ports:
      - "2181:2181"
      - "9092:9092"
    environment:
      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"
      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-service:29092,EXTERNAL://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"
      KAFKA_RESTART_ATTEMPTS: "10"
      KAFKA_RESTART_DELAY: "5"
      ZOOKEEPER_AUTOPURGE_PURGE_INTERVAL: "0"

  kafdrop:
    container_name: horton-kafdrop-local
    image: 'obsidiandynamics/kafdrop:latest'
    restart: always
    ports:
      - '9000:9000'
    environment:
      KAFKA_BROKERCONNECT: kafka-service:29092
    depends_on:
      - kafka-service

Ora ana sing rumit ing kene. Rong pamireng diumumake kanggo kafka: siji (internal) kanggo digunakake ing jaringan komposit, lan sing kapindho (eksternal) kanggo panjalukan saka njaba, mula diterusake ing njaba. 2181 - pelabuhan zookeeper. Liyane, aku, wis cetha.

Nyiapake kerangka proyek

Ing versi dhasar, struktur proyek kita kudu katon kaya iki:

horton
├── docker-compose.yml
└── horton
    ├── agents.py *
    ├── alphavantage.py *
    ├── app.py *
    ├── config.py
    ├── database
    │   ├── connect.py
    │   ├── cruds
    │   │   ├── base.py
    │   │   ├── __init__.py
    │   │   └── security.py *
    │   └── __init__.py
    ├── __init__.py
    ├── records.py *
    └── tasks.py *

*Kabeh aku nyathet Durung ndemek, mung nggawe file kosong.**

Kita nggawe struktur. Saiki ayo nambah dependensi sing dibutuhake, tulis konfigurasi lan sambung menyang mongodb. Aku ora bakal nyedhiyakake teks lengkap file ing artikel kasebut, supaya ora ditundha, nanging aku bakal menehi tautan menyang versi sing dibutuhake.

Ayo miwiti karo dependensi lan meta babagan proyek - pyproject.toml

Sabanjure, kita miwiti nginstal dependensi lan nggawe virtualenv (utawa sampeyan bisa nggawe folder venv dhewe lan ngaktifake lingkungan):

pip3 install poetry (если ещё не установлено)
poetry install

Saiki ayo nggawe config.yml - Kredensial lan ngendi kanggo nuthuk. Sampeyan bisa langsung nyelehake data kanggo alphavantage ing kana. Inggih, ayo pindhah menyang config.py - extract data kanggo aplikasi saka konfigurasi kita. Ya, aku ngaku, aku nggunakake libku - Siti.

Nalika nyambung menyang Mongo, kabeh cukup prasaja. ngumumake kelas klien kanggo nyambung lan kelas dhasar kanggo cruds, kanggo wis luwih gampang kanggo nggawe pitakon ing koleksi.

Apa sing bakal kelakon mengko?

Artikel kasebut ora dawa banget, amarga ing kene aku mung ngomong babagan motivasi lan persiapan, mula aja nyalahake aku - aku janji yen bagean sabanjure bakal duwe aksi lan grafis.

Dadi, ing bagean sabanjure kita:

1. Ayo nulis klien cilik kanggo alphavantage ing aiohttp kanthi panjaluk kanggo titik pungkasan sing dibutuhake.
2. Ayo nggawe agen sing bakal ngumpulake data sekuritase lan rega historis kanggo dheweke.

Kode proyek

Kode kanggo bagean iki

Source: www.habr.com