Условие резьбы Получить блокировку на самом деле не получает блокировку - PullRequest
Новая
       5

Условие резьбы Получить блокировку на самом деле не получает блокировку

0 голосов
/ 26 сентября 2018

Я пытаюсь создать подписчика ZMQ, который пишет в CSV каждый определенный промежуток времени (время не так важно)

Моя проблема в том, что я в классе записи.Когда я запускаю c.acquire для получения блокировки, он ничего не делает, и цикл просто останавливается

C.acquire в моем классе ZMQ_Thread фактически работает, поэтому я не уверен, что делаю неправильно.

Любые указатели или советы были бы удивительными.Заранее спасибо. 

import zmq
import pandas as pd
import time
import threading


c = threading.Condition()
df = pd.DataFrame()
s = 0
m = 0
h = 0
d = 0

def counter():
    global h,s,m,d
    while True:
        s += 1
        print("Second:{}".format(s))
        if s >=60:
            m +=1
            s = 0
        if m >= 60:
            h += 1
            m = 0
        if h >= 24:
            d += 1
            h = 0        
        time.sleep(1)

class write(threading.Thread):
    def __init__(self,name):
        threading.Thread.__init__(self)
        self.name = name
    def run(self):
        global df
        while True:            
            print('sleeping')            
            time.sleep(12)
            c.acquire()
            print('acquired')
            print(df)                 
            with open("FILE", 'a') as f:
                    df.to_csv(f, encoding = 'utf-8', index = False, header = False)
            c.release()
            c.notify()
            z = zmq_thread()            
            z.run()        

class zmq_thread(threading.Thread):
    def __init__(self,name):
        threading.Thread.__init__(self)
        self.name = name
    def run(self):  
        global df
        print('DF Created')
        context = zmq.Context()
        socket = context.socket(zmq.SUB)      
        socket.connect("tcp://localhost:#####")
        socket.setsockopt_string(zmq.SUBSCRIBE, 'TOPIC') 
        print('socket connected')

        count = 0
        while True: 
            c.acquire()
            try:
                count +=1
                if count == 10:
                    print('ZMQ Break') 
                    c.notify()
                    c.release()
                    print('Lock released')
                    return df                
                message = socket.recv_string()
                message2 = socket.recv_string()        
                if message == 'TOPIC':
                    message2_split = message2.split(",")
                    message4 = pd.Series(message2_split)
                    df = df.append(message4, ignore_index=True)               
                    print('ZMQ Running')

            except KeyboardInterrupt:
                break              



counter = threading.Thread(target = counter)
write_csv1 = write("Write_csv")
zmq_loop = zmq_thread('Start_ZMQ')    

counter.start()
write_csv1.start()    
zmq_loop.start()

1 Ответ

0 голосов
/ 27 сентября 2018

Ответ на этот вопрос заключается в том, что я не вызывал c.wait ().это фрагмент кода из моего кода, который я добавил c.wait (), где был c.notify ().Он ожидает, когда следующий процесс получит блокировку, затем запускает класс write_csv. 

while True: 
            c.acquire()
            try:
                count +=1
                if count == 10:
                    print('ZMQ Break') 
                    c.wait()                                      
                message = socket.recv_string()
                message2 = socket.recv_string()        
                if message == 'TOPIC':
                    message2_split = message2.split(",")
                    message4 = pd.Series(message2_split)
                    df = df.append(message4, ignore_index=True)               
                    print('ZMQ Running')
...