feat 包命名规范
This commit is contained in:
113
internal/handler/wshandler/client/client.go
Normal file
113
internal/handler/wshandler/client/client.go
Normal file
@@ -0,0 +1,113 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.hlsq.asia/mmorpg/service-common/db/redis"
|
||||
"git.hlsq.asia/mmorpg/service-common/log"
|
||||
"git.hlsq.asia/mmorpg/service-common/net/socket"
|
||||
"git.hlsq.asia/mmorpg/service-gateway/internal/global"
|
||||
"go.uber.org/zap"
|
||||
"runtime/debug"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Client struct {
|
||||
sync.WaitGroup
|
||||
conn socket.ISocketConn // Socket
|
||||
mailChan chan Event // 邮箱队列
|
||||
logger *zap.SugaredLogger // 日志
|
||||
ctx context.Context // 上下文
|
||||
cancel context.CancelFunc // 取消上下文
|
||||
heartBeat time.Time // 最后一次心跳
|
||||
|
||||
Status int32 // 状态：0 登陆中 1 正常 2 离线
|
||||
USN int64 // 用户ID
|
||||
SceneSID int64 // 场景服ID
|
||||
InstanceID int32 // 副本ID，副本类型
|
||||
UniqueNo int64 // 副本唯一编号
|
||||
}
|
||||
|
||||
func NewClient(conn socket.ISocketConn) *Client {
|
||||
client := &Client{
|
||||
conn: conn,
|
||||
logger: log.GetLogger(),
|
||||
heartBeat: time.Now(),
|
||||
mailChan: make(chan Event, 1024),
|
||||
}
|
||||
client.ctx, client.cancel = context.WithCancel(context.Background())
|
||||
client.Add(1)
|
||||
go client.Loop()
|
||||
return client
|
||||
}
|
||||
|
||||
func (c *Client) SetUSN(usn int64) {
|
||||
c.USN = usn
|
||||
c.logger = log.GetLogger().Named(fmt.Sprintf("usn:%v", usn))
|
||||
}
|
||||
|
||||
func (c *Client) Loop() {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
c.logger.Errorf("Client Loop err: %v", err)
|
||||
debug.PrintStack()
|
||||
}
|
||||
}()
|
||||
defer c.onClose()
|
||||
t := time.NewTicker(20 * time.Second)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-c.ctx.Done():
|
||||
return
|
||||
case evt, ok := <-c.mailChan:
|
||||
if ok {
|
||||
c.handle(evt)
|
||||
}
|
||||
case <-t.C:
|
||||
_ = c.conn.Ping()
|
||||
if time.Now().Sub(c.heartBeat) > 60*time.Second {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) OnEvent(event Event) {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
c.logger.Warnf(fmt.Sprintf("send event chan error: %v", err))
|
||||
}
|
||||
}()
|
||||
select {
|
||||
case c.mailChan <- event:
|
||||
default:
|
||||
c.logger.Warnf("Client mailChan full")
|
||||
}
|
||||
}
|
||||
|
||||
// CloseClient 关闭客户端（同步，会等待onClose执行完成）
|
||||
func (c *Client) CloseClient() {
|
||||
if c.cancel != nil {
|
||||
c.cancel()
|
||||
c.Wait()
|
||||
}
|
||||
}
|
||||
|
||||
// onClose 客户端关闭自动触发
|
||||
func (c *Client) onClose() {
|
||||
if c.conn != nil {
|
||||
_ = c.conn.Close()
|
||||
c.conn = nil
|
||||
}
|
||||
if c.mailChan != nil {
|
||||
close(c.mailChan)
|
||||
c.mailChan = nil
|
||||
}
|
||||
c.Status = 2
|
||||
UserMgr.Delete(c.USN)
|
||||
redis.GetClient().HDel(c.ctx, fmt.Sprintf(global.KeyGatewayInfo, c.USN), global.HFieldInfoGatewaySID)
|
||||
c.onLeave()
|
||||
c.Done()
|
||||
}
|
||||
17
internal/handler/wshandler/client/client_event.go
Normal file
17
internal/handler/wshandler/client/client_event.go
Normal file
@@ -0,0 +1,17 @@
|
||||
package client
|
||||
|
||||
type Event interface {
|
||||
}
|
||||
|
||||
type ClientEvent struct {
|
||||
Event
|
||||
Msg []byte
|
||||
}
|
||||
|
||||
type PongEvent struct {
|
||||
Event
|
||||
}
|
||||
|
||||
type SystemLoginSuccessEvent struct {
|
||||
Event
|
||||
}
|
||||
111
internal/handler/wshandler/client/client_handler.go
Normal file
111
internal/handler/wshandler/client/client_handler.go
Normal file
@@ -0,0 +1,111 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.hlsq.asia/mmorpg/service-common/db/redis"
|
||||
"git.hlsq.asia/mmorpg/service-common/net/grpc/grpcclient"
|
||||
"git.hlsq.asia/mmorpg/service-common/proto/rs/rspb"
|
||||
"git.hlsq.asia/mmorpg/service-common/proto/ss/sspb"
|
||||
"git.hlsq.asia/mmorpg/service-gateway/internal/global"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"time"
|
||||
)
|
||||
|
||||
func (c *Client) handle(event Event) {
|
||||
switch e := event.(type) {
|
||||
case *ClientEvent:
|
||||
msg := &sspb.Message{}
|
||||
if err := proto.Unmarshal(e.Msg, msg); err != nil {
|
||||
c.logger.Errorf("handle event proto.Unmarshal err: %v", err)
|
||||
c.cancel()
|
||||
return
|
||||
}
|
||||
switch msg.ID {
|
||||
case sspb.MessageID_MESSAGE_ID_ENTER_INSTANCE:
|
||||
m := &sspb.C2S_EnterInstance{}
|
||||
if err := proto.Unmarshal(msg.Payload, m); err != nil {
|
||||
c.logger.Errorf("handle event proto.Unmarshal err: %v", err)
|
||||
c.cancel()
|
||||
return
|
||||
}
|
||||
c.onEnter(m)
|
||||
case sspb.MessageID_MESSAGE_ID_ACTION:
|
||||
m := &sspb.C2S_Action{}
|
||||
if err := proto.Unmarshal(msg.Payload, m); err != nil {
|
||||
c.logger.Errorf("handle event proto.Unmarshal err: %v", err)
|
||||
c.cancel()
|
||||
return
|
||||
}
|
||||
c.onAction(m)
|
||||
}
|
||||
case *PongEvent:
|
||||
c.heartBeat = time.Now()
|
||||
case *SystemLoginSuccessEvent:
|
||||
if c.Status == 0 {
|
||||
c.Status = 1
|
||||
UserMgr.Add(c.USN, c)
|
||||
redis.GetClient().HSet(c.ctx, fmt.Sprintf(global.KeyGatewayInfo, c.USN), global.HFieldInfoGatewaySID, global.GatewaySID)
|
||||
c.WriteMessage(sspb.MessageID_MESSAGE_ID_LOGIN_SUCCESS, &sspb.S2C_LoginSuccess{
|
||||
InstanceID: 1,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) onEnter(msg *sspb.C2S_EnterInstance) {
|
||||
client, err := grpcclient.SceneNewClientLB()
|
||||
if err != nil {
|
||||
c.logger.Errorf("SceneNewClient err: %v", err)
|
||||
return
|
||||
}
|
||||
resp, err := client.Enter(c.ctx, &rspb.EnterReq{
|
||||
USN: c.USN,
|
||||
GatewaySID: global.GatewaySID,
|
||||
InstanceID: msg.InstanceID,
|
||||
})
|
||||
if err != nil {
|
||||
c.logger.Errorf("enter err: %v", err)
|
||||
return
|
||||
}
|
||||
c.SceneSID = resp.SceneSID
|
||||
c.UniqueNo = resp.UniqueNo
|
||||
c.InstanceID = msg.InstanceID
|
||||
c.WriteBytes(sspb.MessageID(resp.MessageID), resp.Payload)
|
||||
}
|
||||
|
||||
func (c *Client) onLeave() {
|
||||
if c.SceneSID == 0 {
|
||||
return
|
||||
}
|
||||
client, err := grpcclient.SceneNewClient(c.SceneSID)
|
||||
if err != nil {
|
||||
c.logger.Errorf("SceneNewClient err: %v", err)
|
||||
return
|
||||
}
|
||||
_, err = client.Leave(context.Background(), &rspb.LeaveReq{
|
||||
USN: c.USN,
|
||||
UniqueNo: c.UniqueNo,
|
||||
})
|
||||
if err != nil {
|
||||
c.logger.Errorf("leave err: %v", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) onAction(msg *sspb.C2S_Action) {
|
||||
if c.SceneSID == 0 {
|
||||
return
|
||||
}
|
||||
if err := grpcclient.SendMessageToScene(c.SceneSID, grpcclient.FunAction, &rspb.ActionReq{
|
||||
UniqueNo: c.UniqueNo,
|
||||
USN: c.USN,
|
||||
Action: int32(msg.Action),
|
||||
DirX: msg.DirX,
|
||||
DirY: msg.DirY,
|
||||
SkillID: msg.SkillID,
|
||||
}); err != nil {
|
||||
c.logger.Errorf("send action err: %v", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
57
internal/handler/wshandler/client/client_write.go
Normal file
57
internal/handler/wshandler/client/client_write.go
Normal file
@@ -0,0 +1,57 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"git.hlsq.asia/mmorpg/service-common/proto/ss/sspb"
|
||||
"google.golang.org/protobuf/proto"
|
||||
)
|
||||
|
||||
// WriteMessage 向客户端发送消息
|
||||
func (c *Client) WriteMessage(id sspb.MessageID, data proto.Message) {
|
||||
if c.conn == nil || c.conn.IsClose() {
|
||||
return
|
||||
}
|
||||
d, err := proto.Marshal(data)
|
||||
if err != nil {
|
||||
c.logger.Errorf("WriteMessage proto.Marshal err: %v", err)
|
||||
return
|
||||
}
|
||||
m, err := proto.Marshal(&sspb.Message{
|
||||
ID: id,
|
||||
Payload: d,
|
||||
})
|
||||
if err != nil {
|
||||
c.logger.Errorf("WriteMessage proto.Marshal err: %v", err)
|
||||
return
|
||||
}
|
||||
if err = c.conn.Write(m); err != nil {
|
||||
c.logger.Errorf("WriteMessage err: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// WriteBytes 向客户端发送字节数据
|
||||
func (c *Client) WriteBytes(id sspb.MessageID, data []byte) {
|
||||
if c.conn == nil || c.conn.IsClose() {
|
||||
return
|
||||
}
|
||||
m, err := proto.Marshal(&sspb.Message{
|
||||
ID: id,
|
||||
Payload: data,
|
||||
})
|
||||
if err != nil {
|
||||
c.logger.Errorf("WriteBytes proto.Marshal err: %v", err)
|
||||
return
|
||||
}
|
||||
if err = c.conn.Write(m); err != nil {
|
||||
c.logger.Errorf("WriteBytes err: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// WriteBytesPreMarshal 向客户端发送字节数据（需要预先打包，适合广播相同数据）
|
||||
func (c *Client) WriteBytesPreMarshal(data []byte) {
|
||||
if c.conn == nil || c.conn.IsClose() {
|
||||
return
|
||||
}
|
||||
if err := c.conn.Write(data); err != nil {
|
||||
c.logger.Errorf("WriteBytes err: %v", err)
|
||||
}
|
||||
}
|
||||
67
internal/handler/wshandler/client/manager.go
Normal file
67
internal/handler/wshandler/client/manager.go
Normal file
@@ -0,0 +1,67 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"git.hlsq.asia/mmorpg/service-gateway/internal/global"
|
||||
"sync"
|
||||
)
|
||||
|
||||
var UserMgr *userManager
|
||||
|
||||
type userManager struct {
|
||||
userMap map[int64]*Client
|
||||
sync.RWMutex
|
||||
}
|
||||
|
||||
func init() {
|
||||
UserMgr = &userManager{
|
||||
userMap: make(map[int64]*Client),
|
||||
}
|
||||
}
|
||||
|
||||
func (m *userManager) Add(usn int64, client *Client) {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
m.userMap[usn] = client
|
||||
global.OnlineUsersGauge.Inc()
|
||||
}
|
||||
|
||||
func (m *userManager) Delete(usn int64) {
|
||||
m.Lock()
|
||||
defer m.Unlock()
|
||||
delete(m.userMap, usn)
|
||||
global.OnlineUsersGauge.Dec()
|
||||
}
|
||||
|
||||
func (m *userManager) GetAll() map[int64]*Client {
|
||||
m.RLock()
|
||||
defer m.RUnlock()
|
||||
|
||||
copyMap := make(map[int64]*Client, len(m.userMap))
|
||||
for k, v := range m.userMap {
|
||||
copyMap[k] = v
|
||||
}
|
||||
return copyMap
|
||||
}
|
||||
|
||||
func (m *userManager) GetAllInterface() []interface{} {
|
||||
m.RLock()
|
||||
defer m.RUnlock()
|
||||
|
||||
r := make([]interface{}, 0)
|
||||
for _, v := range m.userMap {
|
||||
r = append(r, v)
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
func (m *userManager) GetByUSN(usn int64) *Client {
|
||||
m.RLock()
|
||||
defer m.RUnlock()
|
||||
return m.userMap[usn]
|
||||
}
|
||||
|
||||
func (m *userManager) GetSize() int32 {
|
||||
m.RLock()
|
||||
defer m.RUnlock()
|
||||
return int32(len(m.userMap))
|
||||
}
|
||||
184
internal/handler/wshandler/login/login.go
Normal file
184
internal/handler/wshandler/login/login.go
Normal file
@@ -0,0 +1,184 @@
|
||||
package login
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.hlsq.asia/mmorpg/service-common/db/redis"
|
||||
"git.hlsq.asia/mmorpg/service-common/log"
|
||||
"git.hlsq.asia/mmorpg/service-common/net/grpc/grpcclient"
|
||||
"git.hlsq.asia/mmorpg/service-common/proto/rs/rspb"
|
||||
"git.hlsq.asia/mmorpg/service-common/proto/ss/sspb"
|
||||
"git.hlsq.asia/mmorpg/service-common/utils"
|
||||
"git.hlsq.asia/mmorpg/service-gateway/internal/global"
|
||||
"git.hlsq.asia/mmorpg/service-gateway/internal/handler/wshandler/client"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
var loginQueue *Login
|
||||
|
||||
// Login 登录队列结构
|
||||
type Login struct {
|
||||
queue chan *User // 用户队列
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
type User struct {
|
||||
Cli *client.Client
|
||||
Token string
|
||||
}
|
||||
|
||||
func NewLoginQueue(maxSize int) *Login {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
loginQueue = &Login{
|
||||
queue: make(chan *User, maxSize),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
return loginQueue
|
||||
}
|
||||
|
||||
func GetLoginQueue() *Login {
|
||||
return loginQueue
|
||||
}
|
||||
|
||||
// AddToLoginQueue 添加到登录队列
|
||||
func (l *Login) AddToLoginQueue(user *User) bool {
|
||||
select {
|
||||
case l.queue <- user:
|
||||
return true
|
||||
default:
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (l *Login) Start(num int) {
|
||||
for i := 0; i < num; i++ {
|
||||
l.wg.Add(1)
|
||||
go func() {
|
||||
defer l.wg.Done()
|
||||
for {
|
||||
select {
|
||||
case <-l.ctx.Done():
|
||||
return
|
||||
case user, ok := <-l.queue:
|
||||
if ok {
|
||||
l.StartLogin(user)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
// 定时从排队的队列中取用户
|
||||
l.wg.Add(1)
|
||||
go func() {
|
||||
defer l.wg.Done()
|
||||
tick := time.NewTicker(1 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case <-tick.C:
|
||||
for {
|
||||
if client.UserMgr.GetSize() < global.MaxOnlineSize {
|
||||
cli, err := GetQueueUp().Dequeue()
|
||||
if err != nil {
|
||||
log.Errorf("dequeue err: %v", err)
|
||||
return
|
||||
}
|
||||
if cli == nil {
|
||||
break
|
||||
}
|
||||
l.LoginSuccess(cli)
|
||||
} else {
|
||||
break
|
||||
}
|
||||
}
|
||||
case <-l.ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (l *Login) Stop() {
|
||||
l.cancel()
|
||||
l.wg.Wait()
|
||||
}
|
||||
|
||||
// StartLogin 开始登录流程
|
||||
func (l *Login) StartLogin(user *User) {
|
||||
if !l.CheckToken(user) {
|
||||
user.Cli.WriteMessage(sspb.MessageID_MESSAGE_ID_KICK_OUT, &sspb.S2C_KickOut{
|
||||
ID: sspb.KickOutID_KICK_OUT_ID_TOKEN_INVALID,
|
||||
})
|
||||
user.Cli.CloseClient()
|
||||
return
|
||||
}
|
||||
if gatewaySID := l.CheckOnline(user); len(gatewaySID) > 0 {
|
||||
// 如果在线就要踢，如果踢失败了就返回服务器繁忙，一般不应该走到这里
|
||||
if !l.KickUser(user.Cli.SceneSID, user.Cli.USN) {
|
||||
user.Cli.WriteMessage(sspb.MessageID_MESSAGE_ID_KICK_OUT, &sspb.S2C_KickOut{
|
||||
ID: sspb.KickOutID_KICK_OUT_ID_SERVER_BUSY,
|
||||
})
|
||||
user.Cli.CloseClient()
|
||||
return
|
||||
}
|
||||
}
|
||||
if client.UserMgr.GetSize() >= global.MaxOnlineSize {
|
||||
// 如果人数满了就排队
|
||||
if err := GetQueueUp().Enqueue(user.Cli); err != nil {
|
||||
user.Cli.WriteMessage(sspb.MessageID_MESSAGE_ID_KICK_OUT, &sspb.S2C_KickOut{
|
||||
ID: sspb.KickOutID_KICK_OUT_ID_QUEUE_UP_FULL,
|
||||
})
|
||||
user.Cli.CloseClient()
|
||||
return
|
||||
}
|
||||
// 告诉客户端正在排队
|
||||
position, ok := GetQueueUp().GetPosition(user.Cli.USN)
|
||||
if !ok {
|
||||
user.Cli.WriteMessage(sspb.MessageID_MESSAGE_ID_KICK_OUT, &sspb.S2C_KickOut{
|
||||
ID: sspb.KickOutID_KICK_OUT_ID_SERVER_BUSY,
|
||||
})
|
||||
user.Cli.CloseClient()
|
||||
return
|
||||
}
|
||||
user.Cli.WriteMessage(sspb.MessageID_MESSAGE_ID_QUEUE_UP, &sspb.S2C_QueueUp{
|
||||
QueueUpCount: int32(position),
|
||||
})
|
||||
} else {
|
||||
l.LoginSuccess(user.Cli)
|
||||
}
|
||||
}
|
||||
|
||||
// CheckToken 校验Token是否有效
|
||||
func (l *Login) CheckToken(user *User) bool {
|
||||
usn, _ := redis.GetClient().HGet(context.Background(), global.KeyGatewayAccessToken+user.Token, (&utils.UserSession{}).GetUsnKey()).Int64()
|
||||
user.Cli.SetUSN(usn)
|
||||
return usn > 0
|
||||
}
|
||||
|
||||
// CheckOnline 校验是否在线
|
||||
func (l *Login) CheckOnline(user *User) string {
|
||||
return redis.GetClient().HGet(l.ctx, fmt.Sprintf(global.KeyGatewayInfo, user.Cli.USN), global.HFieldInfoGatewaySID).Val()
|
||||
}
|
||||
|
||||
// KickUser 把玩家踢下线
|
||||
func (l *Login) KickUser(gatewaySID int64, usn int64) bool {
|
||||
gc, err := grpcclient.GatewayNewClient(gatewaySID)
|
||||
if err != nil {
|
||||
log.Errorf("KickUser cannot find gateway client: %v, sid: %v", err, gatewaySID)
|
||||
return false
|
||||
}
|
||||
_, err = gc.KickUser(l.ctx, &rspb.KickUserReq{USN: usn})
|
||||
if err != nil {
|
||||
log.Errorf("KickUser err: %v, sid: %v, usn: %v", err, gatewaySID, usn)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// LoginSuccess 登录成功
|
||||
func (l *Login) LoginSuccess(user *client.Client) {
|
||||
user.OnEvent(&client.SystemLoginSuccessEvent{})
|
||||
}
|
||||
104
internal/handler/wshandler/login/queue_up.go
Normal file
104
internal/handler/wshandler/login/queue_up.go
Normal file
@@ -0,0 +1,104 @@
|
||||
package login
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"git.hlsq.asia/mmorpg/service-gateway/internal/handler/wshandler/client"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
var queueUp *QueueUp
|
||||
|
||||
// QueueUp 排队队列结构
|
||||
type QueueUp struct {
|
||||
queue chan *QueueUser // 用户队列
|
||||
waiting sync.Map // map[usn]*QueueUser
|
||||
minTicket atomic.Int64 // 最小的Ticket
|
||||
maxTicket atomic.Int64 // 最大的Ticket
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
type QueueUser struct {
|
||||
Cli *client.Client
|
||||
Ticket int64
|
||||
}
|
||||
|
||||
func NewQueueUp(maxSize int) *QueueUp {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
queueUp = &QueueUp{
|
||||
queue: make(chan *QueueUser, maxSize),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
return queueUp
|
||||
}
|
||||
|
||||
func GetQueueUp() *QueueUp {
|
||||
return queueUp
|
||||
}
|
||||
|
||||
// Enqueue 将用户加入排队队列
|
||||
func (q *QueueUp) Enqueue(cli *client.Client) error {
|
||||
select {
|
||||
case <-q.ctx.Done():
|
||||
return errors.New("queue stopped")
|
||||
default:
|
||||
}
|
||||
|
||||
ticket := q.maxTicket.Add(1)
|
||||
item := &QueueUser{Cli: cli, Ticket: ticket}
|
||||
|
||||
select {
|
||||
case q.queue <- item:
|
||||
q.waiting.Store(cli.USN, item)
|
||||
return nil
|
||||
default:
|
||||
return errors.New("queue is full")
|
||||
}
|
||||
}
|
||||
|
||||
// Dequeue 从排队队列中取出下一个有效用户
|
||||
func (q *QueueUp) Dequeue() (*client.Client, error) {
|
||||
select {
|
||||
case item, ok := <-q.queue:
|
||||
if ok {
|
||||
q.minTicket.Store(item.Ticket)
|
||||
if _, loaded := q.waiting.LoadAndDelete(item.Cli.USN); loaded {
|
||||
return item.Cli, nil
|
||||
}
|
||||
return q.Dequeue()
|
||||
}
|
||||
case <-q.ctx.Done():
|
||||
return nil, q.ctx.Err()
|
||||
default:
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// GetPosition 返回用户前面还有多少人在排队
|
||||
func (q *QueueUp) GetPosition(usn int64) (int64, bool) {
|
||||
val, ok := q.waiting.Load(usn)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
user := val.(*QueueUser)
|
||||
return user.Ticket - q.minTicket.Load() - 1, true
|
||||
}
|
||||
|
||||
// RemoveUser 安全移除用户（标记为取消）
|
||||
func (q *QueueUp) RemoveUser(usn int64) bool {
|
||||
_, loaded := q.waiting.LoadAndDelete(usn)
|
||||
return loaded
|
||||
}
|
||||
|
||||
// GetQueueSize 获取当前排队人数
|
||||
func (q *QueueUp) GetQueueSize() int {
|
||||
return len(q.queue)
|
||||
}
|
||||
|
||||
// Stop 停止整个队列服务
|
||||
func (q *QueueUp) Stop() {
|
||||
q.cancel()
|
||||
}
|
||||
Reference in New Issue
Block a user