chenyh
/
vRp.CD2g_test
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: 移除分段锁map实现及 hash.Map、hash.ReadonlyMap 接口，移除 asynchronous 包，同步包更名为 concurrent

This commit is contained in:
kercylan 2023-07-21 23:32:46 +08:00
parent 2cbffbf967
commit d0d2087fee
30 changed files with 363 additions and 922 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
component/components/lockstep.go
View File

@ -3,7 +3,7 @@ package components
import (
"encoding/json"
"github.com/kercylan98/minotaur/component"
"github.com/kercylan98/minotaur/utils/synchronization"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/timer"
"sync"
"sync/atomic"
@ -13,8 +13,8 @@ import (
// NewLockstep 创建一个锁步（帧）同步默认实现的组件(Lockstep)进行返回
func NewLockstep[ClientID comparable, Command any](options ...LockstepOption[ClientID, Command]) *Lockstep[ClientID, Command] {
lockstep := &Lockstep[ClientID, Command]{
clients: synchronization.NewMap[ClientID, component.LockstepClient[ClientID]](),
frames: synchronization.NewMap[int, []Command](),
clients: concurrent.NewBalanceMap[ClientID, component.LockstepClient[ClientID]](),
frames: concurrent.NewBalanceMap[int, []Command](),
ticker: timer.GetTicker(10),
frameRate: 15,
serialization: func(frame int, commands []Command) []byte {
@ -25,7 +25,7 @@ func NewLockstep[ClientID comparable, Command any](options ...LockstepOption[Cli
data, _ := json.Marshal(frameStruct)
return data
},
clientCurrentFrame: synchronization.NewMap[ClientID, int](),
clientCurrentFrame: concurrent.NewBalanceMap[ClientID, int](),
}
for _, option := range options {
option(lockstep)
@ -40,12 +40,12 @@ func NewLockstep[ClientID comparable, Command any](options ...LockstepOption[Cli
// - 从特定帧开始追帧
// - 兼容各种基于TCP/UDP/Unix的网络类型，可通过客户端实现其他网络类型同步
type Lockstep[ClientID comparable, Command any] struct {
clients *synchronization.Map[ClientID, component.LockstepClient[ClientID]] // 接受广播的客户端
frames *synchronization.Map[int, []Command] // 所有帧指令
clients *concurrent.BalanceMap[ClientID, component.LockstepClient[ClientID]] // 接受广播的客户端
frames *concurrent.BalanceMap[int, []Command] // 所有帧指令
ticker *timer.Ticker // 定时器
frameMutex sync.Mutex // 帧锁
currentFrame int // 当前帧
clientCurrentFrame *synchronization.Map[ClientID, int] // 客户端当前帧数
clientCurrentFrame *concurrent.BalanceMap[ClientID, int] // 客户端当前帧数
running atomic.Bool
frameRate int // 帧率（每秒N帧）
@ -123,8 +123,8 @@ func (slf *Lockstep[ClientID, Command]) StopBroadcast() {
// AddCommand 添加命令到当前帧
func (slf *Lockstep[ClientID, Command]) AddCommand(command Command) {
slf.frames.AtomGetSet(slf.currentFrame, func(value []Command, exist bool) (newValue []Command, isSet bool) {
return append(value, command), true
slf.frames.Atom(func(m map[int][]Command) {
m[slf.currentFrame] = append(m[slf.currentFrame], command)
})
}

6
game/builtin/attrs.go
View File

@ -2,18 +2,18 @@ package builtin
import (
"github.com/kercylan98/minotaur/game"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/huge"
"github.com/kercylan98/minotaur/utils/synchronization"
)
func NewAttrs() *Attrs {
return &Attrs{
attrs: synchronization.NewMap[int, any](),
attrs: concurrent.NewBalanceMap[int, any](),
}
}
type Attrs struct {
attrs *synchronization.Map[int, any]
attrs *concurrent.BalanceMap[int, any]
attrChangeEventHandles []game.AttrChangeEventHandle
attrIdChangeEventHandles map[int][]game.AttrChangeEventHandle

12
game/builtin/ranking_list.go
View File

@ -3,15 +3,15 @@ package builtin
import (
"encoding/json"
"github.com/kercylan98/minotaur/game"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/generic"
"github.com/kercylan98/minotaur/utils/synchronization"
)
// NewRankingList 创建一个排名从0开始的排行榜
func NewRankingList[CompetitorID comparable, Score generic.Ordered](options ...RankingListOption[CompetitorID, Score]) *RankingList[CompetitorID, Score] {
rankingList := &RankingList[CompetitorID, Score]{
rankCount: 100,
competitors: synchronization.NewMap[CompetitorID, Score](),
competitors: concurrent.NewBalanceMap[CompetitorID, Score](),
}
for _, option := range options {
option(rankingList)
@ -22,7 +22,7 @@ func NewRankingList[CompetitorID comparable, Score generic.Ordered](options ...R
type RankingList[CompetitorID comparable, Score generic.Ordered] struct {
asc bool
rankCount int
competitors *synchronization.Map[CompetitorID, Score]
competitors *concurrent.BalanceMap[CompetitorID, Score]
scores []*scoreItem[CompetitorID, Score] // CompetitorID, Score
rankChangeEventHandles []game.RankChangeEventHandle[CompetitorID, Score]
@ -244,11 +244,11 @@ func (slf *RankingList[CompetitorID, Score]) competitor(competitorId CompetitorI
func (slf *RankingList[CompetitorID, Score]) UnmarshalJSON(bytes []byte) error {
var t struct {
Competitors *synchronization.Map[CompetitorID, Score] `json:"competitors,omitempty"`
Competitors *concurrent.BalanceMap[CompetitorID, Score] `json:"competitors,omitempty"`
Scores []*scoreItem[CompetitorID, Score] `json:"scores,omitempty"`
Asc bool `json:"asc,omitempty"`
}
t.Competitors = synchronization.NewMap[CompetitorID, Score]()
t.Competitors = concurrent.NewBalanceMap[CompetitorID, Score]()
if err := json.Unmarshal(bytes, &t); err != nil {
return err
}
@ -260,7 +260,7 @@ func (slf *RankingList[CompetitorID, Score]) UnmarshalJSON(bytes []byte) error {
func (slf *RankingList[CompetitorID, Score]) MarshalJSON() ([]byte, error) {
var t struct {
Competitors *synchronization.Map[CompetitorID, Score] `json:"competitors,omitempty"`
Competitors *concurrent.BalanceMap[CompetitorID, Score] `json:"competitors,omitempty"`
Scores []*scoreItem[CompetitorID, Score] `json:"scores,omitempty"`
Asc bool `json:"asc,omitempty"`
}

11
game/builtin/room.go
View File

@ -2,8 +2,7 @@ package builtin
import (
"github.com/kercylan98/minotaur/game"
"github.com/kercylan98/minotaur/utils/asynchronous"
"github.com/kercylan98/minotaur/utils/hash"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/log"
)
@ -11,7 +10,7 @@ import (
func NewRoom[PlayerID comparable, Player game.Player[PlayerID]](guid int64, options ...RoomOption[PlayerID, Player]) *Room[PlayerID, Player] {
room := &Room[PlayerID, Player]{
guid: guid,
players: asynchronous.NewMap[PlayerID, Player](),
players: concurrent.NewBalanceMap[PlayerID, Player](),
}
for _, option := range options {
option(room)
@ -27,7 +26,7 @@ type Room[PlayerID comparable, Player game.Player[PlayerID]] struct {
owner PlayerID
noMaster bool
playerLimit int
players hash.Map[PlayerID, Player]
players *concurrent.BalanceMap[PlayerID, Player]
kickCheckHandle func(room *Room[PlayerID, Player], id, target PlayerID) error
playerJoinRoomEventHandles []game.PlayerJoinRoomEventHandle[PlayerID, Player]
@ -51,8 +50,8 @@ func (slf *Room[PlayerID, Player]) GetPlayer(id PlayerID) Player {
}
// GetPlayers 获取所有玩家
func (slf *Room[PlayerID, Player]) GetPlayers() hash.MapReadonly[PlayerID, Player] {
return slf.players.(hash.MapReadonly[PlayerID, Player])
func (slf *Room[PlayerID, Player]) GetPlayers() map[PlayerID]Player {
return slf.players.Map()
}
// GetPlayerCount 获取玩家数量

6
game/builtin/room_manager.go
View File

@ -2,20 +2,20 @@ package builtin
import (
"github.com/kercylan98/minotaur/game"
"github.com/kercylan98/minotaur/utils/synchronization"
"github.com/kercylan98/minotaur/utils/concurrent"
"sync/atomic"
)
func NewRoomManager[PlayerID comparable, Room game.Room[PlayerID, game.Player[PlayerID]]]() *RoomManager[PlayerID, Room] {
return &RoomManager[PlayerID, Room]{
rooms: synchronization.NewMap[int64, Room](),
rooms: concurrent.NewBalanceMap[int64, Room](),
}
}
// RoomManager 房间管理器
type RoomManager[PlayerID comparable, Room game.Room[PlayerID, game.Player[PlayerID]]] struct {
guid atomic.Int64
rooms *synchronization.Map[int64, Room]
rooms *concurrent.BalanceMap[int64, Room]
}
// GenGuid 生成一个新的房间guid

4
game/builtin/room_options.go
View File

@ -2,7 +2,7 @@ package builtin
import (
"github.com/kercylan98/minotaur/game"
"github.com/kercylan98/minotaur/utils/synchronization"
"github.com/kercylan98/minotaur/utils/concurrent"
)
// RoomOption 房间构建可选项
@ -11,7 +11,7 @@ type RoomOption[PlayerID comparable, Player game.Player[PlayerID]] func(room *Ro
// WithRoomSync 通过线程安全的方式创建房间
func WithRoomSync[PlayerID comparable, Player game.Player[PlayerID]]() RoomOption[PlayerID, Player] {
return func(room *Room[PlayerID, Player]) {
room.players = synchronization.NewMap[PlayerID, Player]()
room.players = concurrent.NewBalanceMap[PlayerID, Player]()
}
}

41
game/builtin/room_seat.go
View File

@ -2,7 +2,7 @@ package builtin
import (
"github.com/kercylan98/minotaur/game"
"github.com/kercylan98/minotaur/utils/asynchronous"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/hash"
"github.com/kercylan98/minotaur/utils/slice"
"sync"
@ -12,7 +12,7 @@ import (
func NewRoomSeat[PlayerID comparable, Player game.Player[PlayerID]](room game.Room[PlayerID, Player], options ...RoomSeatOption[PlayerID, Player]) *RoomSeat[PlayerID, Player] {
roomSeat := &RoomSeat[PlayerID, Player]{
Room: room,
seatPS: asynchronous.NewMap[PlayerID, int](),
seatPS: concurrent.NewBalanceMap[PlayerID, int](),
}
for _, option := range options {
option(roomSeat)
@ -26,7 +26,7 @@ type RoomSeat[PlayerID comparable, Player game.Player[PlayerID]] struct {
game.Room[PlayerID, Player]
mutex sync.RWMutex
vacancy []int
seatPS hash.Map[PlayerID, int]
seatPS *concurrent.BalanceMap[PlayerID, int]
seatSP []*PlayerID
duplicateLock bool
fillIn bool
@ -240,6 +240,41 @@ func (slf *RoomSeat[PlayerID, Player]) GetNextSeatVacancy(seat int) int {
return seat
}
// GetPrevSeat 获取特定座位号上一个未缺席的座位号
func (slf *RoomSeat[PlayerID, Player]) GetPrevSeat(seat int) int {
l := len(slf.seatSP)
if l == 0 || seat >= l || seat < 0 {
return -1
}
var target = seat
for {
target--
if target < 0 {
target = l - 1
}
if target == seat {
return seat
}
if slf.seatSP[target] != nil {
return target
}
}
}
// GetPrevSeatVacancy 获取特定座位号上一个座位号
// - 缺席将不会被忽略
func (slf *RoomSeat[PlayerID, Player]) GetPrevSeatVacancy(seat int) int {
l := len(slf.seatSP)
if l == 0 || seat >= l || seat < 0 {
return -1
}
seat--
if seat < 0 {
seat = l - 1
}
return seat
}
func (slf *RoomSeat[PlayerID, Player]) onJoinRoom(room game.Room[PlayerID, Player], player Player) {
slf.AddSeat(player.GetID())
}

39
game/builtin/world.go
View File

@ -2,9 +2,8 @@ package builtin
import (
"github.com/kercylan98/minotaur/game"
"github.com/kercylan98/minotaur/utils/hash"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/log"
"github.com/kercylan98/minotaur/utils/synchronization"
"sync/atomic"
)
@ -12,10 +11,10 @@ import (
func NewWorld[PlayerID comparable, Player game.Player[PlayerID]](guid int64, options ...WorldOption[PlayerID, Player]) *World[PlayerID, Player] {
world := &World[PlayerID, Player]{
guid: guid,
players: synchronization.NewMap[PlayerID, Player](),
playerActors: synchronization.NewMap[PlayerID, hash.Map[int64, game.Actor]](),
owners: synchronization.NewMap[int64, PlayerID](),
actors: synchronization.NewMap[int64, game.Actor](),
players: concurrent.NewBalanceMap[PlayerID, Player](),
playerActors: concurrent.NewBalanceMap[PlayerID, *concurrent.BalanceMap[int64, game.Actor]](),
owners: concurrent.NewBalanceMap[int64, PlayerID](),
actors: concurrent.NewBalanceMap[int64, game.Actor](),
}
for _, option := range options {
option(world)
@ -28,10 +27,10 @@ type World[PlayerID comparable, Player game.Player[PlayerID]] struct {
guid int64
actorGuid atomic.Int64
playerLimit int
players hash.Map[PlayerID, Player]
playerActors hash.Map[PlayerID, hash.Map[int64, game.Actor]]
owners hash.Map[int64, PlayerID]
actors hash.Map[int64, game.Actor]
players *concurrent.BalanceMap[PlayerID, Player]
playerActors *concurrent.BalanceMap[PlayerID, *concurrent.BalanceMap[int64, game.Actor]]
owners *concurrent.BalanceMap[int64, PlayerID]
actors *concurrent.BalanceMap[int64, game.Actor]
playerJoinWorldEventHandles []game.PlayerJoinWorldEventHandle[PlayerID, Player]
playerLeaveWorldEventHandles []game.PlayerLeaveWorldEventHandle[PlayerID, Player]
@ -56,16 +55,16 @@ func (slf *World[PlayerID, Player]) GetPlayer(id PlayerID) Player {
return slf.players.Get(id)
}
func (slf *World[PlayerID, Player]) GetPlayers() hash.MapReadonly[PlayerID, Player] {
return slf.players.(hash.MapReadonly[PlayerID, Player])
func (slf *World[PlayerID, Player]) GetPlayers() map[PlayerID]Player {
return slf.players.Map()
}
func (slf *World[PlayerID, Player]) GetActor(guid int64) game.Actor {
return slf.actors.Get(guid)
}
func (slf *World[PlayerID, Player]) GetActors() hash.MapReadonly[int64, game.Actor] {
return slf.actors.(hash.MapReadonly[int64, game.Actor])
func (slf *World[PlayerID, Player]) GetActors() map[int64]game.Actor {
return slf.actors.Map()
}
func (slf *World[PlayerID, Player]) GetPlayerActor(id PlayerID, guid int64) game.Actor {
@ -75,8 +74,8 @@ func (slf *World[PlayerID, Player]) GetPlayerActor(id PlayerID, guid int64) game
return nil
}
func (slf *World[PlayerID, Player]) GetPlayerActors(id PlayerID) hash.MapReadonly[int64, game.Actor] {
return slf.playerActors.Get(id).(hash.MapReadonly[int64, game.Actor])
func (slf *World[PlayerID, Player]) GetPlayerActors(id PlayerID) map[int64]game.Actor {
return slf.playerActors.Get(id).Map()
}
func (slf *World[PlayerID, Player]) IsExistPlayer(id PlayerID) bool {
@ -105,7 +104,7 @@ func (slf *World[PlayerID, Player]) Join(player Player) error {
log.Debug("World.Join", log.Int64("guid", slf.GetGuid()), log.Any("player", player.GetID()))
slf.players.Set(player.GetID(), player)
if actors := slf.playerActors.Get(player.GetID()); actors == nil {
actors = synchronization.NewMap[int64, game.Actor]()
actors = concurrent.NewBalanceMap[int64, game.Actor]()
slf.playerActors.Set(player.GetID(), actors)
}
slf.OnPlayerJoinWorldEvent(player)
@ -119,9 +118,10 @@ func (slf *World[PlayerID, Player]) Leave(id PlayerID) {
}
log.Debug("World.Leave", log.Int64("guid", slf.GetGuid()), log.Any("player", player.GetID()))
slf.OnPlayerLeaveWorldEvent(player)
slf.playerActors.Get(player.GetID()).Range(func(guid int64, actor game.Actor) {
slf.playerActors.Get(player.GetID()).Range(func(guid int64, actor game.Actor) bool {
slf.OnActorAnnihilationEvent(actor)
slf.owners.Delete(guid)
return false
})
slf.playerActors.Delete(player.GetID())
slf.players.Delete(player.GetID())
@ -171,8 +171,9 @@ func (slf *World[PlayerID, Player]) RemoveActorOwner(guid int64) {
func (slf *World[PlayerID, Player]) Reset() {
log.Debug("World", log.Int64("Reset", slf.guid))
slf.players.Clear()
slf.playerActors.Range(func(id PlayerID, actors hash.Map[int64, game.Actor]) {
slf.playerActors.Range(func(id PlayerID, actors *concurrent.BalanceMap[int64, game.Actor]) bool {
actors.Clear()
return false
})
slf.playerActors.Clear()
slf.owners.Clear()

20
game/poker/poker.go
View File

@ -68,3 +68,23 @@ func GetCardsColor(cards ...Card) []Color {
}
return colors
}
// IsContain 一组扑克牌是否包含某张牌
func IsContain(cards []Card, card Card) bool {
for _, c := range cards {
if c.Equal(card) {
return true
}
}
return false
}
// IsContainAll 一组扑克牌是否包含另一组扑克牌
func IsContainAll(cards []Card, cards2 []Card) bool {
for _, card := range cards2 {
if !IsContain(cards, card) {
return false
}
}
return true
}

6
game/room.go
View File

@ -1,9 +1,5 @@
package game
import (
"github.com/kercylan98/minotaur/utils/hash"
)
// Room 房间类似于简版的游戏世界(World)，不过没有游戏实体
type Room[PlayerID comparable, P Player[PlayerID]] interface {
// GetGuid 获取房间的唯一标识符
@ -13,7 +9,7 @@ type Room[PlayerID comparable, P Player[PlayerID]] interface {
// GetPlayer 根据玩家id获取玩家
GetPlayer(id PlayerID) P
// GetPlayers 获取房间中的所有玩家
GetPlayers() hash.MapReadonly[PlayerID, P]
GetPlayers() map[PlayerID]P
// GetPlayerCount 获取玩家数量
GetPlayerCount() int
// IsExistPlayer 检查房间中是否存在特定玩家

8
game/room_seat.go
View File

@ -36,4 +36,12 @@ type RoomSeat[PlayerID comparable, P Player[PlayerID]] interface {
// - 超出范围将返回-1
// - 当没有下一个座位号时将始终返回本身
GetNextSeatVacancy(seat int) int
// GetPrevSeat 获取上一个座位号，空缺的位置将会被跳过
// - 超出范围将返回-1
// - 当没有上一个座位号时将始终返回本身
GetPrevSeat(seat int) int
// GetPrevSeatVacancy 获取上一个座位号，空缺的位置将被保留
// - 超出范围将返回-1
// - 当没有上一个座位号时将始终返回本身
GetPrevSeatVacancy(seat int) int
}

10
game/world.go
View File

@ -1,9 +1,5 @@
package game
import (
"github.com/kercylan98/minotaur/utils/hash"
)
// World 游戏世界接口定义
type World[PlayerID comparable, P Player[PlayerID]] interface {
// GetGuid 获取世界的唯一标识符
@ -13,15 +9,15 @@ type World[PlayerID comparable, P Player[PlayerID]] interface {
// GetPlayer 根据玩家id获取玩家
GetPlayer(id PlayerID) P
// GetPlayers 获取世界中的所有玩家
GetPlayers() hash.MapReadonly[PlayerID, P]
GetPlayers() map[PlayerID]P
// GetActor 根据唯一标识符获取世界中的游戏对象
GetActor(guid int64) Actor
// GetActors 获取世界中的所有游戏对象
GetActors() hash.MapReadonly[int64, Actor]
GetActors() map[int64]Actor
// GetPlayerActor 获取游戏世界中归属特定玩家的特定游戏对象
GetPlayerActor(id PlayerID, guid int64) Actor
// GetPlayerActors 获取游戏世界中归属特定玩家的所有游戏对象
GetPlayerActors(id PlayerID) hash.MapReadonly[int64, Actor]
GetPlayerActors(id PlayerID) map[int64]Actor
// IsExistPlayer 检查游戏世界中是否存在特定玩家
IsExistPlayer(id PlayerID) bool
// IsExistActor 检查游戏世界中是否存在特定游戏对象

2
go.mod
View File

@ -23,6 +23,7 @@ require (
)
require (
github.com/alphadose/haxmap v1.2.0 // indirect
github.com/bytedance/sonic v1.9.1 // indirect
github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect
github.com/gabriel-vasile/mimetype v1.4.2 // indirect
@ -61,6 +62,7 @@ require (
go.uber.org/multierr v1.11.0 // indirect
golang.org/x/arch v0.3.0 // indirect
golang.org/x/crypto v0.9.0 // indirect
golang.org/x/exp v0.0.0-20221031165847-c99f073a8326 // indirect
golang.org/x/net v0.10.0 // indirect
golang.org/x/sys v0.8.0 // indirect
golang.org/x/text v0.9.0 // indirect

4
go.sum
View File

@ -2,6 +2,8 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
github.com/RussellLuo/timingwheel v0.0.0-20220218152713-54845bda3108 h1:iPugyBI7oFtbDZXC4dnY093M1kZx6k/95sen92gafbY=
github.com/RussellLuo/timingwheel v0.0.0-20220218152713-54845bda3108/go.mod h1:WAMLHwunr1hi3u7OjGV6/VWG9QbdMhGpEKjROiSFd10=
github.com/alphadose/haxmap v1.2.0 h1:noGrAmCE+gNheZ4KpW+sYj9W5uMcO1UAjbAq9XBOAfM=
github.com/alphadose/haxmap v1.2.0/go.mod h1:rjHw1IAqbxm0S3U5tD16GoKsiAd8FWx5BJ2IYqXwgmM=
github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM=
@ -219,6 +221,8 @@ golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5y
golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
golang.org/x/exp v0.0.0-20221031165847-c99f073a8326 h1:QfTh0HpN6hlw6D3vu8DAwC8pBIwikq0AI1evdm+FksE=
golang.org/x/exp v0.0.0-20221031165847-c99f073a8326/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=

6
server/conn.go
View File

@ -2,7 +2,7 @@ package server
import (
"github.com/gorilla/websocket"
"github.com/kercylan98/minotaur/utils/synchronization"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/panjf2000/gnet"
"github.com/xtaci/kcp-go/v5"
"net"
@ -75,7 +75,7 @@ type Conn struct {
kcp *kcp.UDPSession
data map[any]any
mutex sync.Mutex
packetPool *synchronization.Pool[*connPacket]
packetPool *concurrent.Pool[*connPacket]
packets []*connPacket
}
@ -187,7 +187,7 @@ func (slf *Conn) WriteWithCallback(packet Packet, callback func(err error), mess
// writeLoop 写循环
func (slf *Conn) writeLoop(wait *sync.WaitGroup) {
slf.packetPool = synchronization.NewPool[*connPacket](10*1024,
slf.packetPool = concurrent.NewPool[*connPacket](10*1024,
func() *connPacket {
return &connPacket{}
}, func(data *connPacket) {

6
server/cross/nats.go
View File

@ -4,8 +4,8 @@ import (
"encoding/json"
"fmt"
"github.com/kercylan98/minotaur/server"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/log"
"github.com/kercylan98/minotaur/utils/synchronization"
"github.com/nats-io/nats.go"
"time"
)
@ -18,7 +18,7 @@ func NewNats(url string, options ...NatsOption) *Nats {
n := &Nats{
url: url,
subject: "MINOTAUR_CROSS",
messagePool: synchronization.NewPool[*Message](1024*100, func() *Message {
messagePool: concurrent.NewPool[*Message](1024*100, func() *Message {
return new(Message)
}, func(data *Message) {
data.ServerId = 0
@ -36,7 +36,7 @@ type Nats struct {
url string
subject string
options []nats.Option
messagePool *synchronization.Pool[*Message]
messagePool *concurrent.Pool[*Message]
}
func (slf *Nats) Init(server *server.Server, packetHandle func(serverId int64, packet []byte)) (err error) {

10
server/server.go
View File

@ -5,9 +5,9 @@ import (
"fmt"
"github.com/gin-gonic/gin"
"github.com/gorilla/websocket"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/log"
"github.com/kercylan98/minotaur/utils/super"
"github.com/kercylan98/minotaur/utils/synchronization"
"github.com/kercylan98/minotaur/utils/timer"
"github.com/kercylan98/minotaur/utils/times"
"github.com/panjf2000/ants/v2"
@ -33,7 +33,7 @@ func New(network Network, options ...Option) *Server {
runtime: &runtime{messagePoolSize: DefaultMessageBufferSize, messageChannelSize: DefaultMessageChannelSize},
option: &option{},
network: network,
online: synchronization.NewMap[string, *Conn](),
online: concurrent.NewBalanceMap[string, *Conn](),
closeChannel: make(chan struct{}, 1),
systemSignal: make(chan os.Signal, 1),
}
@ -78,7 +78,7 @@ type Server struct {
network Network // 网络类型
addr string // 侦听地址
systemSignal chan os.Signal // 系统信号
online *synchronization.Map[string, *Conn] // 在线连接
online *concurrent.BalanceMap[string, *Conn] // 在线连接
ginServer *gin.Engine // HTTP模式下的路由器
httpServer *http.Server // HTTP模式下的服务器
grpcServer *grpc.Server // GRPC模式下的服务器
@ -87,7 +87,7 @@ type Server struct {
isShutdown atomic.Bool // 是否已关闭
closeChannel chan struct{} // 关闭信号
ants *ants.Pool // 协程池
messagePool *synchronization.Pool[*Message] // 消息池
messagePool *concurrent.Pool[*Message] // 消息池
messageChannel chan *Message // 消息管道
multiple *MultipleServer // 多服务器模式下的服务器
multipleRuntimeErrorChan chan error // 多服务器模式下的运行时错误
@ -115,7 +115,7 @@ func (slf *Server) Run(addr string) error {
var protoAddr = fmt.Sprintf("%s://%s", slf.network, slf.addr)
var messageInitFinish = make(chan struct{}, 1)
var connectionInitHandle = func(callback func()) {
slf.messagePool = synchronization.NewPool[*Message](slf.messagePoolSize,
slf.messagePool = concurrent.NewPool[*Message](slf.messagePoolSize,
func() *Message {
return &Message{}
},

195
utils/asynchronous/map.go
View File

@ -1,195 +0,0 @@
package asynchronous
import (
"encoding/json"
"github.com/kercylan98/minotaur/utils/hash"
)
func NewMap[Key comparable, Value any](options ...MapOption[Key, Value]) *Map[Key, Value] {
m := &Map[Key, Value]{
data: make(map[Key]Value),
}
for _, option := range options {
option(m)
}
return m
}
// Map 非并发安全的字典数据结构
// - 可用于对 synchronization.Map 的替代
type Map[Key comparable, Value any] struct {
data map[Key]Value
}
func (slf *Map[Key, Value]) Set(key Key, value Value) {
slf.data[key] = value
}
func (slf *Map[Key, Value]) Get(key Key) Value {
return slf.data[key]
}
// AtomGetSet 原子方式获取一个值并在之后进行赋值
func (slf *Map[Key, Value]) AtomGetSet(key Key, handle func(value Value, exist bool) (newValue Value, isSet bool)) {
value, exist := slf.data[key]
if newValue, isSet := handle(value, exist); isSet {
slf.data[key] = newValue
}
}
// Atom 原子操作
func (slf *Map[Key, Value]) Atom(handle func(m hash.Map[Key, Value])) {
handle(slf)
}
func (slf *Map[Key, Value]) Exist(key Key) bool {
_, exist := slf.data[key]
return exist
}
func (slf *Map[Key, Value]) GetExist(key Key) (Value, bool) {
value, exist := slf.data[key]
return value, exist
}
func (slf *Map[Key, Value]) Delete(key Key) {
delete(slf.data, key)
}
func (slf *Map[Key, Value]) DeleteGet(key Key) Value {
v := slf.data[key]
delete(slf.data, key)
return v
}
func (slf *Map[Key, Value]) DeleteGetExist(key Key) (Value, bool) {
v, exist := slf.data[key]
delete(slf.data, key)
return v, exist
}
func (slf *Map[Key, Value]) DeleteExist(key Key) bool {
if _, exist := slf.data[key]; !exist {
return exist
}
delete(slf.data, key)
return true
}
func (slf *Map[Key, Value]) Clear() {
for k := range slf.data {
delete(slf.data, k)
}
}
func (slf *Map[Key, Value]) ClearHandle(handle func(key Key, value Value)) {
for k, v := range slf.data {
handle(k, v)
delete(slf.data, k)
}
}
func (slf *Map[Key, Value]) Range(handle func(key Key, value Value)) {
for k, v := range slf.data {
key, value := k, v
handle(key, value)
}
}
func (slf *Map[Key, Value]) RangeSkip(handle func(key Key, value Value) bool) {
for k, v := range slf.data {
key, value := k, v
if !handle(key, value) {
continue
}
}
}
func (slf *Map[Key, Value]) RangeBreakout(handle func(key Key, value Value) bool) {
slf.rangeBreakout(handle)
}
func (slf *Map[Key, Value]) rangeBreakout(handle func(key Key, value Value) bool) bool {
for k, v := range slf.data {
key, value := k, v
if !handle(key, value) {
return true
}
}
return false
}
func (slf *Map[Key, Value]) RangeFree(handle func(key Key, value Value, skip func(), breakout func())) {
slf.rangeFree(handle)
}
func (slf *Map[Key, Value]) rangeFree(handle func(key Key, value Value, skip func(), breakout func())) bool {
var skipExec, breakoutExec bool
var skip = func() {
skipExec = true
}
var breakout = func() {
breakoutExec = true
}
for k, v := range slf.data {
key, value := k, v
handle(key, value, skip, breakout)
if skipExec {
continue
}
if breakoutExec {
break
}
}
return breakoutExec
}
func (slf *Map[Key, Value]) Keys() []Key {
var s = make([]Key, 0, len(slf.data))
for k := range slf.data {
s = append(s, k)
}
return s
}
func (slf *Map[Key, Value]) Slice() []Value {
var s = make([]Value, 0, len(slf.data))
for _, v := range slf.data {
s = append(s, v)
}
return s
}
func (slf *Map[Key, Value]) Map() map[Key]Value {
var m = make(map[Key]Value)
for k, v := range slf.data {
m[k] = v
}
return m
}
func (slf *Map[Key, Value]) Size() int {
return len(slf.data)
}
// GetOne 获取一个
func (slf *Map[Key, Value]) GetOne() (value Value) {
for _, v := range slf.data {
return v
}
return value
}
func (slf *Map[Key, Value]) MarshalJSON() ([]byte, error) {
m := slf.Map()
return json.Marshal(m)
}
func (slf *Map[Key, Value]) UnmarshalJSON(bytes []byte) error {
var m = make(map[Key]Value)
if err := json.Unmarshal(bytes, &m); err != nil {
return err
}
slf.data = m
return nil
}

10
utils/asynchronous/map_options.go
View File

@ -1,10 +0,0 @@
package asynchronous
type MapOption[Key comparable, Value any] func(m *Map[Key, Value])
// WithMapSource 通过传入的 map 初始化
func WithMapSource[Key comparable, Value any](source map[Key]Value) MapOption[Key, Value] {
return func(m *Map[Key, Value]) {
m.data = source
}
}

185
utils/concurrent/balance_map.go Normal file
View File

@ -0,0 +1,185 @@
package concurrent
import (
"encoding/json"
"sync"
)
// NewBalanceMap 创建一个并发安全且性能在普通读写和并发读写的情况下较为平衡的字典数据结构
func NewBalanceMap[Key comparable, value any](options ...BalanceMapOption[Key, value]) *BalanceMap[Key, value] {
m := &BalanceMap[Key, value]{
data: make(map[Key]value),
}
for _, option := range options {
option(m)
}
return m
}
// BalanceMap 并发安全且性能在普通读写和并发读写的情况下较为平衡的字典数据结构
// - 适用于要考虑并发读写但是并发读写的频率不高的情况
type BalanceMap[Key comparable, Value any] struct {
lock sync.RWMutex
data map[Key]Value
}
// Set 设置一个值
func (slf *BalanceMap[Key, Value]) Set(key Key, value Value) {
slf.lock.Lock()
defer slf.lock.Unlock()
slf.data[key] = value
}
// Get 获取一个值
func (slf *BalanceMap[Key, Value]) Get(key Key) Value {
slf.lock.RLock()
defer slf.lock.RUnlock()
return slf.data[key]
}
// Atom 原子操作
func (slf *BalanceMap[Key, Value]) Atom(handle func(m map[Key]Value)) {
slf.lock.Lock()
handle(slf.data)
slf.lock.Unlock()
}
// Exist 判断是否存在
func (slf *BalanceMap[Key, Value]) Exist(key Key) bool {
slf.lock.RLock()
_, exist := slf.data[key]
slf.lock.RUnlock()
return exist
}
// GetExist 获取一个值并判断是否存在
func (slf *BalanceMap[Key, Value]) GetExist(key Key) (Value, bool) {
slf.lock.RLock()
value, exist := slf.data[key]
slf.lock.RUnlock()
return value, exist
}
// Delete 删除一个值
func (slf *BalanceMap[Key, Value]) Delete(key Key) {
slf.lock.Lock()
delete(slf.data, key)
defer slf.lock.Unlock()
}
// DeleteGet 删除一个值并返回
func (slf *BalanceMap[Key, Value]) DeleteGet(key Key) Value {
slf.lock.Lock()
v := slf.data[key]
delete(slf.data, key)
slf.lock.Unlock()
return v
}
// DeleteGetExist 删除一个值并返回是否存在
func (slf *BalanceMap[Key, Value]) DeleteGetExist(key Key) (Value, bool) {
slf.lock.Lock()
v, exist := slf.data[key]
delete(slf.data, key)
defer slf.lock.Unlock()
return v, exist
}
// DeleteExist 删除一个值并返回是否存在
func (slf *BalanceMap[Key, Value]) DeleteExist(key Key) bool {
slf.lock.Lock()
if _, exist := slf.data[key]; !exist {
slf.lock.Unlock()
return exist
}
delete(slf.data, key)
slf.lock.Unlock()
return true
}
// Clear 清空
func (slf *BalanceMap[Key, Value]) Clear() {
slf.lock.Lock()
for k := range slf.data {
delete(slf.data, k)
}
slf.lock.Unlock()
}
// ClearHandle 清空并处理
func (slf *BalanceMap[Key, Value]) ClearHandle(handle func(key Key, value Value)) {
slf.lock.Lock()
for k, v := range slf.data {
handle(k, v)
delete(slf.data, k)
}
slf.lock.Unlock()
}
// Range 遍历所有值，如果 handle 返回 true 则停止遍历
func (slf *BalanceMap[Key, Value]) Range(handle func(key Key, value Value) bool) {
slf.lock.Lock()
for k, v := range slf.data {
key, value := k, v
if handle(key, value) {
break
}
}
slf.lock.Unlock()
}
// Keys 获取所有的键
func (slf *BalanceMap[Key, Value]) Keys() []Key {
slf.lock.RLock()
var s = make([]Key, 0, len(slf.data))
for k := range slf.data {
s = append(s, k)
}
slf.lock.RUnlock()
return s
}
// Slice 获取所有的值
func (slf *BalanceMap[Key, Value]) Slice() []Value {
slf.lock.RLock()
var s = make([]Value, 0, len(slf.data))
for _, v := range slf.data {
s = append(s, v)
}
slf.lock.RUnlock()
return s
}
// Map 转换为普通 map
func (slf *BalanceMap[Key, Value]) Map() map[Key]Value {
slf.lock.RLock()
var m = make(map[Key]Value)
for k, v := range slf.data {
m[k] = v
}
slf.lock.RUnlock()
return m
}
// Size 获取数量
func (slf *BalanceMap[Key, Value]) Size() int {
slf.lock.RLock()
defer slf.lock.RUnlock()
return len(slf.data)
}
func (slf *BalanceMap[Key, Value]) MarshalJSON() ([]byte, error) {
m := slf.Map()
return json.Marshal(m)
}
func (slf *BalanceMap[Key, Value]) UnmarshalJSON(bytes []byte) error {
var m = make(map[Key]Value)
if err := json.Unmarshal(bytes, &m); err != nil {
return err
}
slf.lock.Lock()
slf.data = m
slf.lock.Unlock()
return nil
}

11
utils/concurrent/balance_map_options.go Normal file
View File

@ -0,0 +1,11 @@
package concurrent
// BalanceMapOption BalanceMap 的选项
type BalanceMapOption[Key comparable, Value any] func(m *BalanceMap[Key, Value])
// WithBalanceMapSource 通过传入的 map 初始化
func WithBalanceMapSource[Key comparable, Value any](source map[Key]Value) BalanceMapOption[Key, Value] {
return func(m *BalanceMap[Key, Value]) {
m.data = source
}
}

2
utils/synchronization/pool.go → utils/concurrent/pool.go
View File

@ -1,4 +1,4 @@
package synchronization
package concurrent
import (
"github.com/kercylan98/minotaur/utils/log"

2
utils/synchronization/slice.go → utils/concurrent/slice.go
View File

@ -1,4 +1,4 @@
package synchronization
package concurrent
import (
"github.com/kercylan98/minotaur/utils/slice"

2
utils/synchronization/slice_option.go → utils/concurrent/slice_option.go
View File

@ -1,4 +1,4 @@
package synchronization
package concurrent
type SliceOption[T any] func(slice *Slice[T])

29
utils/hash/map.go
View File

@ -1,29 +0,0 @@
package hash
// Map 提供了map集合接口
type Map[Key comparable, Value any] interface {
Set(key Key, value Value)
Get(key Key) Value
// AtomGetSet 原子方式获取一个值并在之后进行赋值
AtomGetSet(key Key, handle func(value Value, exist bool) (newValue Value, isSet bool))
// Atom 原子操作
Atom(handle func(m Map[Key, Value]))
Exist(key Key) bool
GetExist(key Key) (Value, bool)
Delete(key Key)
DeleteGet(key Key) Value
DeleteGetExist(key Key) (Value, bool)
DeleteExist(key Key) bool
Clear()
ClearHandle(handle func(key Key, value Value))
Range(handle func(key Key, value Value))
RangeSkip(handle func(key Key, value Value) bool)
RangeBreakout(handle func(key Key, value Value) bool)
RangeFree(handle func(key Key, value Value, skip func(), breakout func()))
Keys() []Key
Slice() []Value
Map() map[Key]Value
Size() int
// GetOne 获取一个
GetOne() (value Value)
}

17
utils/hash/map_readonly.go
View File

@ -1,17 +0,0 @@
package hash
// MapReadonly 只读字典接口
type MapReadonly[Key comparable, Value any] interface {
Get(key Key) Value
Exist(key Key) bool
GetExist(key Key) (Value, bool)
Range(handle func(key Key, value Value))
RangeSkip(handle func(key Key, value Value) bool)
RangeBreakout(handle func(key Key, value Value) bool)
RangeFree(handle func(key Key, value Value, skip func(), breakout func()))
Keys() []Key
Slice() []Value
Map() map[Key]Value
Size() int
GetOne() (value Value)
}

19
utils/stream/map.go
View File

@ -1,9 +1,8 @@
package stream
import (
"github.com/kercylan98/minotaur/utils/asynchronous"
"github.com/kercylan98/minotaur/utils/concurrent"
"github.com/kercylan98/minotaur/utils/hash"
"github.com/kercylan98/minotaur/utils/synchronization"
"reflect"
)
@ -19,11 +18,6 @@ func WithMapCopy[K comparable, V any](m map[K]V) Map[K, V] {
return hash.Copy(m)
}
// WithHashMap 使用传入的 map 执行链式操作
func WithHashMap[K comparable, V any](m hash.Map[K, V]) Map[K, V] {
return m.Map()
}
// Map 提供了 map 的链式操作
type Map[K comparable, V any] map[K]V
@ -180,14 +174,9 @@ func (slf Map[K, V]) ToSliceStreamWithKey() Slice[K] {
return hash.KeyToSlice(slf)
}
// ToSyncMap 将当前 Map 转换为 synchronization.Map
func (slf Map[K, V]) ToSyncMap() *synchronization.Map[K, V] {
return synchronization.NewMap[K, V](synchronization.WithMapSource(slf))
}
// ToAsyncMap 将当前 Map 转换为 asynchronous.Map
func (slf Map[K, V]) ToAsyncMap() *asynchronous.Map[K, V] {
return asynchronous.NewMap[K, V](asynchronous.WithMapSource(slf))
// ToSyncMap 将当前 Map 转换为 concurrent.BalanceMap
func (slf Map[K, V]) ToSyncMap() *concurrent.BalanceMap[K, V] {
return concurrent.NewBalanceMap[K, V](concurrent.WithBalanceMapSource(slf))
}
// ToMap 将当前 Map 转换为 map

285
utils/synchronization/map.go
View File

@ -1,285 +0,0 @@
package synchronization
import (
"encoding/json"
"github.com/kercylan98/minotaur/utils/hash"
"sync"
)
func NewMap[Key comparable, value any](options ...MapOption[Key, value]) *Map[Key, value] {
m := &Map[Key, value]{
data: make(map[Key]value),
}
for _, option := range options {
option(m)
}
return m
}
// Map 并发安全的字典数据结构
type Map[Key comparable, Value any] struct {
lock sync.RWMutex
data map[Key]Value
atom bool
}
func (slf *Map[Key, Value]) Set(key Key, value Value) {
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
slf.data[key] = value
}
func (slf *Map[Key, Value]) Get(key Key) Value {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
return slf.data[key]
}
// AtomGetSet 原子方式获取一个值并在之后进行赋值
func (slf *Map[Key, Value]) AtomGetSet(key Key, handle func(value Value, exist bool) (newValue Value, isSet bool)) {
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
value, exist := slf.data[key]
if newValue, isSet := handle(value, exist); isSet {
slf.data[key] = newValue
}
}
// Atom 原子操作
func (slf *Map[Key, Value]) Atom(handle func(m hash.Map[Key, Value])) {
slf.lock.Lock()
slf.atom = true
handle(slf)
slf.atom = false
slf.lock.Unlock()
}
func (slf *Map[Key, Value]) Exist(key Key) bool {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
_, exist := slf.data[key]
return exist
}
func (slf *Map[Key, Value]) GetExist(key Key) (Value, bool) {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
value, exist := slf.data[key]
return value, exist
}
func (slf *Map[Key, Value]) Delete(key Key) {
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
delete(slf.data, key)
}
func (slf *Map[Key, Value]) DeleteGet(key Key) Value {
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
v := slf.data[key]
delete(slf.data, key)
return v
}
func (slf *Map[Key, Value]) DeleteGetExist(key Key) (Value, bool) {
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
v, exist := slf.data[key]
delete(slf.data, key)
return v, exist
}
func (slf *Map[Key, Value]) DeleteExist(key Key) bool {
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
if _, exist := slf.data[key]; !exist {
return exist
}
delete(slf.data, key)
return true
}
func (slf *Map[Key, Value]) Clear() {
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
for k := range slf.data {
delete(slf.data, k)
}
}
func (slf *Map[Key, Value]) ClearHandle(handle func(key Key, value Value)) {
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
for k, v := range slf.data {
handle(k, v)
delete(slf.data, k)
}
}
func (slf *Map[Key, Value]) Range(handle func(key Key, value Value)) {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
for k, v := range slf.data {
key, value := k, v
handle(key, value)
}
}
func (slf *Map[Key, Value]) RangeSkip(handle func(key Key, value Value) bool) {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
for k, v := range slf.data {
key, value := k, v
if !handle(key, value) {
continue
}
}
}
func (slf *Map[Key, Value]) RangeBreakout(handle func(key Key, value Value) bool) {
slf.rangeBreakout(handle)
}
func (slf *Map[Key, Value]) rangeBreakout(handle func(key Key, value Value) bool) bool {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
for k, v := range slf.data {
key, value := k, v
if !handle(key, value) {
return true
}
}
return false
}
func (slf *Map[Key, Value]) RangeFree(handle func(key Key, value Value, skip func(), breakout func())) {
slf.rangeFree(handle)
}
func (slf *Map[Key, Value]) rangeFree(handle func(key Key, value Value, skip func(), breakout func())) bool {
var skipExec, breakoutExec bool
var skip = func() {
skipExec = true
}
var breakout = func() {
breakoutExec = true
}
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
for k, v := range slf.data {
key, value := k, v
handle(key, value, skip, breakout)
if skipExec {
continue
}
if breakoutExec {
break
}
}
return breakoutExec
}
func (slf *Map[Key, Value]) Keys() []Key {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
var s = make([]Key, 0, len(slf.data))
for k := range slf.data {
s = append(s, k)
}
return s
}
func (slf *Map[Key, Value]) Slice() []Value {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
var s = make([]Value, 0, len(slf.data))
for _, v := range slf.data {
s = append(s, v)
}
return s
}
func (slf *Map[Key, Value]) Map() map[Key]Value {
var m = make(map[Key]Value)
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
for k, v := range slf.data {
m[k] = v
}
return m
}
func (slf *Map[Key, Value]) Size() int {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
return len(slf.data)
}
// GetOne 获取一个
func (slf *Map[Key, Value]) GetOne() (value Value) {
if !slf.atom {
slf.lock.RLock()
defer slf.lock.RUnlock()
}
for _, v := range slf.data {
return v
}
return value
}
func (slf *Map[Key, Value]) MarshalJSON() ([]byte, error) {
m := slf.Map()
return json.Marshal(m)
}
func (slf *Map[Key, Value]) UnmarshalJSON(bytes []byte) error {
var m = make(map[Key]Value)
if err := json.Unmarshal(bytes, &m); err != nil {
return err
}
if !slf.atom {
slf.lock.Lock()
defer slf.lock.Unlock()
}
slf.data = m
return nil
}

10
utils/synchronization/map_options.go
View File

@ -1,10 +0,0 @@
package synchronization
type MapOption[Key comparable, Value any] func(m *Map[Key, Value])
// WithMapSource 通过传入的 map 初始化
func WithMapSource[Key comparable, Value any](source map[Key]Value) MapOption[Key, Value] {
return func(m *Map[Key, Value]) {
m.data = source
}
}

259
utils/synchronization/map_segment.go
View File

@ -1,259 +0,0 @@
package synchronization
import (
"encoding/json"
"github.com/kercylan98/minotaur/utils/hash"
"sync"
)
func NewMapSegment[Key comparable, value any](segmentCount int) *MapSegment[Key, value] {
ms := &MapSegment[Key, value]{
segments: map[int]*Map[Key, value]{},
cache: map[Key]int{},
consistency: hash.NewConsistency(segmentCount),
}
for i := 0; i < segmentCount; i++ {
ms.consistency.AddNode(i)
ms.segments[i] = NewMap[Key, value]()
}
return ms
}
// MapSegment 基于分段锁实现的并发安全的字典数据结构map
type MapSegment[Key comparable, Value any] struct {
segments map[int]*Map[Key, Value]
cache map[Key]int
consistency *hash.Consistency
lock sync.RWMutex
}
func (slf *MapSegment[Key, Value]) Atom(handle func(m hash.Map[Key, Value])) {
panic("this function is currently not supported")
}
func (slf *MapSegment[Key, Value]) Set(key Key, value Value) {
slf.lock.RLock()
s, exist := slf.cache[key]
slf.lock.RUnlock()
if !exist {
s = slf.consistency.PickNode(key)
slf.lock.Lock()
slf.cache[key] = s
slf.lock.Unlock()
}
slf.segments[s].Set(key, value)
}
func (slf *MapSegment[Key, Value]) Get(key Key) (value Value) {
slf.lock.RLock()
s, exist := slf.cache[key]
slf.lock.RUnlock()
if !exist {
return value
}
return slf.segments[s].Get(key)
}
// AtomGetSet 原子方式获取一个值并在之后进行赋值
func (slf *MapSegment[Key, Value]) AtomGetSet(key Key, handle func(value Value, exist bool) (newValue Value, isSet bool)) {
var value Value
slf.lock.RLock()
s, exist := slf.cache[key]
slf.lock.RUnlock()
if !exist {
if newValue, isSet := handle(value, exist); isSet {
slf.Set(key, newValue)
}
return
}
slf.segments[s].AtomGetSet(key, handle)
}
func (slf *MapSegment[Key, Value]) Exist(key Key) bool {
slf.lock.RLock()
_, exist := slf.cache[key]
slf.lock.RUnlock()
return exist
}
func (slf *MapSegment[Key, Value]) GetExist(key Key) (value Value, exist bool) {
slf.lock.RLock()
s, exist := slf.cache[key]
slf.lock.RUnlock()
if !exist {
return value, false
}
return slf.segments[s].GetExist(key)
}
func (slf *MapSegment[Key, Value]) Delete(key Key) {
slf.lock.Lock()
s, exist := slf.cache[key]
delete(slf.cache, key)
slf.lock.Unlock()
if exist {
slf.segments[s].Delete(key)
}
}
func (slf *MapSegment[Key, Value]) DeleteGet(key Key) (value Value) {
slf.lock.Lock()
s, exist := slf.cache[key]
delete(slf.cache, key)
slf.lock.Unlock()
if exist {
return slf.segments[s].DeleteGet(key)
}
return
}
func (slf *MapSegment[Key, Value]) DeleteGetExist(key Key) (value Value, exist bool) {
slf.lock.Lock()
s, exist := slf.cache[key]
delete(slf.cache, key)
slf.lock.Unlock()
if exist {
return slf.segments[s].DeleteGetExist(key)
}
return value, exist
}
func (slf *MapSegment[Key, Value]) DeleteExist(key Key) bool {
slf.lock.Lock()
s, exist := slf.cache[key]
delete(slf.cache, key)
slf.lock.Unlock()
if exist {
return slf.segments[s].DeleteExist(key)
}
return false
}
func (slf *MapSegment[Key, Value]) Clear() {
slf.lock.Lock()
for k := range slf.cache {
delete(slf.cache, k)
}
for _, m := range slf.segments {
m.Clear()
}
slf.lock.Unlock()
}
func (slf *MapSegment[Key, Value]) ClearHandle(handle func(key Key, value Value)) {
slf.lock.Lock()
for k := range slf.cache {
delete(slf.cache, k)
}
for _, m := range slf.segments {
m.ClearHandle(handle)
}
slf.lock.Unlock()
}
func (slf *MapSegment[Key, Value]) Range(handle func(key Key, value Value)) {
for _, m := range slf.segments {
m.Range(handle)
}
}
func (slf *MapSegment[Key, Value]) RangeSkip(handle func(key Key, value Value) bool) {
for _, m := range slf.segments {
m.RangeSkip(handle)
}
}
func (slf *MapSegment[Key, Value]) RangeBreakout(handle func(key Key, value Value) bool) {
for _, m := range slf.segments {
if m.rangeBreakout(handle) {
break
}
}
}
func (slf *MapSegment[Key, Value]) RangeFree(handle func(key Key, value Value, skip func(), breakout func())) {
for _, m := range slf.segments {
if m.rangeFree(handle) {
break
}
}
}
func (slf *MapSegment[Key, Value]) Keys() []Key {
var s = make([]Key, 0, len(slf.cache))
slf.lock.RLock()
for k := range slf.cache {
s = append(s, k)
}
defer slf.lock.RUnlock()
return s
}
func (slf *MapSegment[Key, Value]) Slice() []Value {
slf.lock.RLock()
var s = make([]Value, 0, len(slf.cache))
slf.lock.RUnlock()
for _, m := range slf.segments {
s = append(s, m.Slice()...)
}
return s
}
func (slf *MapSegment[Key, Value]) Map() map[Key]Value {
slf.lock.RLock()
var s = map[Key]Value{}
slf.lock.RUnlock()
for _, m := range slf.segments {
for k, v := range m.Map() {
s[k] = v
}
}
return s
}
func (slf *MapSegment[Key, Value]) Size() int {
slf.lock.RLock()
defer slf.lock.RUnlock()
return len(slf.cache)
}
// GetOne 获取一个
func (slf *MapSegment[Key, Value]) GetOne() (value Value) {
for k, s := range slf.cache {
return slf.segments[s].Get(k)
}
return value
}
func (slf *MapSegment[Key, Value]) MarshalJSON() ([]byte, error) {
var ms struct {
Segments map[int]*Map[Key, Value]
Cache map[Key]int
SegmentCount int
}
ms.Segments = slf.segments
ms.Cache = slf.cache
ms.SegmentCount = len(slf.segments)
return json.Marshal(ms)
}
func (slf *MapSegment[Key, Value]) UnmarshalJSON(bytes []byte) error {
var ms struct {
Segments map[int]*Map[Key, Value]
Cache map[Key]int
SegmentCount int
}
if err := json.Unmarshal(bytes, &ms); err != nil {
return err
}
slf.lock.Lock()
slf.consistency = hash.NewConsistency(ms.SegmentCount)
for i := 0; i < ms.SegmentCount; i++ {
slf.consistency.AddNode(i)
}
slf.segments = ms.Segments
slf.cache = ms.Cache
slf.lock.Unlock()
return nil
}