ວຽກງານຄວາມເປັນມາກ່ຽວກັບ Faust, ພາກທີ I: ແນະນໍາ

ຂ້ອຍມາຢູ່ແບບນີ້ໄດ້ແນວໃດ?

ບໍ່ດົນກ່ອນຫນ້ານີ້ຂ້ອຍຕ້ອງເຮັດວຽກຢູ່ໃນ backend ຂອງໂຄງການທີ່ມີການໂຫຼດສູງ, ໃນນັ້ນມັນຈໍາເປັນຕ້ອງຈັດລະບຽບການປະຕິບັດປົກກະຕິຂອງວຽກງານພື້ນຫລັງຈໍານວນຫລາຍທີ່ມີການຄິດໄລ່ສະລັບສັບຊ້ອນແລະການຮ້ອງຂໍສໍາລັບການບໍລິການຂອງພາກສ່ວນທີສາມ. ໂຄງການແມ່ນ asynchronous ແລະກ່ອນທີ່ຂ້ອຍຈະມາ, ມັນມີກົນໄກງ່າຍໆສໍາລັບວຽກງານການເປີດຕົວ cron: loop ກວດເບິ່ງເວລາປະຈຸບັນແລະການເປີດຕົວກຸ່ມ coroutines ໂດຍຜ່ານການລວບລວມ - ວິທີການນີ້ໄດ້ກາຍເປັນທີ່ຍອມຮັບຈົນກ່ວາມີຫລາຍສິບແລະຫຼາຍຮ້ອຍ coroutines ດັ່ງກ່າວ. ຢ່າງໃດກໍຕາມ, ເມື່ອຈໍານວນຂອງພວກເຂົາເກີນສອງພັນຄົນ, ຂ້ອຍຕ້ອງຄິດກ່ຽວກັບການຈັດແຖວວຽກປົກກະຕິກັບນາຍຫນ້າ, ຄົນງານຫຼາຍຄົນ, ແລະອື່ນໆ.

ທໍາອິດຂ້ອຍໄດ້ຕັດສິນໃຈທົດລອງໃຊ້ Celery, ເຊິ່ງຂ້ອຍໄດ້ໃຊ້ກ່ອນ. ເນື່ອງຈາກລັກສະນະທີ່ບໍ່ສອດຄ່ອງກັນຂອງໂຄງການ, ຂ້າພະເຈົ້າໄດ້ເຂົ້າໄປໃນຄໍາຖາມແລະໄດ້ເຫັນ ບົດຄວາມ, ເຊັ່ນດຽວກັນກັບ ໂຄງການ, ສ້າງໂດຍຜູ້ຂຽນຂອງບົດຄວາມ.

ຂ້າພະເຈົ້າຈະເວົ້ານີ້, ໂຄງການແມ່ນຫນ້າສົນໃຈຫຼາຍແລະເຮັດວຽກຂ້ອນຂ້າງສົບຜົນສໍາເລັດໃນຄໍາຮ້ອງສະຫມັກອື່ນໆຂອງທີມງານຂອງພວກເຮົາ, ແລະຜູ້ຂຽນເອງເວົ້າວ່າລາວສາມາດມ້ວນມັນອອກໃນການຜະລິດໂດຍໃຊ້ສະນຸກເກີ asynchronous. ແຕ່ຫນ້າເສຍດາຍ, ມັນບໍ່ເຫມາະສົມກັບຂ້ອຍແທ້ໆ, ຍ້ອນວ່າມັນອອກມາ ບັນຫາ ດ້ວຍການເປີດຕົວກຸ່ມຂອງວຽກງານ (ເບິ່ງ. ກຸ່ມ). ໃນເວລາຂຽນ ບັນຫາ ໄດ້ປິດແລ້ວ, ຢ່າງໃດກໍຕາມ, ວຽກງານໄດ້ດໍາເນີນໄປເປັນເວລາຫນຶ່ງເດືອນ. ໃນກໍລະນີໃດກໍ່ຕາມ, ໂຊກດີກັບຜູ້ຂຽນແລະທັງຫມົດທີ່ດີທີ່ສຸດ, ເນື່ອງຈາກວ່າມີແລ້ວສິ່ງທີ່ເຮັດວຽກຢູ່ໃນ lib ... ໂດຍທົ່ວໄປ, ຈຸດແມ່ນຢູ່ໃນຂ້າພະເຈົ້າແລະເຄື່ອງມືໄດ້ຫັນອອກເປັນຄວາມຊຸ່ມຊື່ນສໍາລັບຂ້າພະເຈົ້າ. ນອກຈາກນັ້ນ, ວຽກງານບາງຢ່າງມີ 2-3 http ຮ້ອງຂໍການບໍລິການທີ່ແຕກຕ່າງກັນ, ດັ່ງນັ້ນເຖິງແມ່ນວ່າໃນເວລາທີ່ optimizing ວຽກງານ, ພວກເຮົາສ້າງ 4 ພັນ TCP ເຊື່ອມຕໍ່, ປະມານທຸກໆ 2 ຊົ່ວໂມງ - ບໍ່ດີຫຼາຍ ... ຂ້າພະເຈົ້າຢາກຈະສ້າງ session ສໍາລັບປະເພດຫນຶ່ງ. ວຽກ​ງານ​ໃນ​ເວ​ລາ​ທີ່​ເລີ່ມ​ຕົ້ນ​ຄົນ​ງານ​. ເພີ່ມເຕີມເລັກນ້ອຍກ່ຽວກັບການຮ້ອງຂໍຈໍານວນຫລາຍຜ່ານ aiohttp ທີ່ນີ້.

