python의 스레드와 비동기async await문제입니다..너무 어려워요...(사용금지단어몰라서 계속수정하면서등록중입니다)

Question

현재 문제가redis에서 pubsub로 구독으로 데이터를 가져오는데이 데이터가 들어올때마다 async await로 이루어져있는 함수를 호출해서 websocket으로 프론트로 전송하는 것입니다.원래 전부 async await방식으로 되어있어서 크게 문제는 없었지만사수가 데이터 영속성 문제도 있고 불시에 서버가 끊겼다가 복구될때 기존 정보를 자동으로 보관해주는 redis라는 메세지큐? 메세지브로커 써야겠다고 celery로 싹 바꾸고나서 저의 고생길이 시작되엇습니다. (근데 저는 프론트엔드 담당...)뭐 제대로 테스트 안하고 코드를 전부 바꿔버려서;;;;;;;;;;제가 스레드, 비동기에 대한 이해도가 너무너무 낮아서 진짜 해결을 못하겠습니다................def handle_redis_messages(): redis_client &#61; redis.StrictRedis(host&#61;'redis', port&#61;6379, db&#61;0) pubsub &#61; redis_client.pubsub() pubsub.subscribe('whattime') for message in pubsub.listen(): if message['type'] &#61;&#61; 'message': s &#61; str(message['data'], 'utf-8') realtime_info.alarm_area.water_countdown_sec &#61;s loop &#61; asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(send_realtime_info(realtime_info)) loop.close() # thread_send_realtime_info(realtime_info) # asyncio.get_event_loop() # asyncio.set_event_loop(event_loop) # await send_realtime_info(realtime_info) async def send_realtime_info(info) -> None: for client in conn.get_connects().values(): await client.send_json(jsonable_encoder(vars(info))) # Redis 채널을 구독하는 스레드 시작 redis_thread &#61; threading.Thread(target&#61;handle_redis_messages) redis_thread.start()thread를 사용한 이유는for message in pubsub.listen() <- 이게 구독대기상태로 만들어버려서 메인 프로세스를 전부 블록해버리더라구요...그래서 thread로 분리하니까 다른 코드들이 작동하기 시작했습니다.async def send_realtime_info(info) -> None: for client in conn.get_connects().values(): await client.send_json(jsonable_encoder(vars(info)))이건 메인프로세스에 웹소켓으로 접속한 클라이언트들에게 데이터를 전송하는 부분인데에러를 보면 await client.send_json(jsonable_encoder(vars(info)))이부분에서 loop충돌이 발생합니다.진짜 오만거 다해본거같은데.....도저히 모르겠어요.................redis 구독 구현은 귀찮으실테니...아마도 그냥 메소드하나 스레드로 분리하셔서 for반복문으로 카운트다운 10부터 1까지 하면서매회 async def send_realtime_info(info)을 호출해야한다고 보시면됩니다..고수님들 부탁드립니다....가르침을 주시옵소서ㅠㅠ

ingn · Answer

저도 최근에 파이썬 멀티스레드 비동기를 좀 만져서 설명이 될거 같은데파이썬은 1스레드에 1이벤트루프만 가능합니다.폰으로 코드 봤는데 들여쓰기가 제대로 안되게 보여서 컴터로 봤는데, 스레드에서 계속 이벤트 루프 생성하고 있네요.일단 저도 찾아볼때 제일 잘 설명되어있는 블로그가 이거였어요. https://it-eldorado.tistory.com/159

ingn · Answer

def Send():loop &#61; asyncio.get_event_loop() # 이벤트루프 획득, 만약 현재 범위(스레드)에 없으면 생성한다.loop.run_until_complete(Send_Data(datas)) # 모든 비동기 작업이 완료될때까지 함수를 실행한다.loop.close() # 이벤트루프 해제.async def Send_Data(self, datas):start_time &#61; Nonetasks &#61; []data_count &#61; 0;tick_second &#61; 1;async with httpx.AsyncClient() as session:for data in datas:task &#61; asyncio.ensure_future(self.Fetch(session, self.logger, self.host, data)) # 코루틴 생성tasks.append(task);await asyncio.sleep(self.latency); # run coroutine(Passing control to the event loop) 코루틴 실행responses &#61; await asyncio.gather(*tasks, return_exceptions&#61;True) # wait all response 모든 코루틴 완료 대기async def Fetch(self, session, logger, host, data) -> tuple:try:response &#61; await session.post(host, json&#61;data)return (response.json(),None)except Exception as error:logger.debug(&#34;sending error raise %s&#34;, error)return (error, data)finally:pass 그리고 이건 제가 짰던 비동기 http요청 코드인데, 많이 지워서 실행은 안될거지만 간략하게 설명하면 Send함수에서 이벤트루프만들고 Send_Data함수에서 해당 이벤트루프를 사용하고 ensure_future로 코루틴 만들고 asyncio.sleep 으로 해당 코루틴이 실행되게 하는 구조에요. 위에 올려주신 코드도 비슷하게 한다면 def temp_func():loop &#61; asyncio.get_event_loop()loop.run_until_complete(handle_redis_messages())loop.close()async def handle_redis_messages():redis_client &#61; redis.StrictRedis(host&#61;'redis', port&#61;6379, db&#61;0)pubsub &#61; redis_client.pubsub()pubsub.subscribe('whattime')for message in pubsub.listen():if message['type'] &#61;&#61; 'message':s &#61; str(message['data'], 'utf-8')realtime_info.alarm_area.water_countdown_sec &#61;stask &#61; asyncio.ensure_future(send_realtime_info(realtime_info)) # 코루틴 생성await asyncio.sleep(0); # run coroutine(Passing control to the event loop) 코루틴 실행async def send_realtime_info(info) -> None:for client in conn.get_connects().values():await client.send_json(jsonable_encoder(vars(info)))# Redis 채널을 구독하는 스레드 시작redis_thread &#61; threading.Thread(target&#61;temp_func)redis_thread.start() 대충 이정도 구조로 짜면 될거 같네요.실제로는 코루틴이라기보다는 task나 future객체등으로 사용하는데 간단하게 코루틴으로 설명했어요저도 실행해본게 아니라서 완벽한건 아니니 참고만 해주세요.

python의 스레드와 비동기async await문제입니다..너무 어려워요...(사용금지단어몰라서 계속수정하면서등록중입니다)

2개의 답변

Q&A 목록