reduce websocket conn lock contention

(cherry-picked from commit dac8933..83772e2)

go.mod

@@ -4,14 +4,18 @@ go 1.14
 
 require (
 	github.com/Mrs4s/MiraiGo v0.0.0-20200821111822-80481f0022d5
+	github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 // indirect
 	github.com/gin-gonic/gin v1.6.3
 	github.com/gorilla/websocket v1.4.2
 	github.com/guonaihong/gout v0.1.1
+	github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869 // indirect
+	github.com/jonboulle/clockwork v0.2.0 // indirect
 	github.com/lestrrat-go/file-rotatelogs v2.3.0+incompatible
 	github.com/lestrrat-go/strftime v1.0.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/sirupsen/logrus v1.6.0
 	github.com/t-tomalak/logrus-easy-formatter v0.0.0-20190827215021-c074f06c5816
+	github.com/tebeka/strftime v0.1.5 // indirect
 	github.com/tidwall/gjson v1.6.0
 	github.com/xujiajun/nutsdb v0.5.0
 	github.com/yinghau76/go-ascii-art v0.0.0-20190517192627-e7f465a30189

go.sum

@@ -7,14 +7,18 @@ github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239 h1:Ghm4eQYC0nEPnSJdVkTrXpu9KtoVCSo1hg7mtI7G9KU=
+github.com/fastly/go-utils v0.0.0-20180712184237-d95a45783239/go.mod h1:Gdwt2ce0yfBxPvZrHkprdPPTTS3N5rwmLE8T22KBXlw=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
 github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
 github.com/gin-gonic/gin v1.5.0/go.mod h1:Nd6IXA8m5kNZdNEHMBd93KT+mdY3+bewLgRvmCsR2Do=
 github.com/gin-gonic/gin v1.6.3 h1:ahKqKTFpO5KTPHxWZjEdPScmYaGtLo8Y4DMHoEsnp14=
 github.com/gin-gonic/gin v1.6.3/go.mod h1:75u5sXoLsGZoRN5Sgbi1eraJ4GU3++wFwWzhwvtwp4M=
+github.com/go-playground/assert/v2 v2.0.1 h1:MsBgLAaY856+nPRTKrp3/OZK38U/wa0CcBYNjji3q3A=
 github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
 github.com/go-playground/locales v0.12.1/go.mod h1:IUMDtCfWo/w/mtMfIE/IG2K+Ey3ygWanZIBtBW0W2TM=
 github.com/go-playground/locales v0.13.0 h1:HyWk6mgj5qFqCT5fjGBuRArbVDfE4hi8+e8ceBS/t7Q=
@@ -41,14 +45,21 @@ github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5a
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.0 h1:/QaMHBdZ26BB3SSst0Iwl10Epc+xhTquomWX0oZEB6w=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/guonaihong/gout v0.1.1 h1:2i3eqQ1KUhTlj7AFeIHqVUFku5QwUhwE2wNgYTVpbxQ=
 github.com/guonaihong/gout v0.1.1/go.mod h1:vXvv5Kxr70eM5wrp4F0+t9lnLWmq+YPW2GByll2f/EA=
+github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869 h1:IPJ3dvxmJ4uczJe5YQdrYB16oTJlGSC/OyZDqUk9xX4=
+github.com/jehiah/go-strftime v0.0.0-20171201141054-1d33003b3869/go.mod h1:cJ6Cj7dQo+O6GJNiMx+Pa94qKj+TG8ONdKHgMNIyyag=
+github.com/jonboulle/clockwork v0.2.0 h1:J2SLSdy7HgElq8ekSl2Mxh6vrRNFxqbXGenYH2I02Vs=
+github.com/jonboulle/clockwork v0.2.0/go.mod h1:Pkfl5aHPm1nk2H9h0bjmnJD/BcgbGXUBGnn1kMkgxc8=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns=
 github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3 h1:CE8S1cTafDpPvMhIxNJKvHsGVBgn1xWYf1NbHQhywc8=
@@ -56,6 +67,7 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxv
 github.com/leodido/go-urn v1.1.0/go.mod h1:+cyI34gQWZcE1eQU7NVgKkkzdXDQHr1dBMtdAPozLkw=
 github.com/leodido/go-urn v1.2.0 h1:hpXL4XnriNwQ/ABnpepYM/1vCLWNDfUNts8dX3xTG6Y=
 github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
+github.com/lestrrat-go/envload v0.0.0-20180220234015-a3eb8ddeffcc h1:RKf14vYWi2ttpEmkA4aQ3j4u9dStX2t4M8UM6qqNsG8=
 github.com/lestrrat-go/envload v0.0.0-20180220234015-a3eb8ddeffcc/go.mod h1:kopuH9ugFRkIXf3YoqHKyrJ9YfUFsckUU9S7B+XP+is=
 github.com/lestrrat-go/file-rotatelogs v2.3.0+incompatible h1:4mNlp+/SvALIPFpbXV3kxNJJno9iKFWGxSDE13Kl66Q=
 github.com/lestrrat-go/file-rotatelogs v2.3.0+incompatible/go.mod h1:ZQnN8lSECaebrkQytbHj4xNgtg8CR7RYXnPok8e0EHA=
@@ -64,13 +76,16 @@ github.com/lestrrat-go/strftime v1.0.1/go.mod h1:E1nN3pCbtMSu1yjSVeyuRFVm/U0xoR7
 github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
 github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 h1:Esafd1046DLDQ0W1YjYsBW+p8U2u7vzgW2SQVmlNazg=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ=
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646/go.mod h1:jpp1/29i3P1S/RLdc7JQKbRpFeM1dOBd8T9ki5s+AY8=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
@@ -80,9 +95,12 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/t-tomalak/logrus-easy-formatter v0.0.0-20190827215021-c074f06c5816 h1:J6v8awz+me+xeb/cUTotKgceAYouhIB3pjzgRd6IlGk=
 github.com/t-tomalak/logrus-easy-formatter v0.0.0-20190827215021-c074f06c5816/go.mod h1:tzym/CEb5jnFI+Q0k4Qq3+LvRF4gO3E2pxS8fHP8jcA=
+github.com/tebeka/strftime v0.1.5 h1:1NQKN1NiQgkqd/2moD6ySP/5CoZQsKa1d3ZhJ44Jpmg=
+github.com/tebeka/strftime v0.1.5/go.mod h1:29/OidkoWHdEKZqzyDLUyC+LmgDgdHo4WAFCDT7D/Ig=
 github.com/tidwall/gjson v1.6.0 h1:9VEQWz6LLMUsUl6PueE49ir4Ka6CzLymOAZDxpFsTDc=
 github.com/tidwall/gjson v1.6.0/go.mod h1:P256ACg0Mn+j1RXIDXoss50DeIABTYK1PULOJHhxOls=
 github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=
@@ -132,6 +150,7 @@ golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
@@ -151,6 +170,7 @@ google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.25.0 h1:Ejskq+SyPohKW+1uil0JJMtmHCgJPJ/qWTxr8qp+R4c=
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/go-playground/assert.v1 v1.2.1/go.mod h1:9RXL0bg/zibRAgZUYszZSwO/z8Y/a8bDuhia5mkpMnE=
 gopkg.in/go-playground/validator.v9 v9.29.1/go.mod h1:+c9/zcJMFNgbLvly1L1V+PpxWdVbfP1avr/N00E2vyQ=

server/websocket.go

@@ -2,24 +2,25 @@ package server
 
 import (
 	"fmt"
-	"github.com/Mrs4s/go-cqhttp/coolq"
-	"github.com/Mrs4s/go-cqhttp/global"
-	"github.com/gorilla/websocket"
-	log "github.com/sirupsen/logrus"
-	"github.com/tidwall/gjson"
 	"net/http"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/Mrs4s/go-cqhttp/coolq"
+	"github.com/Mrs4s/go-cqhttp/global"
+	"github.com/gorilla/websocket"
+	log "github.com/sirupsen/logrus"
+	"github.com/tidwall/gjson"
 )
 
 type websocketServer struct {
-	bot       *coolq.CQBot
-	token     string
-	eventConn []*websocket.Conn
-	pushLock  *sync.Mutex
-	handshake string
+	bot            *coolq.CQBot
+	token          string
+	eventConn      []*websocketConn
+	eventConnMutex sync.Mutex
+	handshake      string
 }
 
 type websocketClient struct {
@@ -27,9 +28,13 @@ type websocketClient struct {
 	token string
 	bot   *coolq.CQBot
 
-	pushLock      *sync.Mutex
-	universalConn *websocket.Conn
-	eventConn     *websocket.Conn
+	universalConn *websocketConn
+	eventConn     *websocketConn
+}
+
+type websocketConn struct {
+	*websocket.Conn
+	sync.Mutex
 }
 
 var WebsocketServer = &websocketServer{}
@@ -41,7 +46,6 @@ var upgrader = websocket.Upgrader{
 
 func (s *websocketServer) Run(addr, authToken string, b *coolq.CQBot) {
 	s.token = authToken
-	s.pushLock = new(sync.Mutex)
 	s.bot = b
 	s.handshake = fmt.Sprintf(`{"_post_method":2,"meta_event_type":"lifecycle","post_type":"meta_event","self_id":%d,"sub_type":"connect","time":%d}`,
 		s.bot.Client.Uin, time.Now().Unix())
@@ -56,7 +60,7 @@ func (s *websocketServer) Run(addr, authToken string, b *coolq.CQBot) {
 }
 
 func NewWebsocketClient(conf *global.GoCQReverseWebsocketConfig, authToken string, b *coolq.CQBot) *websocketClient {
-	return &websocketClient{conf: conf, token: authToken, bot: b, pushLock: new(sync.Mutex)}
+	return &websocketClient{conf: conf, token: authToken, bot: b}
 }
 
 func (c *websocketClient) Run() {
@@ -96,7 +100,8 @@ func (c *websocketClient) connectApi() {
 		return
 	}
 	log.Infof("已连接到反向Websocket API服务器 %v", c.conf.ReverseApiUrl)
-	go c.listenApi(conn, false)
+	wrappedConn := &websocketConn{Conn: conn}
+	go c.listenApi(wrappedConn, false)
 }
 
 func (c *websocketClient) connectEvent() {
@@ -119,7 +124,7 @@ func (c *websocketClient) connectEvent() {
 		return
 	}
 	log.Infof("已连接到反向Websocket Event服务器 %v", c.conf.ReverseEventUrl)
-	c.eventConn = conn
+	c.eventConn = &websocketConn{Conn: conn}
 }
 
 func (c *websocketClient) connectUniversal() {
@@ -141,11 +146,12 @@ func (c *websocketClient) connectUniversal() {
 		}
 		return
 	}
-	go c.listenApi(conn, true)
-	c.universalConn = conn
+	wrappedConn := &websocketConn{Conn: conn}
+	go c.listenApi(wrappedConn, true)
+	c.universalConn = wrappedConn
 }
 
-func (c *websocketClient) listenApi(conn *websocket.Conn, u bool) {
+func (c *websocketClient) listenApi(conn *websocketConn, u bool) {
 	defer conn.Close()
 	for {
 		_, buf, err := conn.ReadMessage()
@@ -153,35 +159,23 @@ func (c *websocketClient) listenApi(conn *websocket.Conn, u bool) {
 			log.Warnf("监听反向WS API时出现错误: %v", err)
 			break
 		}
-		j := gjson.ParseBytes(buf)
-		t := strings.ReplaceAll(j.Get("action").Str, "_async", "")
-		log.Debugf("反向WS接收到API调用: %v 参数: %v", t, j.Get("params").Raw)
-		if f, ok := wsApi[t]; ok {
-			go func() {
-				ret := f(c.bot, j.Get("params"))
-				if j.Get("echo").Exists() {
-					ret["echo"] = j.Get("echo").Value()
-				}
-				c.pushLock.Lock()
-				log.Debugf("准备发送API %v 处理结果: %v", t, ret.ToJson())
-				_ = conn.WriteJSON(ret)
-				c.pushLock.Unlock()
-			}()
-		}
+
+		go conn.handleRequest(c.bot, buf)
+
 	}
 	if c.conf.ReverseReconnectInterval != 0 {
 		time.Sleep(time.Millisecond * time.Duration(c.conf.ReverseReconnectInterval))
 		if !u {
-			c.connectApi()
+			go c.connectApi()
 		}
 	}
 }
 
 func (c *websocketClient) onBotPushEvent(m coolq.MSG) {
-	c.pushLock.Lock()
-	defer c.pushLock.Unlock()
 	if c.eventConn != nil {
 		log.Debugf("向WS服务器 %v 推送Event: %v", c.eventConn.RemoteAddr().String(), m.ToJson())
+		c.eventConn.Lock()
+		defer c.eventConn.Unlock()
 		if err := c.eventConn.WriteJSON(m); err != nil {
 			log.Warnf("向WS服务器 %v 推送Event时出现错误: %v", c.eventConn.RemoteAddr().String(), err)
 			_ = c.eventConn.Close()
@@ -195,6 +189,8 @@ func (c *websocketClient) onBotPushEvent(m coolq.MSG) {
 	}
 	if c.universalConn != nil {
 		log.Debugf("向WS服务器 %v 推送Event: %v", c.universalConn.RemoteAddr().String(), m.ToJson())
+		c.universalConn.Lock()
+		defer c.universalConn.Unlock()
 		if err := c.universalConn.WriteJSON(m); err != nil {
 			log.Warnf("向WS服务器 %v 推送Event时出现错误: %v", c.universalConn.RemoteAddr().String(), err)
 			_ = c.universalConn.Close()
@@ -222,10 +218,19 @@ func (s *websocketServer) event(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	err = c.WriteMessage(websocket.TextMessage, []byte(s.handshake))
-	if err == nil {
-		log.Infof("接受 Websocket 连接: %v (/event)", r.RemoteAddr)
-		s.eventConn = append(s.eventConn, c)
+	if err != nil {
+		log.Warnf("Websocket 握手时出现错误: %v", err)
+		c.Close()
+		return
 	}
+
+	log.Infof("接受 Websocket 连接: %v (/event)", r.RemoteAddr)
+
+	conn := &websocketConn{Conn: c}
+
+	s.eventConnMutex.Lock()
+	s.eventConn = append(s.eventConn, conn)
+	s.eventConnMutex.Unlock()
 }
 
 func (s *websocketServer) api(w http.ResponseWriter, r *http.Request) {
@@ -242,7 +247,8 @@ func (s *websocketServer) api(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	log.Infof("接受 Websocket 连接: %v (/api)", r.RemoteAddr)
-	go s.listenApi(c)
+	conn := &websocketConn{Conn: c}
+	go s.listenApi(conn)
 }
 
 func (s *websocketServer) any(w http.ResponseWriter, r *http.Request) {
@@ -259,42 +265,55 @@ func (s *websocketServer) any(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	err = c.WriteMessage(websocket.TextMessage, []byte(s.handshake))
-	if err == nil {
-		log.Infof("接受 Websocket 连接: %v (/)", r.RemoteAddr)
-		s.eventConn = append(s.eventConn, c)
-		s.listenApi(c)
+	if err != nil {
+		log.Warnf("Websocket 握手时出现错误: %v", err)
+		c.Close()
+		return
 	}
+
+	log.Infof("接受 Websocket 连接: %v (/)", r.RemoteAddr)
+	conn := &websocketConn{Conn: c}
+	s.eventConn = append(s.eventConn, conn)
+	s.listenApi(conn)
 }
 
-func (s *websocketServer) listenApi(c *websocket.Conn) {
+func (s *websocketServer) listenApi(c *websocketConn) {
 	defer c.Close()
 	for {
 		t, payload, err := c.ReadMessage()
 		if err != nil {
 			break
 		}
+
 		if t == websocket.TextMessage {
-			j := gjson.ParseBytes(payload)
-			t := strings.ReplaceAll(j.Get("action").Str, "_async", "") //TODO: async support
-			log.Debugf("WS接收到API调用: %v 参数: %v", t, j.Get("params").Raw)
-			if f, ok := wsApi[t]; ok {
-				go func() {
-					ret := f(s.bot, j.Get("params"))
-					if j.Get("echo").Exists() {
-						ret["echo"] = j.Get("echo").Value()
-					}
-					s.pushLock.Lock()
-					_ = c.WriteJSON(ret)
-					s.pushLock.Unlock()
-				}()
-			}
+			go c.handleRequest(s.bot, payload)
+		}
+	}
+}
+
+func (c *websocketConn) handleRequest(bot *coolq.CQBot, payload []byte) {
+	defer func() {
+		if err := recover(); err != nil {
+			log.Printf("处置WS命令时发生无法恢复的异常：%v", err)
+			c.Close()
+		}
+	}()
+
+	j := gjson.ParseBytes(payload)
+	t := strings.ReplaceAll(j.Get("action").Str, "_async", "")
+	log.Debugf("WS接收到API调用: %v 参数: %v", t, j.Get("params").Raw)
+	if f, ok := wsApi[t]; ok {
+		ret := f(bot, j.Get("params"))
+		if j.Get("echo").Exists() {
+			ret["echo"] = j.Get("echo").Value()
 		}
+		c.Lock()
+		defer c.Unlock()
+		_ = c.WriteJSON(ret)
 	}
 }
 
 func (s *websocketServer) onBotPushEvent(m coolq.MSG) {
-	s.pushLock.Lock()
-	defer s.pushLock.Unlock()
 	pos := 0
 	for _, conn := range s.eventConn {
 		log.Debugf("向WS客户端 %v 推送Event: %v", conn.RemoteAddr().String(), m.ToJson())