rust学习基于tokio_actor聊天服务器实战(一 )

2024-03-18 服务器 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")}