Розетки и резьба

Question

В настоящее время я извлекаю данные запаса из api, к которым у меня есть доступ. Я делаю это в следующих шагах:

1. петлю через список символов/запасов один на один
2. создать сокет-соединение и отправить соответствующее сообщение на апи
3. получить данные и отдельные его в строки до «! EndMSG!» принимаются в какой момент данных для этого символа завершения
4. преобразования данных (строка) в StringIO, а затем прочитать его в панде dataframe и в конечном итоге записать данные в SQL
5. ли следующий символ/акции

Соответствующий фрагмент кода:

def readlines(sock, recv_buffer=4096, delim='\n'): 
    buffer = '' 
    while True: 
     data = sock.recv(recv_buffer) 
     buffer += str(data.decode('latin-1')) 

    while buffer.find(delim) != -1: 
     line, buffer = buffer.split('\n', 1) 
     yield line 

def main(): 
    syms = ['MSFT', 'AAPL', 'GS', 'F'] 
    for sym in syms: 
     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
     sock.connect((host, port)) 

     data = '' 
     message = sym + #relevant api specific commands 
     sock.sendall(message.encode()) 

     for line in readlines(sock): 
      if "!ENDMSG!" in line: 
       break 
      data += line + '\n' 

     sock.close() 

     data = io.StringIO(data) 
     df = pd.read_csv(data) 
     df.to_sql(...) 

Я хотел бы включить резьб в это так, что я не должен делать один запас по времени. Однако то, что им не уверены в том, где/как реализовать замки, так что я не рискую получать данные о неправильных запасов в неправильных переменных и т.д.

Это то, что я до сих пор:

import threading 
from queue import Queue 

q = Queue() 
my_lock = threading.Lock() 

def readlines(sock, recv_buffer=4096, delim='\n'): 
    buffer = '' 
    while True: 
     data = sock.recv(recv_buffer) 
     buffer += str(data.decode('latin-1')) 

    while buffer.find(delim) != -1: 
     line, buffer = buffer.split('\n', 1) 
     yield line 

def get_symbol_data(sym, sock): 
    with my_lock: 
     data = '' 
     message = sym + #relevant api specific commands 
     sock.sendall(message.encode()) 

     for line in readlines(sock): 
      if "!ENDMSG!" in line: 
       break 
      data += line + '\n' 

     data = io.StringIO(data) 
     df = pd.read_csv(data) 
     df.to_sql(...) 

def threader(): 
    while True: 
     sym_tuple = q.get() 
     sym = sym_tuple[0] 
     sock = sym_tuple[1] 
     get_symbol_data(sym, sock) 
     q.task_done() 

def main(): 
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
    sock.connect((host, port)) 

    # create 4 threads 
    for x in range(4): 
     t = threading.Thread(target=threader) 
     t.daemon = True 
     t.start() 

    syms = ['MSFT', 'AAPL', 'GS', 'F'] 
    for sym in syms: 
     q.put((sym, sock)) 
    q.join() 
    sock.close() 

Моя попытка при включении резьбы просто висит. Нет ошибок, ничего. Он просто висит. Надеюсь, кто-то может указать мне в правильном направлении.

Я также не уверен, что если я воспользуюсь замком в нужном месте? Кстати, если я не использую блокировку, программа все еще висит. Предположительно, он должен работать, даже если данные все перепутаны из-за того, что не используются блокировки?

Answer 1

Вот мой 2 * [маленький единица валюты]:

• Что замок должен делать? Теперь каждый поток должен дождаться блокировки перед получением данных. Это не очень эффективно, так как сетевая операция - это, вероятно, то, что может принести наибольшую пользу от распараллеливания.
• Создайте сокет в каждом потоке. Таким образом, вам не нужно синхронизировать доступ к сокету и, возможно, полностью избавиться от блокировок. Кроме того, используйте пул сокетов.
• Я не уверен, как вы храните свои данные, но при обновлении кадра данных pandas может потребоваться синхронизация между авторами. Вы упоминаете SQL - надеюсь, ваша база данных позаботится об этом для вас. Другой вариант заключается в том, чтобы считыватели API/сокетов сообщали свои данные второму типу потока (или только основного потока), который собирает/записывает данные на ваше хранилище.

Все перечисленное предполагает, что сетевые операции являются причиной, по которой вы хотите распараллелить в первую очередь. Кто-то упомянул в комментарии, что вы можете повторно использовать сокет для всех символов. Я не знаю, как работает ваш API, но мне кажется, что для этого потребуется, чтобы все символы собирались поочередно.