线程 class 将不会启动或初始化
Thread class will not start or Initialize
我正在尝试创建一个 ZMQ 订阅者,每隔一定时间(时间并不重要)写入 CSV。
我遇到的问题是我的 write_csv1 class 甚至没有启动或初始化,初始化代码与我的 ZMQ 线程 class 相同。在我添加 threading.Condition 和 c.acquires/c.release.
之前它正在工作
任何提示或指示都会很棒。
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 write_csv(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()
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_all()
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()
问题是您没有在写入 class 时实施 run()
方法。
您需要将 write_csv(self)
重命名为 run(self)
我正在尝试创建一个 ZMQ 订阅者,每隔一定时间(时间并不重要)写入 CSV。
我遇到的问题是我的 write_csv1 class 甚至没有启动或初始化,初始化代码与我的 ZMQ 线程 class 相同。在我添加 threading.Condition 和 c.acquires/c.release.
之前它正在工作任何提示或指示都会很棒。
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 write_csv(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()
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_all()
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()
问题是您没有在写入 class 时实施 run()
方法。
您需要将 write_csv(self)
重命名为 run(self)