rust学习基于tokio_actor聊天服务器实战(一 )_基于actor模式 实现聊天

前言
tokio是Rust中使用最广泛的异步Runtime，它性能高、功能丰富、便于使用，是使用Rust实现高并发不可不学的一个框架
Actor 背后的基本思想是产生一个独立的任务，该任务独立于程序的其他部分执行某些工作。 通常，这些参与者通过使用消息传递信道与程序的其余部分进行通信。 由于每个 Actor 独立运行，因此使用它们设计的程序自然是并行的。 Actor 的一个常见用法是为 Actor 分配你要共享的某些资源的专有所有权，然后让其他任务通过与 Actor 通信来间接访问彼此的资源。 例如，如果要实现聊天服务器，则可以为每个连接生成一个任务，并在其他任务之间路由一个聊天消息的主任务。 十分有用，因为主任务可以避免必须处理网络IO，而连接任务可以专门处理网络IO；
为什么一定要用actor,这里只是仿照go项目里一部分，go 用的就是actor;

1：环境
rust1.75
ide rustrover64

2:设计及实现
这里使用类似单点登录模式，
useractor
先看go的


一共3个协程/future
接受网络消息 一个协程/future
发送网络消息 一个协程/future
逻辑处理 一个协程/future
协程/future间通信 直接用mpsc

world actor/accmgr 管理useractor 登录，踢人，广播等

一共1个协程/future 处理逻辑消息

rust 版
useractor
说明 receiver: mpsc::UnboundedReceiver, logic future 接受消息并处理
sendclient: mpsc::UnboundedSender 发送消息给 网络future 从而发送给前端
worldsender: mpsc::UnboundedSender, 跟world actor 通信接口

pub enum ActorMessage {
    synmsgwaitrep {
        //同步等待回复
        //需要发送到别处等到别处返回结果，类似于同步操作，只是异步执行的  //oneshot  spsc
        respond_to: crate::synMsgWaitRep, //同步消息
    },
    wtc_userchann {
        respond_to: crate::userChan_WTC, //
    },
    wtc_msg(sendMsgAndType),
    wtc_forwardmsg(sendMsgAndType), //直接转发 data
    ctw_msg(sendMsgAndType),
    ctc_nettologic_msg(sendMsgAndType), //网络消息 to logic
    ctc_logictonet_msg(sendMsgAndType), //logic to net  send
    ctc_signal_event(signalType),
    ctw_signal_event(signalType),
    wtc_signal_event(signalType),
    wtc_getChan_msg(userChannChann),
}
pub struct MyUserActor {
    connid: ConnectID,
    userid: UserID,
    username: String,
    guildid: GuildID,
    userstate: Arc<AtomicU8>,
    receiver: mpsc::UnboundedReceiver<ActorMessage>,
    sendclient: mpsc::UnboundedSender<VU8>,
    worldsender: mpsc::UnboundedSender<ActorMessage>,
    msgmask: u32,
    lasttime: [u32; ChatChannel_Num],
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

world actor
mpscrecv: mpsc::UnboundedReceiver, 接收ActorMessage logic future
chanchan: mpsc::UnboundedReceiver, 接受 ActorMessage2 logic future

pub enum ActorMessage2 {
    synmsgwaitrep {
        //同步等待回复
        //需要发送到别处等到别处返回结果，类似于同步操作，只是异步执行的  //oneshot  spsc
        respond_to: crate::synMsgWaitRep2, //同步消息
    },

    ctw_userhann {
        respond_to: crate::userChan_CTW, //同步消息
    },
}
pub struct userSendChanActorMessage {
    pub(crate) chanchan: Option<mpsc::UnboundedSender<ActorMessage>>,
    pub(crate) username: String,
    pub(crate) userguildid: GuildID,
    pub(crate) connectid: ConnectID,
    pub(crate) chanState: Arc<AtomicU8>, //user 状态
}

pub struct worldActor {
    sharestate: Arc<AtomicU8>,
    mpscrecv: mpsc::UnboundedReceiver<ActorMessage>,
    chanchan: mpsc::UnboundedReceiver<ActorMessage2>,
    usermap: HashMap<UserID, userChan_world>,
    namemap: HashMap<String, UserID>,
    guildmap: HashMap<GuildID, HashSet<UserID>>,
    maxonlinerole: u32,
}

