当前位置： 首页 > news >正文

龙岗区做网站全国唯一一个没有疫情的城市

news 2025/7/13 3:49:58
龙岗区做网站,全国唯一一个没有疫情的城市,丹阳房产网,常州免费做网站tcp模块 定义相关类 Client：表示客户端连接，包含网络连接conn、指向服务器的指针Server和Channel指针c。server：表示TCP服务器，包含服务器地址address、TLS配置config以及三个回调函数： onNewClientCallback&#xf…

tcp模块

定义相关类

  • Client：表示客户端连接，包含网络连接conn、指向服务器的指针Server和Channel指针c。
  • server：表示TCP服务器，包含服务器地址address、TLS配置config以及三个回调函数：
    • onNewClientCallback：新客户端连接时调用。
    • onClientConnectionClosed：客户端连接关闭时调用。
    • onNewMessage：客户端接收新消息时调用。

客户端相关接口

  • Client.listen()：客户端监听方法，读取连接数据，调用onNewMessage回调。
  • Client.Send(message string)：发送文本消息给客户端。
  • Client.SendBytes(b []byte)：发送字节数据给客户端。
  • Client.Conn()：获取客户端的网络连接。
  • Client.Close()：关闭客户端的网络连接。

服务器相关接口

  • server.OnNewClient(callback func(c *Client))：设置新客户端连接的回调。
  • server.OnClientConnectionClosed(callback func(c *Client, err error))：设置客户端连接关闭的回调。
  • server.OnNewMessage(callback func(c *Client, message []byte, size int))：设置客户端接收新消息的回调。
  • server.Listen()：启动网络服务器，监听连接。

服务器初始化

  • NewTcpServer(address string) *server：创建新的TCP服务器实例，不使用TLS。
  • NewTcpServerWithTLS(address, certFile, keyFile string) *server：创建带有TLS功能的TCP服务器实例。

服务器启动流程

  1. 使用NewTcpServer或NewTcpServerWithTLS创建服务器实例。
  2. 设置回调函数，响应新客户端连接、客户端连接关闭和接收新消息事件。
  3. 调用server.Listen()开始监听连接。

TLS支持

  • 如果需要TLS，使用NewTcpServerWithTLS函数，提供证书和密钥文件路径。

参考demo