ໃນເລື່ອງນີ້, ຂ້າພະເຈົ້າໄດ້ເລີ່ມຕົ້ນຊອກຫາ ທາງເລືອກ ແລະພົບມັນ! ຜູ້ສ້າງຂອງ celery, ໂດຍສະເພາະ, ດັ່ງທີ່ຂ້າພະເຈົ້າເຂົ້າໃຈມັນ ຖາມ Solem, ຖືກສ້າງຂື້ນ Faust, ໃນເບື້ອງຕົ້ນສໍາລັບໂຄງການ ຄວາມກ້າຫານ. Faust ໄດ້ຮັບການດົນໃຈໂດຍ Kafka Streams ແລະເຮັດວຽກກັບ Kafka ເປັນນາຍຫນ້າ, rocksdb ຍັງຖືກນໍາໃຊ້ເພື່ອເກັບຜົນໄດ້ຮັບຈາກການເຮັດວຽກຂອງຕົວແທນ, ແລະສິ່ງທີ່ສໍາຄັນທີ່ສຸດແມ່ນວ່າຫ້ອງສະຫມຸດແມ່ນ asynchronous.

ນອກຈາກນີ້, ທ່ານສາມາດເບິ່ງ ການ​ປຽບ​ທຽບ​ໄວ​ celery ແລະ faust ຈາກຜູ້ສ້າງຂອງຍຸກສຸດທ້າຍ: ຄວາມແຕກຕ່າງຂອງພວກເຂົາ, ຄວາມແຕກຕ່າງລະຫວ່າງນາຍຫນ້າ, ການປະຕິບັດວຽກງານປະຖົມ. ທຸກສິ່ງທຸກຢ່າງແມ່ນຂ້ອນຂ້າງງ່າຍດາຍ, ຢ່າງໃດກໍຕາມ, ຄຸນນະສົມບັດງາມໃນ faust ດຶງດູດຄວາມສົນໃຈ - typed data for transmission to the topic .

ພວກເຮົາເຮັດແນວໃດ?

ດັ່ງນັ້ນ, ໃນຊຸດສັ້ນຂອງບົດຄວາມ, ຂ້ອຍຈະສະແດງວິທີການເກັບກໍາຂໍ້ມູນຈາກຫນ້າວຽກພື້ນຖານໂດຍໃຊ້ Faust. ແຫຼ່ງຂໍ້ມູນສໍາລັບໂຄງການຕົວຢ່າງຂອງພວກເຮົາຈະເປັນ, ດັ່ງທີ່ຊື່ແນະນໍາ, alphavantage.co. ຂ້ອຍຈະສະແດງວິທີການຂຽນຕົວແທນ (sink, ຫົວຂໍ້, partitions), ວິທີການດໍາເນີນການປົກກະຕິ (cron), ຄໍາສັ່ງ faust cli ທີ່ສະດວກທີ່ສຸດ (a wrapper over click), clustering ງ່າຍດາຍ, ແລະໃນຕອນທ້າຍພວກເຮົາຈະຕິດ datadog (. ເຮັດວຽກອອກຈາກກ່ອງ) ແລະພະຍາຍາມບາງສິ່ງບາງຢ່າງເພື່ອເບິ່ງ. ເພື່ອເກັບຮັກສາຂໍ້ມູນທີ່ເກັບກໍາ, ພວກເຮົາຈະນໍາໃຊ້ mongodb ແລະ motor ສໍາລັບການເຊື່ອມຕໍ່.

