Go语言WebSocket调用mysql客户端无输出及异常退出问题求助

问题描述 #

我用Golang开发了一个通过WebSocket执行SQL或命令的应用，依赖gorilla/websocket、os/exec包。目前redis-cli和mongoshell适配正常，但MySQL客户端存在两个问题：

• 无法通过WebSocket接收进程返回的消息，只有将进程输出定向到本地stdout时才能正常显示
• 发送错误SQL后进程会莫名退出

问题可通过gorilla/websocket官方命令示例复现，原代码如下：

package main

import (
    "flag"
    "fmt"
    "io"
    "log"
    "net/http"
    "os/exec"
    "time"

    "github.com/gorilla/websocket"
)

var (
    addr    = flag.String("addr", "127.0.0.1:8080", "http service address")
)

const (
    // Time allowed to write a message to the peer.
    writeWait = 10 * time.Second

    // Maximum message size allowed from peer.
    maxMessageSize = 8192

    // Time allowed to read the next pong message from the peer.
    pongWait = 20 * time.Second

    // Send pings to peer with this period. Must be less than pongWait.
    pingPeriod = (pongWait * 9) / 10

)

func ping(ws *websocket.Conn, done chan struct{}) {
    ticker := time.NewTicker(pingPeriod)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            if err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait)); err != nil {
                log.Println("ping:", err)
            }
            done <- struct{}{}
        case <-done:
            return
        }
    }
}

func pumpStdin(ws *websocket.Conn, w io.Writer) {
    defer ws.Close()
    ws.SetReadLimit(maxMessageSize)
    ws.SetReadDeadline(time.Now().Add(pongWait))
    ws.SetPongHandler(func(string) error { ws.SetReadDeadline(time.Now().Add(pongWait)); return nil })
    for {
        _, message, err := ws.ReadMessage()
        if err != nil {
            break
        }
        message = append(message, '\n')
        if _, err := w.Write(message); err != nil {
            break
        }
    }
}

func pumpStdout(ws *websocket.Conn, r io.ReadCloser, done chan struct{}) {
    for {
        x := make([]byte, 1024)
        n, err := r.Read(x)
        if err != nil {
            fmt.Println(err)
        }
        fmt.Println(string(x[:n]))
    }
}

var upgrader = websocket.Upgrader{}

func serveWs(w http.ResponseWriter, r *http.Request) {
    ws, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Println("upgrade:", err)
        return
    }

    defer ws.Close()
    
    stopchan := make(chan struct{})
    cmd := exec.Command("/usr/bin/mysql", "-h127.0.0.1", "-ppassword", "-P3306", "-uroot")
    
    cmdin, err := cmd.StdinPipe()
    if err != nil {
        fmt.Println(err)
    }
    cmdout, err := cmd.StdoutPipe()
    if err!=nil{
        fmt.Println(err)
    }

    if err := cmd.Start(); err != nil {
        log.Fatal(err)
    }
    go ping(ws, stopchan)
    go pumpStdout(ws, cmdout, stopchan)
    pumpStdin(ws, cmdin)
    if err = cmd.Wait(); err != nil {
        fmt.Println(err)
    }

}

func main() {
    flag.Parse()
    http.HandleFunc("/ws", serveWs)
    log.Fatal(http.ListenAndServe(*addr, nil))
}

问题原因分析 #

• 无法接收消息：原pumpStdout函数仅将进程输出打印到本地终端，未将内容通过WebSocket连接发送给客户端；同时未捕获MySQL的stderr输出，MySQL的错误信息默认输出到stderr，导致这部分内容完全丢失。
• 进程莫名退出：当MySQL输出错误信息到stderr时，若stderr未被读取，进程会因输出缓冲区满而阻塞或退出；另外原代码未正确处理goroutine的退出通知，导致进程生命周期管理混乱。

修复后代码 #

package main

import (
    "flag"
    "io"
    "log"
    "net/http"
    "os/exec"
    "time"

    "github.com/gorilla/websocket"
)

var (
    addr = flag.String("addr", "127.0.0.1:8080", "http service address")
)

const (
    writeWait      = 10 * time.Second
    maxMessageSize = 8192
    pongWait       = 20 * time.Second
    pingPeriod     = (pongWait * 9) / 10
)

