提问人:Rajat Bhardwaj 提问时间:8/26/2023 更新时间:8/26/2023 访问量:39
使用 python 多处理实现长轮询,但不起作用
Implementing long polling with python multiprocessing but doesn't work
问:
这是我正在做的:
def _wait_for_long_poll_resp(param_one, q: Queue):
print("long poll called", flush=True)
sys.stdout.flush()
request_url = <some_url>
auth = <auth>
while True:
print(f'I am Calling {request_url}', flush=True)
sys.stdout.flush()
resp = requests.get(request_url, headers=HEADERS, auth=auth)
print(resp.status_code, flush=True)
sys.stdout.flush()
print(resp.content, flush=True)
sys.stdout.flush()
if resp.status_code == 200:
print("Response code is 200", flush=True)
print(resp.content, flush=True)
long_poll_resp = LongPollResponse(
status_code=resp.status_code,
content=resp.content
)
else:
print("Response code is not 200", flush=True)
print(resp.status_code, flush=True)
long_poll_resp = LongPollResponse(status_code=resp.status_code)
q.put(long_poll_resp)
@contextmanager
def long_pool_listener_for_charger(param_one):
q = Queue()
p = Process(target=_wait_for_long_poll_resp, args=[charge_point, q])
try:
print("In try block")
p.start()
sleep(2.0)
yield q
except Exception as e:
raise Exception("Error occurred: ", e)
finally:
p.join()
@contextmanager
def long_pool_listener(param_one):
with long_pool_listener_for_charger(param_one) as listener:
yield listener
然后我正在使用 pytest 为此编写一个测试:
def test_remote_reset():
param_one = "something"
with long_pool_listener(param_one) as q: # type: Queue
sleep(3)
request_url = <some_url>
headers = <some_headers>
auth = <auth>
print(f'Calling {request_url}')
resp = requests.post(request_url, headers=headers, auth=auth)
# Give it some time
sleep(3)
resp = q.get(timeout=15)
# Some assertions
此代码正确吗?因为它似乎甚至没有进入功能,_wait_for_long_poll_resp。我知道这一点,因为它甚至没有打印我在这里编写的调试打印。还想知道它是否足够蟒蛇。
答: 暂无答案
评论
long_pool_listener_for_charger
永远不会回来,因为永远不会退出。它将在通话时阻止。如果你要阻止结果,那么剥离一个进程有什么意义呢?_wait_for_long_poll_resp
join
@contextmanager
yield
return q