Go 언어로 간단한 WebSocket 푸시 서비스 작성-Golang-php.cn

Go 언어로 간단한 WebSocket 푸시 서비스 작성

Push 서비스 구현

기본원리

서버가 구동되면 2개의 Handler가 등록됩니다.

websocketHandler는 업그레이드 요청을 보내고 WebSocket 연결로 업그레이드할 수 있는 브라우저를 제공하는 데 사용됩니다.

pushHandler는 외부 푸시 터미널에 푸시 데이터 전송 요청을 제공하는 데 사용됩니다.

브라우저는 먼저 websocketHandler(기본 주소는 ws://ip:port/ws)에 연결하고 업그레이드 요청은 WebSocket 연결이 설정되면 등록을 위해 등록 정보를 전송해야 합니다. 여기에 있는 등록 정보에는 토큰 정보가 포함되어 있습니다.

서버는 제공된 토큰을 확인하고 해당 userId(일반적으로 userId는 동시에 많은 토큰과 연결될 수 있음)를 획득하고 토큰, userId 및 conn(연결) 간의 관계를 저장하고 유지합니다.

푸시 끝은 pushHandler에 데이터 푸시 요청을 보냅니다(기본 주소는 ws://ip:port/push). 요청에는 userId 필드와 메시지 필드가 포함됩니다. 서버는 userId를 기반으로 현재 서버에 연결된 모든 연결을 얻은 다음 메시지를 하나씩 푸시합니다.

푸시 서비스의 실시간 특성으로 인해 푸시된 데이터는 캐시되지 않으며 캐시될 필요도 없습니다.

자세한 코드 설명

여기서는 코드의 기본 구조에 대해 간략하게 설명하고, Go 언어에서 흔히 사용되는 몇 가지 작성 방법과 패턴에 대해서도 이야기하겠습니다(나도 결국 다른 언어에서 Go 언어로 전환했습니다) , Go 언어도 꽤 젊습니다 . 제안 사항이 있으면 제출해 주세요.

Go 언어의 발명자와 일부 주요 관리자는 대부분 C/C++ 언어 출신이기 때문에 Go 언어의 코드도 C/C++ 시스템에 더 편향되어 있습니다.

먼저 서버의 구조를 살펴보겠습니다:

// Server defines parameters for running websocket server.
type Server struct {
    // Address for server to listen on
    Addr string

    // Path for websocket request, default "/ws".
    WSPath string

    // Path for push message, default "/push".
    PushPath string

    // Upgrader is for upgrade connection to websocket connection using
    // "github.com/gorilla/websocket".
    //
    // If Upgrader is nil, default upgrader will be used. Default upgrader is
    // set ReadBufferSize and WriteBufferSize to 1024, and CheckOrigin always
    // returns true.
    Upgrader *websocket.Upgrader

    // Check token if it's valid and return userID. If token is valid, userID
    // must be returned and ok should be true. Otherwise ok should be false.
    AuthToken func(token string) (userID string, ok bool)

    // Authorize push request. Message will be sent if it returns true,
    // otherwise the request will be discarded. Default nil and push request
    // will always be accepted.
    PushAuth func(r *http.Request) bool

    wh *websocketHandler
    ph *pushHandler
}

로그인 후 복사

HTTP 요청을 업그레이드하는 데 사용되는 gorilla/websocket 패키지의 개체인 Upgrader *websocket.Upgrader에 대해 이야기해 보겠습니다.

구조체에 매개변수가 너무 많으면 일반적으로 직접 초기화하지 않고 구조가 제공하는 New 메서드를 사용하는 것이 좋습니다.

// NewServer creates a new Server.func NewServer(addr string) *Server {    return &Server{
        Addr:     addr,
        WSPath:   serverDefaultWSPath,
        PushPath: serverDefaultPushPath,
    }
}

로그인 후 복사

이것은 외부 세계에 초기화 방법을 제공하기 위해 Go 언어를 일반적으로 사용하는 것이기도 합니다.

그런 다음 서버는 http 패키지 사용과 유사하게 ListenAndServe 메서드를 사용하여 포트를 시작하고 수신합니다.

// ListenAndServe listens on the TCP network address and handle websocket
// request.
func (s *Server) ListenAndServe() error {
    b := &binder{
        userID2EventConnMap: make(map[string]*[]eventConn),
        connID2UserIDMap:    make(map[string]string),
    }

    // websocket request handler
    wh := websocketHandler{
        upgrader: defaultUpgrader,
        binder:   b,
    }
    if s.Upgrader != nil {
        wh.upgrader = s.Upgrader
    }
    if s.AuthToken != nil {
        wh.calcUserIDFunc = s.AuthToken
    }
    s.wh = &wh
    http.Handle(s.WSPath, s.wh)

    // push request handler
    ph := pushHandler{
        binder: b,
    }
    if s.PushAuth != nil {
        ph.authFunc = s.PushAuth
    }
    s.ph = &ph
    http.Handle(s.PushPath, s.ph)

    return http.ListenAndServe(s.Addr, nil)
}

로그인 후 복사

여기서 websocketHandler와 pushHandler라는 두 개의 핸들러를 생성했습니다. websocketHandler는 브라우저와의 연결 설정 및 데이터 전송을 담당하고 pushHandler는 푸시 측 요청을 처리합니다.

여기에서 두 핸들러 모두 바인더 개체를 캡슐화하는 것을 볼 수 있습니다. 이 바인더는 토큰 <-> userID <-> Conn:

// binder is defined to store the relation of userID and eventConn
type binder struct {
    mu sync.RWMutex

    // map stores key: userID and value of related slice of eventConn
    userID2EventConnMap map[string]*[]eventConn

    // map stores key: connID and value: userID
    connID2UserIDMap map[string]string
}

로그인 후 복사

websocketHandler

websocketHandler의 구현을 구체적으로 살펴보세요.

// websocketHandler defines to handle websocket upgrade request.
type websocketHandler struct {
    // upgrader is used to upgrade request.
    upgrader *websocket.Upgrader

    // binder stores relations about websocket connection and userID.
    binder *binder

    // calcUserIDFunc defines to calculate userID by token. The userID will
    // be equal to token if this function is nil.
    calcUserIDFunc func(token string) (userID string, ok bool)
}

로그인 후 복사

매우 간단한 구조입니다. websocketHandler는 http.Handler 인터페이스를 구현합니다.

// First try to upgrade connection to websocket. If success, connection will
// be kept until client send close message or server drop them.
func (wh *websocketHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    wsConn, err := wh.upgrader.Upgrade(w, r, nil)
    if err != nil {
        return
    }
    defer wsConn.Close()

    // handle Websocket request
    conn := NewConn(wsConn)
    conn.AfterReadFunc = func(messageType int, r io.Reader) {
        var rm RegisterMessage
        decoder := json.NewDecoder(r)
        if err := decoder.Decode(&rm); err != nil {
            return
        }

        // calculate userID by token
        userID := rm.Token
        if wh.calcUserIDFunc != nil {
            uID, ok := wh.calcUserIDFunc(rm.Token)
            if !ok {
                return
            }
            userID = uID
        }

        // bind
        wh.binder.Bind(userID, rm.Event, conn)
    }
    conn.BeforeCloseFunc = func() {
        // unbind
        wh.binder.Unbind(conn)
    }

    conn.Listen()
}

로그인 후 복사

먼저 들어오는 http.Request를 websocket.Conn으로 변환한 다음 이를 사용자 정의된 wserver.Conn으로 패키징합니다(캡슐화 또는 조합은 Go 언어의 일반적인 사용법입니다. 기억하세요. Go에서는 상속이 없고 구성만 가능합니다.

그런 다음 Conn의 AfterReadFunc 및 BeforeCloseFunc 메서드를 설정한 다음 conn.Listen()을 시작합니다. AfterReadFunc는 Conn이 데이터를 읽은 후 토큰을 기반으로 userID를 확인하고 계산한 다음 바인딩이 바인딩을 등록한다는 것을 의미합니다. BeforeCloseFunc는 Conn이 닫히기 전에 바인딩 해제 작업을 수행합니다.

pushHandler

pushHandler는 이해하기 쉽습니다. 요청을 구문 분석한 다음 데이터를 푸시합니다.

// Authorize if needed. Then decode the request and push message to each
// realted websocket connection.
func (s *pushHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    if r.Method != http.MethodPost {
        w.WriteHeader(http.StatusMethodNotAllowed)
        return
    }

    // authorize
    if s.authFunc != nil {
        if ok := s.authFunc(r); !ok {
            w.WriteHeader(http.StatusUnauthorized)
            return
        }
    }

    // read request
    var pm PushMessage
    decoder := json.NewDecoder(r.Body)
    if err := decoder.Decode(&pm); err != nil {
        w.WriteHeader(http.StatusBadRequest)
        w.Write([]byte(ErrRequestIllegal.Error()))
        return
    }

    // validate the data
    if pm.UserID == "" || pm.Event == "" || pm.Message == "" {
        w.WriteHeader(http.StatusBadRequest)
        w.Write([]byte(ErrRequestIllegal.Error()))
        return
    }

    cnt, err := s.push(pm.UserID, pm.Event, pm.Message)
    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        w.Write([]byte(err.Error()))
        return
    }

    result := strings.NewReader(fmt.Sprintf("message sent to %d clients", cnt))
    io.Copy(w, result)
}
Conn
Conn (此处指 wserver.Conn) 为 websocket.Conn 的包装。

// Conn wraps websocket.Conn with Conn. It defines to listen and read
// data from Conn.
type Conn struct {
    Conn *websocket.Conn

    AfterReadFunc   func(messageType int, r io.Reader)
    BeforeCloseFunc func()

    once   sync.Once
    id     string
    stopCh chan struct{}
}

로그인 후 복사

주요 메서드는 Listen()입니다.

// Listen listens for receive data from websocket connection. It blocks
// until websocket connection is closed.
func (c *Conn) Listen() {
    c.Conn.SetCloseHandler(func(code int, text string) error {
        if c.BeforeCloseFunc != nil {
            c.BeforeCloseFunc()
        }

        if err := c.Close(); err != nil {
            log.Println(err)
        }

        message := websocket.FormatCloseMessage(code, "")
        c.Conn.WriteControl(websocket.CloseMessage, message, time.Now().Add(time.Second))
        return nil
    })

    // Keeps reading from Conn util get error.
ReadLoop:
    for {
        select {
        case <-c.stopCh:
            break ReadLoop
        default:
            messageType, r, err := c.Conn.NextReader()
            if err != nil {
                // TODO: handle read error maybe
                break ReadLoop
            }

            if c.AfterReadFunc != nil {
                c.AfterReadFunc(messageType, r)
            }
        }
    }
}

로그인 후 복사

웹소켓 연결이 닫힐 때 데이터 처리 및 연속 읽기를 주로 설정합니다.