 async fn run(mut self) {
        // let logic_handle = self.handle_logic(recv);
        loop {
            tokio::select! {
                recvmsg= self.mpscrecv.recv()=> {
                    if let Some(actmsg) = recvmsg {
                        self.handle_logic(actmsg).await ;
                    }
                }
                recvmsgchan= self.chanchan.recv()=>{
                    if let Some(actmsg) = recvmsgchan {
                        self.handle_logic2(actmsg).await ;
                    }
                }
                _=tokio::time::sleep(Duration::from_millis(1000*8)) =>{

                }
            }
        } //end loop
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

同步的方式的异步 go 很简单， rust go 上多一点点
go

rust

网络跟逻辑分开，这样 挤号，只需要把 logic future 里 sendclient mpsc 更新， 把网络 to logic mpsc 更新 及一些 状态重置下 即可，无需重新加载现有useractor 里的信息
类试单点登录 对于聊天服务器来说 ，只需要 角色进入后，由logic服 ase 对称加密(密钥及盐，logic 服 chat 服 共享/配置，共享方式自行决定）或 非对称(ECC) 等都可以，加密的token 由前端发送 给chat 服，chat 解密 得到 相应信息 并验证有效性 参考加解密验证用户的合法性

3:测试
前端简单用go 写了个

var origin = "http://192.168.1.32:8080"
var url = "wss://192.168.1.32:8080/websocket"
func GetProtoMsgID(data []byte) uint32 {
	var sMsgID uint16 = uint16(uint8(data[3] & 0x7f))
	if (uint8(data[3]) & 0x80) > 0 {
		sMsgID += (uint16(data[4]) & 0x7f) << 7
	}
	return uint32(sMsgID)
}

func  sendMsg(ws *websocket.Conn,pb proto.Message) {
	if ws != nil {
		if data, err2 := proto.Marshal(pb); err2 != nil {
			log.Printf("SendMessage pb=%v err2=%v \n", pb, err2)
		} else {
			if err4 := websocket.Message.Send(ws, data); err4 != nil {
				log.Printf("send error =%v \n", err4)
			}
		}
	}
}

func doLogicMsg(data []byte)  {
	msgId := GetProtoMsgID(data)
	fmt.Printf("msgid=%v",msgId)
	switch msgId {
	case uint32(chatproto.CHATMSG_CHC_Login_Rep):
		{
			loginReq := &chatproto.ChatMessageLoginRep{}
			if err := proto.Unmarshal(data, loginReq); err != nil {

			} else {
				fmt.Printf("CHATMSG_CHC_Login_Rep =%v \n",loginReq.Res)
			}
		}
	case uint32(chatproto.CHATMSG_CCH_Chat_Rep):
		{
			chatrep := &chatproto.ChatMessageChatRep{}
			if err := proto.Unmarshal(data, chatrep); err != nil {

			} else {
				fmt.Printf("CHATMSG_CCH_Chat_Rep =%v \n",chatrep.Res)
			}
		}
	case uint32(chatproto.CHATMSG_CHC_Notify_Chat):
		{
			chatmsg := &chatproto.ChatMessageNotifyChat{}
			if err := proto.Unmarshal(data, chatmsg); err != nil {

			} else {
				fmt.Printf("CHATMSG_CHC_Notify_Chat =%v fromuserid=%v text=%v \n",chatmsg.Chattype,chatmsg.Senderid,chatmsg.Strcontext)
			}
		}
	}

}
func getTimestamp() uint32 {
	return  uint32(time.Now().UTC().Unix());
}

func main(){
	//if os.Args[0]
	userid :=  getTimestamp()
	guildid := uint32(0)
	if len(os.Args) > 1 {
		if s,e := strconv.Atoi(os.Args[1]);e ==nil {
			userid = uint32(s)
		}
	}
	if len(os.Args) > 2 {
		if s,e := strconv.Atoi(os.Args[2]);e ==nil {
			guildid = uint32(s)
		}
	}

	ws, err := websocket.Dial(url, "", origin)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("userid=%v guild=%v \n",userid,guildid)
	{
		msg := new(chatproto.ChatMessageLoginReq)
		msg.Msghead = &chatproto.ChatMessageHead{uint32(chatproto.CHATMSG_CCH_Login_Req), 1}
		msg.Userid = userid
		msg.Username = "name_"+strconv.Itoa(int(userid))
		msg.Guildid = guildid
		msg.Tokenmd5 = "md5"
		msg.Tokenstr = "Tokenstr"

		sendMsg(ws, msg)
	}
	disflag :=  false
	{
		go func() {
			for{
				buf := make([]byte, 1024*4)
				err := websocket.Message.Receive(ws, &buf)
				if err != nil {
					//log.Printf("websocket.Message.Receive err=%v  ---%s\n", err,self.getAccName())
					disflag = true
					return
				}

				if len(buf) >= 4 {

					doLogicMsg(buf)

					//self.msgQue.PostUserMessage(&ReceiveNetMsg{buf})
				} else {
					log.Printf("[error]recv data=%v \n", buf)
					return
				}
			}

		}()

	}
	time.Sleep(time.Second*3)

	//pub enum  ChatChannel{
	//	ChatChannel_NONE=0,
	//	ChatChannel_NORMAL,
	//	ChatChannel_GUILD,
	//	ChatChannel_WORLD,
	//	ChatChannel_ALL,
	//}


	{
		sendcount := uint32(1)
		num := uint32(0)
		msg := new(chatproto.ChatMessageChatReq)
		msg.Msghead = &chatproto.ChatMessageHead{uint32(chatproto.CHATMSG_CCH_Chat_Req), 1}
		msg.Chattype = 1
		msg.Context ="normal chat "+ strconv.Itoa(int(num))

		for {
			if disflag { //脏数据
				break
			}
			sendMsg(ws, msg)
			time.Sleep(time.Second*10)
			num++
			m := num % 3 +1
			msg.Chattype = uint32(m)
			msg.Context ="normal chat "+ strconv.Itoa(int(sendcount))
			fmt.Printf("[%v][%v] send chattype=%v \n",sendcount,getTimestamp(),msg.Chattype)
			sendcount++
			//if m == 3  {
			//	time.Sleep(time.Second*10)
			//}
		}
	}

	ws.Close()//关闭连接
	fmt.Printf("client exit\n")
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

相互挤号测试

4:DEMO工程 后续完善了如有需要再上传(当前只能说基本上跑起来)
如果觉得有用，麻烦点个赞，加个收藏