PS ການຕັດສິນໂດຍຄວາມຫມັ້ນໃຈທີ່ຈຸດກ່ຽວກັບການຕິດຕາມກວດກາໄດ້ຖືກຂຽນ, ຂ້າພະເຈົ້າຄິດວ່າຜູ້ອ່ານໃນຕອນທ້າຍຂອງບົດຄວາມທີ່ຜ່ານມາຍັງຈະເບິ່ງບາງສິ່ງບາງຢ່າງເຊັ່ນນີ້:

ຄວາມຕ້ອງການໂຄງການ

ເນື່ອງຈາກຄວາມຈິງທີ່ວ່າຂ້ອຍໄດ້ສັນຍາໄວ້ແລ້ວ, ໃຫ້ພວກເຮົາເຮັດບັນຊີລາຍຊື່ນ້ອຍໆຂອງສິ່ງທີ່ບໍລິການຄວນຈະສາມາດເຮັດໄດ້:

1. ອັບໂຫຼດຫຼັກຊັບແລະພາບລວມຂອງພວກເຂົາ (ລວມທັງຜົນກໍາໄລແລະການສູນເສຍ, ໃບດຸ່ນດ່ຽງ, ກະແສເງິນສົດ - ສໍາລັບປີທີ່ຜ່ານມາ) - ເປັນປະຈໍາ
2. ອັບ​ໂຫຼດ​ຂໍ້​ມູນ​ປະ​ຫວັດ​ສາດ (ສໍາ​ລັບ​ການ​ຊື້​ຂາຍ​ໃນ​ແຕ່​ລະ​ປີ​, ຊອກ​ຫາ​ມູນ​ຄ່າ​ທີ່​ຮ້າຍ​ແຮງ​ຂອງ​ລາ​ຄາ​ປິດ​ຂອງ​ການ​ຊື້​ຂາຍ​) - ເປັນ​ປົກ​ກະ​ຕິ
3. ອັບໂຫຼດຂໍ້ມູນການຊື້ຂາຍຫຼ້າສຸດ - ເປັນປະຈໍາ
4. ອັບ​ໂຫຼດ​ລາຍ​ຊື່​ຕົວ​ຊີ້​ວັດ​ທີ່​ປັບ​ແຕ່ງ​ສໍາ​ລັບ​ແຕ່​ລະ​ຄວາມ​ປອດ​ໄພ - ເປັນ​ປົກ​ກະ​ຕິ​

ດັ່ງທີ່ຄາດໄວ້, ພວກເຮົາເລືອກຊື່ສໍາລັບໂຄງການຕັ້ງແຕ່ເລີ່ມຕົ້ນ: ສວນສາທາລະນະ

ພວກເຮົາກໍາລັງກະກຽມໂຄງສ້າງພື້ນຖານ

ຫົວຂໍ້ແມ່ນແນ່ນອນທີ່ເຂັ້ມແຂງ, ຢ່າງໃດກໍຕາມ, ທັງຫມົດທີ່ທ່ານຈໍາເປັນຕ້ອງເຮັດແມ່ນຂຽນ config ຂະຫນາດນ້ອຍສໍາລັບ docker-compose ກັບ kafka (ແລະ zookeeper - ໃນຫນຶ່ງ container), kafdrop (ຖ້າພວກເຮົາຕ້ອງການເບິ່ງຂໍ້ຄວາມໃນຫົວຂໍ້), mongodb. ພວກເຮົາໄດ້ຮັບ [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) ຂອງ​ຮູບ​ແບບ​ຕໍ່​ໄປ​ນີ້​:

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

ບໍ່ມີຫຍັງສັບສົນຢູ່ທີ່ນີ້. ສອງຜູ້ຟັງໄດ້ຖືກປະກາດສໍາລັບ kafka: ຫນຶ່ງ (ພາຍໃນ) ສໍາລັບການນໍາໃຊ້ພາຍໃນເຄືອຂ່າຍປະສົມປະສານ, ແລະທີສອງ (ພາຍນອກ) ສໍາລັບການຮ້ອງຂໍຈາກພາຍນອກ, ດັ່ງນັ້ນເຂົາເຈົ້າໄດ້ສົ່ງຕໍ່ພາຍນອກ. 2181 — ທີ່ Port zookeeper. ສ່ວນທີ່ເຫຼືອ, ຂ້າພະເຈົ້າຄິດວ່າ, ແມ່ນຈະແຈ້ງ.

ການກະກຽມໂຄງກະດູກຂອງໂຄງການ

ໃນ​ສະ​ບັບ​ພື້ນ​ຖານ​, ໂຄງ​ປະ​ກອບ​ການ​ຂອງ​ໂຄງ​ການ​ຂອງ​ພວກ​ເຮົາ​ຄວນ​ຈະ​ເປັນ​ດັ່ງ​ນີ້​:

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 *

