前言
多线程/进程编程是每个程序员的基本功，同时也是开发中的难点，处理各种“锁”的问题是让人十分头痛的一件事。
Actor模型，在1973由Carl Hewitt定义，被Erlang OTP推广，其消息传递更加符合面向对象的原始意图。Actor模型属于并发组件模型，通过组件方式定义并发编程范式的高级阶段，避免使用者直接接触多线程并发或线程池等基础概念。
Actor模型的基础就是消息传递，一个Actor模型可以认为是一个基本的计算单元，它能接收消息并基于消息执行运算，也可以发送消息给其他Actor模型。各个Actor模型之间相互隔离，不共享内存。
Actor模型本身封装了状态和行为，在进行并发编程时，Actor模型只需要关注消息和其本身。而消息是一个不可变对象，所以Actor模型不需要去关注锁和内存原子性等一系列多线程常见的问题。

1:上代码

核心代码
MySynMsgChan 同步时使用

package myqueueimport ("gameserver/myqueue/queue/mpsc""runtime""sync/atomic""log""time"
)const (idle int32 = iotarunning
)// //
type Dispatcher interface {Schedule(fn func())Throughput() int
}type goroutineDispatcher intfunc (goroutineDispatcher) Schedule(fn func()) {go fn()
}func (d goroutineDispatcher) Throughput() int {return int(d)
}func NewDefaultDispatcher(throughput int) Dispatcher {return goroutineDispatcher(throughput)
}type synchronizedDispatcher intfunc (synchronizedDispatcher) Schedule(fn func()) {fn()
}func (d synchronizedDispatcher) Throughput() int {return int(d)
}func NewSynchronizedDispatcher(throughput int) Dispatcher {return synchronizedDispatcher(throughput)
}/type Statistics interface {MessageReceived(message interface{})
}type MyQueue struct {userMailbox     *mpsc.QueueschedulerStatus int32userMessages    int32suspended       booldispatcher      DispatchermailboxStats    []StatisticsbStop           bool
}func (m *MyQueue) PostUserMessage(message interface{}) {if m.isStop() {return}//	for _, ms := range m.mailboxStats {//		ms.MessagePosted(message)//	}m.userMailbox.Push(message)atomic.AddInt32(&m.userMessages, 1)//m.schedule()if atomic.CompareAndSwapInt32(&m.schedulerStatus, idle, running) {m.dispatcher.Schedule(m.processMessages)}
}func (m *MyQueue) processMessages() {process:m.run()// set mailbox to idleatomic.StoreInt32(&m.schedulerStatus, idle)user := atomic.LoadInt32(&m.userMessages)// check if there are still messages to process (sent after the message loop ended)if user > 0 {// try setting the mailbox back to runningif atomic.CompareAndSwapInt32(&m.schedulerStatus, idle, running) {//	fmt.Printf("looping %v %v %v\n", sys, user, m.suspended)goto process}}}func (m *MyQueue) run() {var msg interface{}//	var bok booldefer func() {if r := recover(); r != nil {log.Printf("(m *MyQueue) run() err=%v  msg=%#v \n", r, msg)}}()i, t := 0, m.dispatcher.Throughput()for {if i > t {i = 0runtime.Gosched()}i++if msg = m.userMailbox.Pop(); msg != nil {atomic.AddInt32(&m.userMessages, -1)if len(m.mailboxStats) > 0 && m.mailboxStats[0] != nil {m.mailboxStats[0].MessageReceived(msg)}} else {return}}
}func (m *MyQueue) Stop() {if !m.bStop {m.bStop = true}
}// 
func (m *MyQueue) isStop() bool {return m.bStop
}func UnboundedMyQueue(dispatcher Dispatcher, mailboxStats ...Statistics) *MyQueue {return &MyQueue{userMailbox:  mpsc.New(),mailboxStats: mailboxStats,dispatcher:   dispatcher,}
}type MySynMsgChan struct {MsgData   interface{}MsgChan   chan interface{}ChanState *int32
}func GenerateMySynMsg(data interface{}) *MySynMsgChan {chanState := new(int32)*chanState = 1return &MySynMsgChan{data, make(chan interface{}, 1), chanState}
}func (self *MySynMsgChan) WaitSynReply(d time.Duration) (interface{}, bool) {if d < time.Millisecond {d = time.Millisecond}timetick := time.NewTimer(d)for {select {case <-timetick.C:atomic.StoreInt32(self.ChanState, 0)return nil, falsecase msg, ok := <-self.MsgChan:atomic.StoreInt32(self.ChanState, 0)timetick.Reset(0)if ok {return msg, true}}}return nil, false
}

使用也很简单
实现 MessageReceived 接口

func CeateUserActor(ws *websocket.Conn, wsIndex uint32, accname, headrul string, platformId, serverid uint32) {useractor := new(UserActor)useractor.WsConn = wsuseractor.shareUserState = new(shr.ShareState)useractor.shareUserState.State = 1useractor.remoteip = RetmoteIp(ws.RemoteAddr().String())......................................................useractor.wsConnIndex = wsIndexuseractor.u32State = STATE_READY_RECV_LOGINuseractor.CurMaxItemID = base.ITEM_ID_MIN_NUM //默认开始值useractor.CurMaxHeroID = base.HERO_ID_MIN_NUM //默认开始值useractor.msgQue = myqueue.UnboundedMyQueue(myqueue.NewDefaultDispatcher(64), useractor)useractor.commonBagItemMap = map[uint64]uint64{}useractor.endTimeItemMap = treemap.NewWith(treeutils.UInt64Comparator)useractor.msgQue.PostUserMessage(&shr.NewSessionQueryUserInfo{})SetUpRecv(ws, useractor.msgQue, wsIndex, accname, serverid)
}
..........................
func (self *UserActor) MessageReceived(r interface{}) {if self.u32State == STATE_CLOSE {return}switch r.(type) {case *shr.NewSessionQueryUserInfo: //新连接查询self.doNewSessionQueryPlayer()case *shr.ClientRecvNetMsg: //protobufif !self.doClienNetData(r.(*shr.ClientRecvNetMsg).Clientdata) {self.doDisconnect(self.wsConnIndex)}case *shr.NetDisconnect: //网络断开dis := r.(*shr.NetDisconnect)self.doDisconnect(dis.DisIndex)...................................................
}//
..........................发送消息 会放入mailbox里
func SendMsgToAccMgr(v interface{}) {*****.msgque.PostUserMessage(v)
}

同步等待回复

2:测试，这个已经在项目上用过了，这里不专门再写DEMO测试

3：如果觉得有用，麻烦点个赞，加个收藏