在 Golang 中将 websocket 循环与通道同步

Question

我在这里面临两难境地，试图让特定用户的某些 websocket 保持同步。这是基本设置：

type msg struct {
    Key         string
    Value   string
}

type connStruct struct {
    //...

    ConnRoutineChans []*chan string
    LoggedIn        bool
    Login       string

    //...

    Sockets         []*websocket.Conn
}

var (
    //...

    /*  LIST OF CONNECTED USERS AN THEIR IP ADDRESSES  */

        guestMap sync.Map

)

func main() {
    post("Started...")
    rand.Seed(time.Now().UTC().UnixNano())
    http.HandleFunc("/wss", wsHandler)
    panic(http.ListenAndServeTLS("...", "...", "...", nil))
}

func wsHandler(w http.ResponseWriter, r *http.Request) {
    if r.Header.Get("Origin")+":8080" != "https://...:8080" {
        http.Error(w, "Origin not allowed", 403)
        fmt.Println("Client origin not allowed! (https://"+r.Host+")")
        fmt.Println("r.Header Origin: "+r.Header.Get("Origin"))
        return
    }
    ///
    conn, err := websocket.Upgrade(w, r, w.Header(), 1024, 1024)
    if err != nil {
        http.Error(w, "Could not open websocket connection", http.StatusBadRequest)
        fmt.Println("Could not open websocket connection with client!")
    }

    //ADD CONNECTION TO guestMap IF CONNECTION IS nil
    var authString string = /*gets device identity*/;
    var authChan chan string = make(chan string);
    authValue, authOK := guestMap.Load(authString);
    if !authOK {
        // NO SESSION, CREATE A NEW ONE
        newSession = getSession();
        //defer newSession.Close();
        guestMap.Store(authString, connStruct{ LoggedIn: false,
                                ConnRoutineChans: []*chan string{&authChan},
                                         Login: "",
                                        Sockets: []*websocket.Conn{conn}
                                        /* .... */ });
    }else{
        //SESSION STARTED, ADD NEW SOCKET TO Sockets
        var tempConn connStruct = authValue.(connStruct);
        tempConn.Sockets = append(tempConn.Sockets, conn);
        tempConn.ConnRoutineChans = append(tempConn.ConnRoutineChans, &authChan)
        guestMap.Store(authString, tempConn);
    }

    //
    go echo(conn, authString, &authChan);
}

func echo(conn *websocket.Conn, authString string, authChan *chan string) {

    var message msg;

    //TEST CHANNEL
    authValue, _ := guestMap.Load(authString);
    go sendToChans(authValue.(connStruct).ConnRoutineChans, "sup dude?")

    fmt.Println("got past send...");

    for true {
        select {
            case val := <-*authChan:
                // use value of channel
                fmt.Println("AuthChan for user #"+strconv.Itoa(myConnNumb)+" spat out: ", val)
            default:
                // if channels are empty, this is executed
        }

        readError := conn.ReadJSON(&message)
        fmt.Println("got past readJson...");

        if readError != nil || message.Key == "" {
            //DISCONNECT USER
            //.....
            return
        }

        //
        _key, _value := chief(message.Key, message.Value, &*conn, browserAndOS, authString)

        if writeError := conn.WriteJSON(_key + "|" + _value); writeError != nil {
            //...
            return
        }

        fmt.Println("got past writeJson...");
    }
}

func sendToChans(chans []*chan string, message string){
    for i := 0; i < len(chans); i++ {
        *chans[i] <- message
    }
}

我知道，一大段代码是吗？我注释掉了大部分...

不管怎样，如果你曾经使用过 websocket，大部分应该都非常熟悉：

1) func wsHandler() 在每次用户连接时触发。它在guestMap（对于每个连接的唯一设备）中创建一个条目，其中包含一个connStruct，其中包含一个频道列表：ConnRoutineChans []*chan string。这一切都传递给：

2) echo()，这是一个不断为每个 websocket 连接运行的 goroutine。在这里，我只是在测试向其他正在运行的 goroutine 发送消息，但似乎我的 for 循环实际上并没有持续触发。它仅在 websocket 从它所连接的打开的选项卡/窗口接收到消息时触发。 （如果有人能澄清这个机制，我很想知道为什么它不经常循环？）

3) 对于用户在给定设备上打开的每个窗口或选项卡，都有一个 websocket 和存储在数组中的通道。我希望能够向数组中的所有通道（本质上是该设备上打开的选项卡/窗口的其他 goroutines）发送消息，并在其他 goroutines 中接收消息以更改在不断运行的 goroutine 中设置的一些变量。

我现在所拥有的仅适用于设备上的第一次连接，并且（当然）它会发送“sup dude？”因为它是当时阵列中唯一的通道。然后，如果您打开一个新标签（甚至很多），则消息根本不会发送给任何人！奇怪？...然后当我关闭所有选项卡（并且我注释掉的逻辑从guestMap 中删除设备项）并启动新设备会话时，仍然只有第一个连接得到它自己的消息。

我已经有了向设备上所有其他 websocket 发送消息的方法，但是发送到 goroutine 似乎比我想象的要复杂一些。

Answer 1

回答我自己的问题：

首先，我已从 sync.map 切换到法线贴图。其次，为了让没有人同时读/写它，我创建了一个通道，您可以调用它来在地图上进行任何读/写操作。我一直在尽最大努力保持我的数据访问和操作快速执行，这样通道就不会那么容易拥挤。这是一个小例子：

package main

import (
    "fmt"
)

var (
  guestMap map[string]*guestStruct = make(map[string]*guestStruct);
  guestMapActionChan = make (chan actionStruct);

)

type actionStruct struct {
    Action      func([]interface{})[]interface{}
    Params      []interface{}
    ReturnChan  chan []interface{}
}

type guestStruct struct {
    Name string
    Numb int
}

func main(){
    //make chan listener
    go guestMapActionChanListener(guestMapActionChan)

    //some guest logs in...
    newGuest := guestStruct{Name: "Larry Josher", Numb: 1337}

    //add to the map
    addRetChan := make(chan []interface{})
    guestMapActionChan <- actionStruct{Action: guestMapAdd,
                                       Params: []interface{}{&newGuest},
                                       ReturnChan: addRetChan}
    addReturned := <-addRetChan

    fmt.Println(addReturned)
    fmt.Println("Also, numb was changed by listener to:", newGuest.Numb)

    // Same kind of thing for removing, except (of course) there's
    // a lot more logic to a real-life application.
}

func guestMapActionChanListener (c chan actionStruct){
    for{
        value := <-c;
        //
        returned := value.Action(value.Params);
        value.ReturnChan <- returned;
        close(value.ReturnChan)
    }
}

func guestMapAdd(params []interface{}) []interface{} {
    //.. do some parameter verification checks
    theStruct := params[0].(*guestStruct)
    name := theStruct.Name
    theStruct.Numb = 75
    guestMap[name] = &*theStruct

    return []interface{}{"Added '"+name+"' to the guestMap"}
}

对于连接之间的通信，我只是让每个套接字循环保持它们的guestStruct，并有更多的guestMapActionChan 函数负责将数据分发给其他客人的guestStructs

现在，我不会将此标记为正确答案，除非我得到一些更好的建议，说明如何以正确的方式做这样的事情。但目前这是可行的，应该保证不会出现读取/写入地图的竞赛。

编辑：正确的方法应该是只使用sync.Mutex，就像我在（大部分）完成的项目GopherGameServer中所做的那样