[Python] 여러 웹소켓 연결하기

개요

Python으로 WebSocket 통신이 필요한 프로그램을 개발하고 있는데, 여러 웹소켓 통신 연결 및 유지가 필요한 상황이다. 평소 Python으로 개발을 할 때 특정 서버, 그것도 여러 서버와 지속적인 통신 연결을 유지할 상황이 많지 않았다. 이번 기회에 적당한 방법을 정리해 둔다.

 

 

웹소켓 통신 함수

먼저 웹소켓 통신을 유지할 함수를 작성한다. 함수는 2025.04.06-[Python] websockets - 연결 종료 시 자동 재연결에서 작성한 코드이다. 다만 재사용성을 확보할 수 있도록 통신할 URL과 전송할 데이터를 매개변수로 전달받을 수 있도록 수정했다.

import asyncio
from websockets.asyncio.client import connect
import json

async def connect_test(url, send_data):
    # 웹소켓 통신
    async with connect(url) as websocket:
        # 요청
        await websocket.send(json.dumps(send_data))
        while True:
            message = await websocket.recv()
            print(message)
            await asyncio.sleep(1)

 

 

다중 연결 유지

여러 웹소켓 통신을 유지하기 위한 방법으로 asyncio.create_task 및 asyncio.gather를 활용한다. 

asyncio.create_task는 task를 생성과 동시에 실행을 예약하며, asyncio.gather는 여러 개의 코루틴 또는 태스크를 모아서 처리한다.

async def main():
    # 연결 시 사용할 정보
    lst_request_info = [
        ("wss://ws-api.korbit.co.kr/v2/public", [{"method": "subscribe", "type": "orderbook", "symbols": ["btc_krw"]}]),
        ("wss://api.upbit.com/websocket/v1",
         [{"ticket": "test"}, {"type": "orderbook", "codes": ["KRW-BTC", "KRW-ETH.3"], "level": 10000},
          {"format": "DEFAULT"}])]

    # websocket 연결 task 생성 및 백그라운드 수행
    tasks = [asyncio.create_task(connect_test(url, send_data)) for url, send_data in lst_request_info]
    await asyncio.gather(*tasks)


if __name__ == "__main__":
    asyncio.run(main())

이제 여러 서버에서 요청한 데이터를 동시에 수집하여 출력하는 모습을 확인할 수 있다.

꼭 웹소켓 통신뿐만 아니라 이와 비슷한 여러 시간이 많이 필요한 작업에 사용하면 좋을 거 같다.

 

 

참고 문서

https://this-programmer.tistory.com/411

https://keun.me/pythoneseo-asyncio-gather-reul-hwalyonghan-goseongneung-bidonggi-peurogeuraeming/