rust学习基于tokio_actor聊天服务器实战(一 )_业界新闻

发布时间:2024-07-28 18:29

阅读量:0

前言
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], }

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     }

同步的方式的异步 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") }