package MediasoupLibimport ("bufio""time""crypto/tls""log""net"
)// Client holds info about connectiontype Client struct {conn net.ConnServer *serverc *Channel
}// TCP server
type server struct {address string // Address to open connection: localhost:9999config *tls.ConfigonNewClientCallback func(c *Client)onClientConnectionClosed func(c *Client, err error)onNewMessage func(c *Client, message []byte, size int)
}// Read client data from channel
func (c *Client) listen() {fmt.Printf("tcp client listen() ")c.Server.onNewClientCallback(c)reader := bufio.NewReader(c.conn)for {recv := make([]byte, 1500) //MTU 1500size, err := reader.Read(recv)if err != nil {c.conn.Close()c.Server.onClientConnectionClosed(c, err)fmt.Printf("tcp client close! %s", err.Error())return}if size == 0 {time.Sleep(time.Millisecond * 250)fmt.Printf("tcp client recv size=0")continue}recv = recv[0:size]c.Server.onNewMessage(c, recv, size)}
}// Send text message to client
func (c *Client) Send(message string) error {return c.SendBytes([]byte(message))
}// Send bytes to client
func (c *Client) SendBytes(b []byte) error {_, err := c.conn.Write(b)if err != nil {c.conn.Close()c.Server.onClientConnectionClosed(c, err)}return err
}func (c *Client) Conn() net.Conn {return c.conn
}func (c *Client) Close() error {return c.conn.Close()
}// Called right after server starts listening new client
func (s *server) OnNewClient(callback func(c *Client)) {s.onNewClientCallback = callback
}// Called right after connection closed
func (s *server) OnClientConnectionClosed(callback func(c *Client, err error)) {s.onClientConnectionClosed = callback
}// Called when Client receives new message
func (s *server) OnNewMessage(callback func(c *Client, message []byte, size int)) {s.onNewMessage = callback
}// Listen starts network server
func (s *server) Listen() {var listener net.Listenervar err errorif s.config == nil {listener, err = net.Listen("tcp", s.address)} else {listener, err = tls.Listen("tcp", s.address, s.config)}if err != nil {fmt.Printf("Error starting TCP server.\r\n", err)}defer listener.Close()for {conn, err := listener.Accept()if err != nil {fmt.Printf("tcpserver listner Accept error:%s", err.Error())}client := &Client{conn: conn,Server: s,}go client.listen()}
}// Creates new tcp server instance
func NewTcpServer(address string) *server {fmt.Printf("Creating server with address %s", address)server := &server{address: address,}server.OnNewClient(func(c *Client) {c.c = NewChannel(c)})server.OnNewMessage(func(c *Client, message []byte, size int) {c.c.onData(c, message, size)})server.OnClientConnectionClosed(func(c *Client, err error) {c.c.Close()fmt.Printf("OnClientConnectionClosed   err = %s", err.Error())})return server
}func NewTcpServerWithTLS(address, certFile, keyFile string) *server {cert, err := tls.LoadX509KeyPair(certFile, keyFile)if err != nil {fmt.Printf("Error loading certificate files. Unable to create TCP server with TLS functionality.\r\n", err)}config := &tls.Config{Certificates: []tls.Certificate{cert},}server := NewTcpServer(address)server.config = configreturn server}

channel模块

ChannelListener接口定义:

  • ChannelListener：定义了一个接口，包含两个方法OnChannelEvent和OnChannelStringEvent，用于监听通道事件。

Channel结构体:

  • 包含字段如MediasoupClient（指向Client的指针）、PendingSent（一个同步映射，用于存储待发送的数据）、LastBinaryNotification、ID、Pid、udpAddress、udpPort、queue和messageQueue（两个循环队列）、Num和Listeners（一个映射，存储监听器）。
  • 包含一个互斥锁mutex，用于并发控制。

Channel的接口:

  • NewChannel：构造函数，创建并返回一个新的Channel实例。
  • AddListener和RemoveListener：用于添加和移除监听器。
  • processMessage：处理接收到的消息。
  • onData：处理接收到的数据。
  • handle：一个循环，从队列中取出项目并处理。
  • handleMessage：处理消息队列中的消息。
  • process：根据消息类型进行不同的处理。
  • Close：关闭通道，清理资源。
  • Request：发送请求并返回一个通道用于接收响应。
  • SetUdp：设置UDP地址和端口。

并发处理:

  • 使用sync.Map和sync.RWMutex来处理并发，确保数据的一致性和线程安全。

循环队列:

  • 使用MeetGo.CycleQueue作为循环队列，用于存储消息和数据。

参考demo

import ("encoding/json""fmt""strconv""sync""time""strings"
)//###########SendRequest begin############/
var REQUEST_TIMEOUT = 30000type SendRequest struct {ID       stringMethod   stringInternal map[string]interface{}Data     map[string]interface{}
}//async chan SendReponse
type SendReponse struct {ID       intTargetId intEvent    stringAccepted boolRejected boolInternal map[string]interface{}Data     map[string]interface{}Reason   stringBinary   bool
}type AsyncSingal struct {Async chan SendReponse
}
###########SendRequest End############/////###########CycleQueue begin############//
type CycleQueue struct {data  []interface{} //存储空间front int           //前指针,前指针负责弹出数据移动rear  int           //尾指针,后指针负责添加数据移动cap   int           //设置切片最大容量
}func NewCycleQueue(cap int) *CycleQueue {return &CycleQueue{data:  make([]interface{}, cap),cap:   cap,front: 0,rear:  0,}
}//入队操作
//判断队列是否队满,队满则不允许添加数据
func (q *CycleQueue) Push(data interface{}) bool {//check queue is fullif (q.rear+1)%q.cap == q.front { //队列已满时，不执行入队操作return false}q.data[q.rear] = data         //将元素放入队列尾部q.rear = (q.rear + 1) % q.cap //尾部元素指向下一个空间位置,取模运算保证了索引不越界（余数一定小于除数）return true
}//出队操作
//需要考虑: 队队为空没有数据返回了
func (q *CycleQueue) Pop() interface{} {if q.rear == q.front {return nil}data := q.data[q.front]q.data[q.front] = nilq.front = (q.front + 1) % q.capreturn data
}//因为是循环队列, 后指针减去前指针 加上最大值, 然后与最大值 取余
func (q *CycleQueue) QueueLength() int {return (q.rear - q.front + q.cap) % q.cap
}func (q *CycleQueue) FindDataByRequestId(requestId string) string {for i := 0; i < q.QueueLength(); i++ {if strings.Count(q.data[i].(string), requestId) == 1 {emitData := q.data[i].(string)q.data = append(q.data[:i], q.data[i+1:]...)return emitData}}return ""
}
///###########CycleQueue############import ("encoding/json""fmt""strconv""sync""time"MeetGo "vrv.meeting.server/MeetGo"
)const NS_MAX_SIZE int = 655350var messageBuffer = make([]byte, NS_MAX_SIZE)
var messageIndex = 0type ChannelListener interface {OnChannelEvent(string, map[string]interface{})OnChannelStringEvent(string, string)
}type Channel struct {MediasoupClient        *ClientPendingSent            sync.MapLastBinaryNotification interface{}ID                     intPid                    intudpAddress             stringudpPort                intqueue                  CycleQueuemessageQueue          CycleQueueNum                    intListeners              map[string]ChannelListenermutex                  sync.RWMutex
}func NewChannel(tcpClient *Client) *Channel {channel := new(Channel)channel.MediasoupClient = tcpClientchannel.queue = MeetGo.NewCycleQueue(1000)channel.messageQueue = MeetGo.NewCycleQueue(10000)channel.Num = 0channel.Listeners = make(map[string]ChannelListener, 100)go channel.handle()go channel.handleMessage()return channel
}func (channel *Channel) AddListener(id string, listener ChannelListener) {channel.mutex.Lock()channel.Listeners[id] = listenerchannel.mutex.Unlock()
}func (channel *Channel) RemoveListener(id string) {channel.mutex.Lock()delete(channel.Listeners, id)channel.mutex.Unlock()
}func (channel *Channel) processMessage(message string) {jsonMessage := make(map[string]interface{})err := json.Unmarshal([]byte(message), &jsonMessage)if err != nil {MeetGo.Log.Error("Channel processMessage error:%s", err.Error())return}if jsonMessage["registId"] != nil {MeetGo.Log.Debug("client registId succeeded [id:%s]", jsonMessage["registId"].(string))channel.ID, _ = strconv.Atoi(jsonMessage["registId"].(string))channel.Pid = int(jsonMessage["pid"].(float64))Global_Worker.OnMediasoupWorkerOnline(channel.ID, channel, jsonMessage["registId"].(string))} else if jsonMessage["id"] != nil {idd := int(jsonMessage["id"].(float64))value, ret := channel.PendingSent.Load(idd)if !ret {fmt.Printf("received Response does not match any sent Request")return}channel.PendingSent.Delete(idd)asyncReponse := value.(*MeetGo.AsyncSingal)if jsonMessage["accepted"] != nil && jsonMessage["accepted"].(bool) {MeetGo.Log.Debug("request succeeded [id:%d]", int(jsonMessage["id"].(float64)))sendReponse := MeetGo.SendReponse{ID:       idd,Accepted: jsonMessage["accepted"].(bool),Data:     jsonMessage["data"].(interface{}).(map[string]interface{}),}asyncReponse.Async <- sendReponse} else {MeetGo.Log.Debug("request failed [id:%d, reason: %s]", int(jsonMessage["id"].(float64)), jsonMessage["reason"].(string))sendReponse := MeetGo.SendReponse{ID:     int(jsonMessage["id"].(float64)),Reason: jsonMessage["reason"].(string),}asyncReponse.Async <- sendReponse}} else if jsonMessage["targetId"] != nil && jsonMessage["event"] != nil {if jsonMessage["binary"] != nil {channel.LastBinaryNotification = jsonMessagereturn} else if jsonMessage["data"] != nil {listenerKey := fmt.Sprintf("%d", int(jsonMessage["targetId"].(float64)))channel.mutex.RLock()listener := channel.Listeners[listenerKey]channel.mutex.RUnlock()if listener != nil {listener.OnChannelEvent(jsonMessage["event"].(string), jsonMessage["data"].(map[string]interface{}))}} else {data := make(map[string]interface{})listenerKey := fmt.Sprintf("%d", int(jsonMessage["targetId"].(float64)))channel.mutex.RLock()listener := channel.Listeners[listenerKey]channel.mutex.RUnlock()if listener != nil {listener.OnChannelEvent(jsonMessage["event"].(string), data)}}} else {fmt.Printf("received message is not a Response nor a Notification")return}
}func (channel *Channel) onData(client *Client, message []byte, size int) {for {ret := channel.messageQueue.Push(message)if ret {break} else {time.Sleep(40 * time.Millisecond)}}
}func (channel *Channel) handle() {for {item := channel.queue.Pop()if item == nil {time.Sleep(40 * time.Millisecond)continue}channel.process(item)time.Sleep(1 * time.Millisecond)}
}func (channel *Channel) handleMessage() {ns := NetString{bufLen: 0, length: 0, state: 0}for {item := channel.messageQueue.Pop()if item == nil {time.Sleep(40 * time.Millisecond)continue}message := item.([]byte)var nsPayloads [][]byteerr := ns.NsUnmarshal(message, &nsPayloads)if err != nil {fmt.Printf("Channel handleMessage nsPayload error %s", err.Error())return}for _, nsPayload := range nsPayloads {channel.queue.Push(nsPayload)}time.Sleep(1 * time.Millisecond)}
}func (channel *Channel) process(data interface{}) {nsPayload := data.([]byte)if channel.LastBinaryNotification == nil {switch nsPayload[0] {// 123 = '{' (a Channel JSON messsage).case 123:channel.processMessage(string(nsPayload))break// 68 = 'D' (a debug log).case 68:fmt.Printf(string(nsPayload))break// 87 = 'W' (a warning log).case 87:fmt.Printf(string(nsPayload))break// 69 = 'E' (an error log).case 69:fmt.Printf(string(nsPayload))breakdefault:fmt.Printf("unexpected data: %s", string(nsPayload))}} else {msg := channel.LastBinaryNotificationchannel.LastBinaryNotification = niljsonMsg := make(map[string]interface{})err := json.Unmarshal([]byte(msg.(string)), &jsonMsg)if err != nil {panic(err)}listenerKey := fmt.Sprintf("%d", int(jsonMsg["targetId"].(float64)))channel.mutex.RLock()listener := channel.Listeners[listenerKey]channel.mutex.RUnlock()if listener != nil {listener.OnChannelStringEvent(jsonMsg["event"].(string), jsonMsg["data"].(string))}}
}func (channel *Channel) Close() {channel.PendingSent.Range(func(k, v interface{}) bool {channel.PendingSent.Delete(k)return true})registId := strconv.Itoa(channel.ID)Global_Worker.OnMediasoupWorkerOffline(registId)time.Sleep(time.Millisecond * 250) //?fmt.Printf("channel.MediasoupClient.Close() ")channel.MediasoupClient.Close()
}func (c *Channel) Request(method string, internal,data map[string]interface{}) (chan MeetGo.SendReponse, int64) {id := RandomNumberGenerator(10000000, 99999999)fmt.Printf("MediasoupLib Channel [method:%s, id:%d]", method, id)request := MeetGo.RequestJson{ID:       id,Method:   method,Internal: internal,Data:     data,}requestJson := request.Encode()requestSend := nsEncode(requestJson)fmt.Printf("___requestSend : %s", requestSend)sendReponse := new(MeetGo.AsyncSingal)sendReponse.Async = make(chan MeetGo.SendReponse)if sendReponse != nil {c.PendingSent.Store(int(id), sendReponse)}defer c.MediasoupClient.Send(requestSend)return sendReponse.Async, id
}func (channel *Channel) SetUdp(udpAddress string, udpPort int) {channel.udpAddress = udpAddresschannel.udpPort = udpPort
}
http://www.dinnco.com/news/42823.html

相关文章：

  • 养老网站建设公司品牌营销策划
  • 企业建站系统是什么拓客平台有哪些
  • wordpress下载不了seo是什么的
  • 用php做美食网站深圳推广系统
  • 武汉外贸网站制作百度安装
  • 鹤壁做网站价格品牌型网站设计推荐
  • 网站标题关键词描述莱阳seo排名
  • 蜘蛛爬取网站口碑营销属于什么营销
  • 做聚会的网站一个关键词要刷多久
  • 万维网网站注册哈尔滨最新疫情通报
  • 临沧网站开发南昌seo排名公司
  • 微信商城小程序搭建教程网络优化公司哪家好
  • 信息服务类网站怎么做seo外链招聘
  • 做行业网站赚钱吗seo按照搜索引擎的什么对网站
  • 资阳seo公司广州百度推广排名优化
  • 搞笑视频网站建设策划书深圳营销推广引流公司
  • WordPress可以做社交网站嘛石家庄谷歌seo
  • 一个网站能卖多少钱?网站怎么收录
  • 网络推广及网站建设合作协议排名第一的手机清理软件
  • 佛山网站优化步骤网络营销公司如何建立
  • 导购分享网站模板seo什么意思简单来说
  • 网站优化意见百度热搜榜小说排名
  • 广西建设协会网站首页营销文案
  • 网站被攻击怎么让百度重新蜘蛛自动抓上海网站seo策划
  • 网站架构计划书百度官方推广平台
  • 做教育网站用什么颜色平台推广引流
  • 南阳专业网站排名推广基本seo
  • 上海网络推广外包广告seo是什么意思
  • 哈尔滨专业网站建设中国舆情在线
  • 遵义做网站的网络营销平台排名