var upgrader = websocket.Upgrader{
    CheckOrigin: func(r *http.Request) bool {
        return true // 根据实际业务需求修改跨域校验规则
    },
}

func ping(ws *websocket.Conn, done chan struct{}) {
    ticker := time.NewTicker(pingPeriod)
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            ws.SetWriteDeadline(time.Now().Add(writeWait))
            if err := ws.WriteMessage(websocket.PingMessage, nil); err != nil {
                log.Println("ping error:", err)
                return
            }
        case <-done:
            return
        }
    }
}

func pumpStdin(ws *websocket.Conn, w io.WriteCloser, done chan struct{}) {
    defer func() {
        w.Close()
        close(done)
    }()
    ws.SetReadLimit(maxMessageSize)
    ws.SetReadDeadline(time.Now().Add(pongWait))
    ws.SetPongHandler(func(string) error {
        ws.SetReadDeadline(time.Now().Add(pongWait))
        return nil
    })
    for {
        _, message, err := ws.ReadMessage()
        if err != nil {
            log.Println("read message error:", err)
            break
        }
        message = append(message, '\n')
        ws.SetWriteDeadline(time.Now().Add(writeWait))
        if _, err := w.Write(message); err != nil {
            log.Println("write to stdin error:", err)
            break
        }
    }
}

func pumpOutput(ws *websocket.Conn, r io.ReadCloser, done chan struct{}) {
    defer r.Close()
    buf := make([]byte, 1024)
    for {
        select {
        case <-done:
            return
        default:
            n, err := r.Read(buf)
            if err != nil {
                if err != io.EOF {
                    log.Println("read output error:", err)
                }
                return
            }
            if n == 0 {
                continue
            }
            ws.SetWriteDeadline(time.Now().Add(writeWait))
            if err := ws.WriteMessage(websocket.TextMessage, buf[:n]); err != nil {
                log.Println("write to websocket error:", err)
                return
            }
        }
    }
}

func serveWs(w http.ResponseWriter, r *http.Request) {
    ws, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Println("upgrade error:", err)
        return
    }
    defer ws.Close()

    done := make(chan struct{})
    defer close(done)

    // 添加--force参数让MySQL执行错误SQL后不退出会话，可根据需求移除
    cmd := exec.Command("/usr/bin/mysql", "-h127.0.0.1", "-ppassword", "-P3306", "-uroot", "--force")

    stdin, err := cmd.StdinPipe()
    if err != nil {
        log.Println("stdin pipe error:", err)
        return
    }
    stdout, err := cmd.StdoutPipe()
    if err != nil {
        log.Println("stdout pipe error:", err)
        return
    }
    stderr, err := cmd.StderrPipe()
    if err != nil {
        log.Println("stderr pipe error:", err)
        return
    }

    if err := cmd.Start(); err != nil {
        log.Fatal("start cmd error:", err)
    }
    defer func() {
        cmd.Process.Kill()
        cmd.Wait()
    }()

    go ping(ws, done)
    go pumpOutput(ws, stdout, done)
    go pumpOutput(ws, stderr, done)
    pumpStdin(ws, stdin, done)
}

func main() {
    flag.Parse()
    http.HandleFunc("/ws", serveWs)
    log.Fatal(http.ListenAndServe(*addr, nil))
}

关键修复点 #

• 转发输出到WebSocket：重构pumpOutput函数，将stdout和stderr的内容通过ws.WriteMessage发送给客户端，不再仅打印到本地。
• 捕获stderr输出：新增stderrPipe捕获MySQL的错误信息，确保错误内容能传递到客户端。
• 进程生命周期管理：
  • 在pumpStdin退出时关闭stdin并关闭done chan，通知其他goroutine退出
  • 给serveWs添加defer逻辑，确保WebSocket断开后杀死MySQL进程并等待退出
• MySQL参数调整：添加--force参数，让MySQL在执行错误SQL后不退出，保持交互会话（可根据需求移除）
• 完善错误处理：增加各步骤的错误日志，便于排查问题；设置WebSocket的读写超时，避免连接僵死

内容的提问来源于stack exchange，提问作者shushuangyan