Golang - API Server 和 Socket 同时进行

Question

我尝试制作套接字以与我的客户进行通信。在对我的 API 发出一些请求后，将创建一个套接字。这意味着，客户端连接自己（仅通过请求），但随后，他加入了聊天，因此创建了一个套接字并将其链接到良好的频道。我已经使用过套接字，所以我了解它是如何工作的（C、C++、C#、Java），但是我想做的，我在网上看到的，我认为这是可能的，但我不明白如何处理它与golang。我创建了第一台服务器：func main() {    r := mux.NewRouter()    r.HandleFunc("/", HomeHandler)    r.HandleFunc("/products", ProductsHandler)    r.HandleFunc("/articles", ArticlesHandler)    http.Handle("/", r)}但是对于套接字，我需要另一个吗？package mainimport "net"import "fmt"import "bufio"import "strings" // only needed below for sample processingfunc main() {    fmt.Println("Launching server...")    // listen on all interfaces    ln, _ := net.Listen("tcp", ":8081")    // accept connection on port    conn, _ := ln.Accept()       // run loop forever (or until ctrl-c)    for {             // will listen for message to process ending in newline (\n)        message, _ := bufio.NewReader(conn).ReadString('\n')        // output message received             fmt.Print("Message Received:", string(message))        // sample process for string received             newmessage := strings.ToUpper(message)        // send new string back to client             conn.Write([]byte(newmessage + "\n"))       } }感谢您的帮助！

Answer 1

默认 http 服务器仅接受一个“主机：端口”上的连接

答案取决于您将使用什么协议通过套接字进行通信。

我建议这样做：（非常简化）

让 http.Server 单独为您的 API 提供服务（它实现了 HTTP 1.*/2 协议，因此您无需担心）

实现你自己的“MultiSocketServer”，这样做：

2.1 实现 GracefulListener（必须实现 net.Listener）（当你不再需要它们时，你需要关闭你的套接字，对吧？）

2.2 实现 MultiSocketServer.Serve(l GracefulListener) (hello http.Server.Serve() ) 来服务单个连接（你通过套接字与客户端通信的协议在这里。像 net.textproto 这样的东西很容易实现，因为你 GracefulListener.Accept () 将返回 net.Conn)

2.3 添加方法MultiSocketServer.ListenAndServe(addr), MultiSocketServer.StopServe(l GracefulListener) 到你的MultiSocketServer

type MultiSocketServer struct {

    listeners GracefulListener[] //or map?

    // lots of other stuff

}

// looks familiar? (http.Server.ListenAndServe) 

func (s *MultiSocketServer) ListenAndServe(addr string) {

    ln, err := net.Listen("tcp", addr)

    graceful_listner = &GracefulListener(ln)

    s.listeners = append(s.listeners, graceful_listner)

    go s.Serve(graceful_listner)

    return graceful_listner

}

func (s *MultiSocketServer) StopServe(graceful_listner  GracefulListener) {

    graceful_listner.Stop()

    //pseudocode

    remove_listener_from_slice(s.listeners, graceful_listner)

}

当然，您需要添加错误检查和互斥锁（可能）来保护 MultiSocketServer.listeners 以使其线程安全。

在您的 main() 中启动您的 API http.Server，并初始化您的 MultiSocketServer。现在，从 http.Server 的 http.Handler/http.HandlerFunc 中，您应该能够调用 MultiSocketServer.ListenAndServe(addr) 来监听和服务您的套接字连接。

根据问题更新

但是，我不确定是否理解“在您的 main() 中”部分。如果我理解得很好，你说我有我的API，启动它后，我初始化MultiSocketServer。但是哪里？在我的 API 启动之后？或者您的意思是我将您的代码逻辑用作 API 会更好？每个请求都通过一个套接字

顺便说一句：更新 MultiSocketServer.ListenAndServe 以启动 Listen 并返回 graceful_listner

func main() {

    //init MultiSocketServer

    multi_socket_server = &MultiSocketServer{} //nil for GracefulListener[] is fine for now, complex initialization will be added later

    // no listners yet, serves nothing

    // create new Handeler for your "socket requests"

    SocketRequestHandler := function(w http.ResponseWriter, r *http.Request) {

       // identify client, assign him an address to connect

       addr_to_listen := parse_request(r) //pseudocode

       listener := multi_socket_server.ListenAndServe(addr_to_listen)

       // TODO: handle errors

       // now your multi_socket_server listen to addr_to_listen and serves it with multi_socket_server.Serve method in its own goroutine

       // as i said MultiSocketServer.Serve method must implement your protocol (plaintext Reader/Writer on listener for now?)

       save_listener_in_context_or_whatever_you_like_to_track_it(listener) //pseudo

   }  

    SocketDisconnectHandler := function(w http.ResponseWriter, r *http.Request) {

      // identify client 

      some_client := parse_request(r) //pseudocode

      // get listener based on info

      listener := get_listener_from_context_or_whatever(some_client) //pseudo

      multi_socket_server.StopServe(listener)

      // TODO: handle errors

    }

    //initialize your API http.Server

    r := mux.NewRouter()

    r.HandleFunc("/", HomeHandler)

    r.HandleFunc("/products", ProductsHandler)

    r.HandleFunc("/articles", ArticlesHandler)

    r.HandleFunc("/socket_request", SocketRequestHandler) // added

    r.HandleFunc("/socket_disconnect", SocketDisconnectHandler) //added

    http.Handle("/", r)

    // it creates new http.Server with DefaultServeMux as Handler (which is configured with your http.Handle("/", r) call)

   http.ListenAndServe(":8080") // start serving API via HTTP proto

}

