파이썬으로 웹소켓 서버 돌리기

 

웹 프론트엔드가 데이터를 실시간으로 보내기 위해 파이썬으로 웹소켓 서버를 구성해 보았다.

 

웹서버는 sanic을 사용했고, pub/sub기능을 사용하기 위해 redis(인메모리 DB)를 사용했다.

 

(처음에 fastapi, websocket패키지만 사용해서 해봤지만 자꾸 Broken pipe에러가 발생해서 포기했다.)

 

 

대강의 구조는 아래와 같다. (웹소켓이나 pub/sub기능쪽에서 원래 사용하는 용어는 잘 모른다.)

 

데이터는 데이터 생성자에서 사용자쪽으로 흘러간다.

 

 

fastapi,websocket 패키지로 대충 해보려던 시도가 끝나고, 뭔가 다른 방식이 있나 검색해보다 뭔가 있어 보이는 코드를 발견했다.

 

코드 가져다가 그냥 돌려보고 싶으나 귀찮게도 웹서버 돌리는 부분이 없고, 테스트해볼 클라이언트 코드도 없다. 

 

 

https://gist.github.com/ahopkins/5b6d380560d8e9d49e25281ff964ed81

Sanic websocket feeds - v3

 

별도로 설치해줘야 하는 패키지는 aioredis, sanic, rel 정도이다.

 

다음 코드에서 앱이름, redis주소, 웹서버 주소를 적당히 넣어주고 python 파일이름.py로 돌리면 된다.

 

from sanic import Sanic
from blueprint import bp
import aioredis
from typing import Any

app = Sanic("앱이름_아무렇게나")
app.blueprint(bp)

@app.before_server_start
async def setup_redis(app: Sanic, _: Any) -> None:
    app.ctx.redis_pool = aioredis.BlockingConnectionPool.from_url(
        app.config.REDIS_DSN, max_connections=app.config.REDIS_MAX
    )
    app.ctx.redis = aioredis.Redis(connection_pool=app.ctx.redis_pool)
    app.ctx.pubsub = app.ctx.redis.pubsub()

if __name__ == "__main__":
    app.config.REDIS_DSN = 'redis://REID주소:포트/1'
    app.config.REDIS_MAX = 10000
    app.run(host='웹서버 주소',port=포트,access_log=True)

 

서버가 돌아가기 시작하면 '데이터생성자'로부터 redis를 통해 데이터를 받을 준비 및 redis에서 받은 데이터를 사용자들에게 보내줄 준비가 끝난다. 

 

데이터를 넣는 코드는 아래와 같다. redis주소랑 채널 이름을 적당히 적어주고 역시나 python 파일명.py로 돌려주면 된다.

 

import aioredis
import tim
import asyncio 


REDIS_DSN = 'redis://REDIS주소:포트/1'
REDIS_MAX = 10000 # 이건 그냥 막 쓴 숫자임

redis_pool = aioredis.BlockingConnectionPool.from_url(
    REDIS_DSN, max_connections=REDIS_MAX
)
redis = aioredis.Redis(connection_pool=redis_pool)

async def pub():
    while True:
        await redis.publish('채널이름',price)   
        
asyncio.run(pub())

 

사용자 쪽은 웹인데 아직 안 만들었고, 테스트로 받은 파이썬 코드는 아래와 같다.(출처)

 

import websocket
import _thread
import time
import rel

def on_message(ws, message):
    print(message)

def on_error(ws, error):
    print(error)

def on_close(ws, close_status_code, close_msg):
    print("### closed ###")

def on_open(ws):
    print("Opened connection")

if __name__ == "__main__":
    websocket.enableTrace(False)
    ws = websocket.WebSocketApp("ws://웹서버주소:포트/feed/채널이름",    
                              on_open=on_open,
                              on_message=on_message,
                              on_error=on_error,
                              on_close=on_close)

    ws.run_forever(dispatcher=rel)  # Set dispatcher to automatic reconnection
    rel.signal(2, rel.abort)  # Keyboard Interrupt
    rel.dispatch()