🥇Achtergrondtaken voor Faust, deel I: Inleiding

Hoe kwam ik ertoe zo te leven?

Nog niet zo lang geleden moest ik werken aan de backend van een zwaarbelast project, waarin het nodig was om de reguliere uitvoering van een groot aantal achtergrondtaken met complexe berekeningen en verzoeken om diensten van derden te organiseren. Het project is asynchroon en voordat ik kwam, had het een eenvoudig mechanisme voor het starten van cron-taken: een lus die de huidige tijd controleert en groepen coroutines start via collect - deze aanpak bleek acceptabel totdat er tientallen en honderden van dergelijke coroutines waren Toen hun aantal echter de tweeduizend overschreed, moest ik nadenken over het organiseren van een normale taakwachtrij met een makelaar, verschillende werknemers, enzovoort.

Eerst besloot ik Celery, die ik eerder had gebruikt, uit te proberen. Vanwege het asynchrone karakter van het project dook ik in de vraag en zag статьюEvenals project, gemaakt door de auteur van het artikel.

Ik zal dit zeggen, het project is erg interessant en werkt behoorlijk succesvol in andere toepassingen van ons team, en de auteur zelf zegt dat hij het in productie heeft kunnen brengen door een asynchrone pool te gebruiken. Maar helaas paste het niet echt bij mij, zo bleek проблема met groepslancering van taken (zie. groep). Op het moment van schrijven kwestie is al gesloten, maar er wordt al een maand aan gewerkt. In ieder geval veel succes voor de auteur en het allerbeste, aangezien er al werkende dingen op de lib staan... over het algemeen zit het punt in mij en bleek de tool vochtig voor mij. Bovendien hadden sommige taken 2-3 http-verzoeken naar verschillende services, dus zelfs bij het optimaliseren van taken creëren we 4 TCP-verbindingen, ongeveer elke 2 uur - niet erg goed... Ik zou graag een sessie willen maken voor één type taak bij het starten van werknemers. Iets meer over het grote aantal verzoeken via aiohttp hier.

In dit opzicht begon ik te zoeken alternatieven en gevonden! De makers van selderij, in het bijzonder, zoals ik het begrijp Vraag Solem, werd gecreëerd Faust, oorspronkelijk voor het project RobinHood. Faust is geïnspireerd door Kafka Streams en werkt samen met Kafka als makelaar, rocksdb wordt ook gebruikt om resultaten van het werk van agenten op te slaan, en het belangrijkste is dat de bibliotheek asynchroon is.

Je kunt ook kijken snelle vergelijking selderij en faust van de makers van de laatste: hun verschillen, verschillen tussen makelaars, implementatie van een elementaire taak. Alles is vrij eenvoudig, maar een leuke functie in Faust trekt de aandacht: getypte gegevens voor verzending naar het onderwerp.

Wat doen we?

Daarom laat ik je in een korte reeks artikelen zien hoe je met Faust gegevens van achtergrondtaken verzamelt. De bron voor ons voorbeeldproject zal, zoals de naam al doet vermoeden, zijn alphavantage.co. Ik zal demonstreren hoe je agents schrijft (sink, topics, partities), hoe je reguliere (cron) uitvoering uitvoert, de handigste faust cli-opdrachten (een wrapper over click), eenvoudige clustering, en aan het einde zullen we een datadog ( out of the box werken) en probeer iets te zien. Om de verzamelde gegevens op te slaan, gebruiken we mongodb en motor voor verbinding.

PS Afgaande op het vertrouwen waarmee het punt over monitoring is geschreven, denk ik dat de lezer aan het einde van het laatste artikel er nog steeds ongeveer zo uit zal zien:

Projectvereisten

Omdat ik het al heb beloofd, laten we een kleine lijst maken van wat de service zou moeten kunnen doen:

Upload regelmatig effecten en een overzicht ervan (inclusief winst en verlies, balans, cashflow - van het afgelopen jaar).
Upload historische gegevens (vind voor elk handelsjaar extreme waarden van de slotkoers van de handel) - regelmatig
Upload regelmatig de nieuwste handelsgegevens
Upload regelmatig een aangepaste lijst met indicatoren voor elk effect

Zoals verwacht kiezen we helemaal opnieuw een naam voor het project: Horton

We bereiden de infrastructuur voor

De titel is zeker sterk, maar het enige dat je hoeft te doen is een kleine configuratie schrijven voor docker-compose met kafka (en zookeeper - in één container), kafdrop (als we naar berichten in onderwerpen willen kijken), mongodb. We krijgen [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) van de volgende vorm:

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

Er is hier helemaal niets ingewikkelds. Er zijn twee luisteraars gedeclareerd voor kafka: één (intern) voor gebruik binnen het samengestelde netwerk en de tweede (extern) voor verzoeken van buitenaf, dus hebben ze deze naar buiten doorgestuurd. 2181 — poort voor dierenverzorgers. De rest is volgens mij duidelijk.

Het skelet van het project voorbereiden

In de basisversie zou de structuur van ons project er als volgt uit moeten zien:

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 *

*Alles wat ik heb opgemerkt We raken er nog niet aan, we creëren alleen lege bestanden.**

We hebben een structuur gecreëerd. Laten we nu de nodige afhankelijkheden toevoegen, de configuratie schrijven en verbinding maken met mongodb. Om het niet te vertragen, zal ik niet de volledige tekst van de bestanden in het artikel verstrekken, maar wel links naar de benodigde versies.

Laten we beginnen met afhankelijkheden en meta over het project - pyproject.toml

Vervolgens beginnen we met het installeren van afhankelijkheden en het maken van een virtualenv (of u kunt zelf de venv-map maken en de omgeving activeren):

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

Laten we nu creëren config.yml - Inloggegevens en waar u kunt aankloppen. Daar kunt u direct gegevens voor alphavantage plaatsen. Nou, laten we verder gaan config.py - extraheer gegevens voor de applicatie uit onze configuratie. Ja, ik moet bekennen, ik heb mijn lib gebruikt - SITRI.

Bij het verbinden met Mongo is alles vrij eenvoudig. aangekondigd cliënt klasse verbinden en basisklasse voor cruds, om het gemakkelijker te maken om vragen over collecties te stellen.

Wat gebeurt er daarna?

Het artikel is niet erg lang, aangezien ik het hier alleen heb over motivatie en voorbereiding, dus geef mij niet de schuld - ik beloof dat het volgende deel actie en graphics zal bevatten.

Dus in dit volgende deel:

Laten we een kleine client voor alphavantage op aiohttp schrijven met verzoeken voor de eindpunten die we nodig hebben.
Laten we een agent creëren die gegevens over effecten en historische prijzen daarvoor verzamelt.

Projectcode

Codeer voor dit onderdeel

Bron: www.habr.com

Achtergrondtaken over Faust, deel I: inleiding