实际上，您可以调用multi_socket_server.ListenAndServe(addr_to_listen)和multi_socket_server.StopServe(listener)从您的 API 服务器中的任何处理程序。

每次调用multi_socket_server.ListenAndServe(addr_to_listen)它都会创建新的侦听器并为其提供服务，您必须控制它（不要在同一个地址上侦听一次以上（我认为它无论如何都会出错））

您的 MultiSocketServer.Serve 可能如下所示：

func (s *MultiSocketServer) Serve(l net.Listener) {

    defer l.Close()

    for {     

        // will listen for message to process ending in newline (\n)

        message, _ := bufio.NewReader(conn).ReadString('\n')

        // output message received     

        fmt.Print("Message Received:", string(message))

        // sample process for string received     

        newmessage := strings.ToUpper(message)

        // send new string back to client     

        conn.Write([]byte(newmessage + "\n"))   

    } 

}

可能的 GracefulListener实现 github

还是您想实现完全不同的目标？=)

Answer 2

基于我们的聊天讨论。

OVER带有大量伪代码的简化示例

import (

    "net"

    "encoding/json"

    "errors"

)

type User struct {

    name string

}

type Message {

    Action string

    Params map[string]string

}

type Server struct {

    connected_users map[*User]net.Conn

    users_connected_with_each_other map[*User][]*User

    good_users map[string]*User

}

func (srv *Server) ListenAndServe(addr string) error {

    ln, err := net.Listen("tcp", addr)

    if err != nil {

        return err

    }

    return srv.Serve(tcpKeepAliveListener{ln.(*net.TCPListener)})

}

func (srv *Server) Serve(l net.Listener) error {

    defer l.Close()

    for {

        rw, e := l.Accept()

        if e != nil {

            return e

        }

        // you want to create server_conn here with buffers, channels and stuff

        // to use async thread safe read/write from it

        go srv.serve_conn(rw)

    }

}

func (srv *Server) serve_conn(rw net.Conn) error {  

    dec := json.NewDecoder(rw)

    var message Message

    //read 1st message he sent, should be token to connect

    dec.Decode(&message)

    token := get_token(Message)

    user, ok := srv.good_users[token]

    if !ok {

        return errors.New("BAD USER!")

    }

    // store connected user

    srv.connected_users[user] = rw

    for {

        // async reader will be nice

        dec.Decode(&message)

        switch message.Action {

            case "Message":

                // find users to send message to

                if chats_with, err := users_connected_with_each_other[user]; err == nil {                   

                    for user_to_send_message_to := range chats_with {

                        // find connections to send message to

                        if conn, err := srv.connected_users[user_to_send_message_to]; err == nil {

                            // send json encoded message

                            err := json.NewEncoder(conn).Encode(message)

                            //if write failed store message for later

                        }

                    }

                }

            //other cases

            default:

                // log?

        }       

    }

}

func main() {

    known_users_with_tokens := make(map[string]*User)

    srv := &Server{

        connected_users: make(map[*User]net.Conn),

        users_connected_with_each_other: make(map[*User][]*User),

        good_users: known_users_with_tokens, // map is reference type, so treat it like pointer

    }

    // start our server

    go srv.ListenAndServe(":54321")

    ConnRequestHandler := function(w http.ResponseWriter, r *http.Request) {

        user := create_user_based_on_request(r)

        token := create_token(user)

        // now user will be able to connect to server with token

        known_users_with_tokens[token] = user

    }

    ConnectUsersHandler := function(user1,user2) {

        // you should guard your srv.* members to avoid concurrent read/writes to map

        srv.users_connected_with_each_other[user1] = append(srv.users_connected_with_each_other[user1], user2)

        srv.users_connected_with_each_other[user2] = append(srv.users_connected_with_each_other[user2], user1)

    }

    //initialize your API http.Server

    r := mux.NewRouter()

    r.HandleFunc("/", HomeHandler)

    r.HandleFunc("/products", ProductsHandler)

    r.HandleFunc("/articles", ArticlesHandler)

    r.HandleFunc("/connection_request", ConnRequestHandler) // added

    http.Handle("/", r)

}

打电话ConnectUsersHandler(user1, user2)让他们互相交流。

known_users_with_tokens[token] = user 允许用户连接到服务器

您需要实现异步读取器/写入器以连接到您的服务器。保持良好用户的有用结构。保护服务器结构成员并提供线程安全访问来更新它。

UDP

看起来像json.NewEncoder(connection).Encode(&message)并且json.NewDecoder(connection).Decode(&message)是异步和线程安全的。因此，您可以从不同的 goroutine 同时编写代码。无需手动同步，耶！