*ທຸກຢ່າງທີ່ຂ້ອຍສັງເກດເຫັນ ພວກເຮົາບໍ່ໄດ້ແຕະມັນເທື່ອ, ພວກເຮົາພຽງແຕ່ສ້າງໄຟລ໌ເປົ່າ.**

ພວກເຮົາສ້າງໂຄງສ້າງ. ຕອນນີ້ໃຫ້ເພີ່ມຄວາມເພິ່ງພາອາໄສທີ່ຈໍາເປັນ, ຂຽນ config ແລະເຊື່ອມຕໍ່ກັບ mongodb. ຂ້ອຍຈະບໍ່ສະຫນອງຂໍ້ຄວາມເຕັມຂອງໄຟລ໌ໃນບົດຄວາມ, ເພື່ອບໍ່ໃຫ້ມັນຊັກຊ້າ, ແຕ່ຂ້ອຍຈະສະຫນອງການເຊື່ອມຕໍ່ກັບສະບັບທີ່ຈໍາເປັນ.

ໃຫ້ເລີ່ມຕົ້ນດ້ວຍການເພິ່ງພາອາໄສແລະ meta ກ່ຽວກັບໂຄງການ - pyproject.toml

ຕໍ່ໄປ, ພວກເຮົາເລີ່ມຕົ້ນການຕິດຕັ້ງ dependencies ແລະສ້າງ virtualenv (ຫຼືທ່ານສາມາດສ້າງໂຟນເດີ venv ຕົວທ່ານເອງແລະກະຕຸ້ນສະພາບແວດລ້ອມ):

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

ໃນປັດຈຸບັນໃຫ້ພວກເຮົາສ້າງ config.yml - ຂໍ້ມູນປະຈໍາຕົວແລະບ່ອນທີ່ຈະລົບ. ທ່ານທັນທີສາມາດວາງຂໍ້ມູນສໍາລັບ alphavantage ຢູ່ທີ່ນັ້ນ. ແລ້ວ, ໃຫ້ເຮົາກ້າວຕໍ່ໄປ config.py — ເອົາ​ຂໍ້​ມູນ​ສໍາ​ລັບ​ຄໍາ​ຮ້ອງ​ສະ​ຫມັກ​ຈາກ config ຂອງ​ພວກ​ເຮົາ​. ແມ່ນແລ້ວ, ຂ້ອຍສາລະພາບ, ຂ້ອຍໃຊ້ lib ຂອງຂ້ອຍ - sitri.

ເມື່ອເຊື່ອມຕໍ່ກັບ Mongo, ທຸກສິ່ງທຸກຢ່າງແມ່ນຂ້ອນຂ້າງງ່າຍດາຍ. ປະ​ກາດ ຫ້ອງຮຽນລູກຄ້າ ການ​ເຊື່ອມ​ຕໍ່​ແລະ​ ຫ້ອງຮຽນພື້ນຖານ ສໍາລັບ cruds, ເພື່ອເຮັດໃຫ້ມັນງ່າຍຂຶ້ນເພື່ອເຮັດໃຫ້ການສອບຖາມກ່ຽວກັບການລວບລວມ.

ຈະມີຫຍັງເກີດຂື້ນຕໍ່ໄປ?

ບົດຄວາມບໍ່ຍາວຫຼາຍ, ເພາະວ່າໃນທີ່ນີ້ຂ້າພະເຈົ້າພຽງແຕ່ເວົ້າກ່ຽວກັບແຮງຈູງໃຈແລະການກະກຽມ, ສະນັ້ນຢ່າຕໍານິຕິຕຽນຂ້ອຍ - ຂ້ອຍສັນຍາວ່າພາກຕໍ່ໄປຈະມີການປະຕິບັດແລະຮູບພາບ.

ດັ່ງນັ້ນ, ໃນພາກຕໍ່ໄປນີ້ພວກເຮົາ:

1. ໃຫ້ຂຽນລູກຄ້າຂະຫນາດນ້ອຍສໍາລັບ alphavantage ໃນ aiohttp ດ້ວຍການຮ້ອງຂໍສໍາລັບຈຸດສິ້ນສຸດທີ່ພວກເຮົາຕ້ອງການ.
2. ໃຫ້ສ້າງຕົວແທນຜູ້ທີ່ຈະເກັບກໍາຂໍ້ມູນກ່ຽວກັບຫຼັກຊັບແລະລາຄາປະຫວັດສາດສໍາລັບພວກເຂົາ.

ລະຫັດໂຄງການ

ລະຫັດສໍາລັບສ່ວນນີ້

ແຫຼ່ງຂໍ້ມູນ: www.habr.com