Go语言结合grpc和protobuf实现去中心化的聊天室
作者：AlpsMonaco
介绍
传统的聊天室主要是基于c/s架构，需要有一个服务端完成各个客户端的聊天转发。今天我们使用golang+grpc+protobuf，设计一个去中心化、局域网自发现的聊天客户端。
完整代码地址在 github.com/AlpsMonaco/proximity-chat
模块
协议
我们先定义proto消息格式 message/message.proto
syntax = "proto3";
option go_package = "proximity-chat/message";
package message;
service Chat {
rpc NewNode (stream NodeRequest) returns (stream NodeReply){ }
}
message NodeRequest {
string msg = 1;
}
message NodeReply {
string msg = 1;
}
聊天软件一般需要全双工保证时效性，所以这边使用了 stream NodeRequest 和 stream NodeReply。 这边消息只有两个，请求和回复直接透传string就行。
执行
protoc --go_out=. --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative message\message.proto
会在相同目录下生成相关的go代码文件。在文件 message_grpc.pb.go 中会包含rpc的interface
type ChatServer interface {
NewNode(Chat_NewNodeServer) error
mustEmbedUnimplementedChatServer()
}
我们需要实现这个接口中的 NewNode 服务。
交互
在 service/message.go 中实现 NewNode(Chat_NewNodeServer) error
type MessageWriter interface {
Write(string)
}
type Message struct {
Writer MessageWriter
message.UnimplementedChatServer
}
...
func (m *Message) NewNode(ss message.Chat_NewNodeServer) error {
head, err := ss.Recv()
if err != nil {
m.Writer.Write(fmt.Sprint(err))
return err
}
addr := head.GetMsg()
if controller.IsChatNodeExist(addr) {
return nil
}
if !controller.AddChatNode(&ServerChatNode{s: ss}, addr) {
return nil
}
err = ss.Send(&message.NodeReply{Msg: "ok"})
if err != nil {
return err
}
m.Writer.Write("new node " + addr + " has joined")
for {
msg, err := ss.Recv()
if err != nil {
controller.RemoveNode(addr)
fmt.Println(err)
return err
}
m.Writer.Write(msg.GetMsg())
}
}由于是去中心化，所以没有客户端服务端的概念，我们将它称为一个节点 node。在同一个局域网内，node监听的ip+port做唯一key，用于避免重复进入聊天室。
上面的代码中 controller 模块主要是用来控制和管理断点的，后续会讲。
整体流程是先接收其他node发来的 ip+port ，判断是否已经加入过这个端点，如果没加入过就用controller绑定节点，进行后续的聊天请求，否则中止交互。
控制
在 controller/node.go ，我们使用map和读写锁来维护node的唯一性。
package controller
import (
"sync"
)
type ChatNode interface {
SendChatMsg(string) error
RecvChatMsg() (string, error)
}
var nodeMap map[string]ChatNode = make(map[string]ChatNode)
var nodeMapLock sync.RWMutex
func AddChatNode(node ChatNode, addr string) bool {
nodeMapLock.Lock()
defer nodeMapLock.Unlock()
_, ok := nodeMap[addr]
if !ok {
nodeMap[addr] = node
return true
}
return false
}
func RemoveNode(addr string) {
nodeMapLock.Lock()
defer nodeMapLock.Unlock()
delete(nodeMap, addr)
}
func IsChatNodeExist(addr string) bool {
nodeMapLock.RLock()
defer nodeMapLock.RUnlock()
_, ok := nodeMap[addr]
return ok
}
func Publish(s string) {
nodeMapLock.RLock()
defer nodeMapLock.RUnlock()
for _, n := range nodeMap {
n.SendChatMsg(s)
}
}发现
discover/discover.go 下定义如何发现相同网段上的其他服务。
这边使用 ipnetgen 库来获取相同网段下的所有IP。定期去遍历其他网段上的相同服务。 将自己的监听ip+端口发送给其他node，若返回'ok'则建立通讯。
func BeginDiscoverService() {
minPort := config.GetConfig().GetMinPort()
maxPort := config.GetConfig().GetMaxPort()
if minPort > maxPort {
minPort = maxPort
}
for {
time.Sleep(time.Second)
gen, err := ipnetgen.New(config.GetConfig().GetCIDR())
if err != nil {
panic(err)
}
for ip := gen.Next(); ip != nil; ip = gen.Next() {
for i := minPort; i <= maxPort; i++ {
addr := fmt.Sprintf("%s:%d", ip.String(), i)
if addr == GetAddr() {
continue
}
if controller.IsChatNodeExist(addr) {
continue
}
conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
fmt.Printf("did not connect: %v\n", err)
continue
}
client := message.NewChatClient(conn)
cli, err := client.NewNode(context.Background())
if err != nil {
continue
}
err = cli.Send(&message.NodeRequest{Msg: GetAddr()})
if err != nil {
writer.Write(fmt.Sprint(err))
continue
}
resp, err := cli.Recv()
if err != nil {
cli.CloseSend()
writer.Write(fmt.Sprint(err))
continue
}
if resp.GetMsg() != "ok" {
cli.CloseSend()
continue
}
if !controller.AddChatNode(&service.ClientChatNode{C: cli}, addr) {
cli.CloseSend()
continue
}
writer.Write("discover " + addr)
go func() {
for {
msg, err := cli.Recv()
if err != nil {
writer.Write(fmt.Sprint(err))
controller.RemoveNode(addr)
return
}
writer.Write(msg.GetMsg())
}
}()
}
}
}
}配置
我们定义配置的获取方式，配置文件格式为json，定义配置获取的方式 config.go 。
package config
type NetworkConfig struct {
CIDR string `json:"cidr"`
MaxPort int `json:"max_port"`
MinPort int `json:"min_port"`
}
func DefaultNetworkConfig() *NetworkConfig {
return &NetworkConfig{
"127.0.0.1/32", 4569, 4565,
}
}
type ConstNetworkConfig struct {
c *NetworkConfig
}
func (c *ConstNetworkConfig) GetCIDR() string { return c.c.CIDR }
func (c *ConstNetworkConfig) GetMaxPort() int { return c.c.MaxPort }
func (c *ConstNetworkConfig) GetMinPort() int { return c.c.MinPort }
var config = &ConstNetworkConfig{DefaultNetworkConfig()}
func GetConfig() *ConstNetworkConfig { return config }
func SetConfig(nc *NetworkConfig) { config = &ConstNetworkConfig{nc} }这边最主要定义三个字段，内网的ip网段，服务的最小到最大的端口范围。这个配置主要用于搜寻同网段同端口上的相同服务。为了方便调试我们加一个 DefaultNetworkConfig()，监听127.0.0.1上的4565~4569。 同时还加了一个 ConstNetworkConfig 类，供其他模块访问全局配置，同时保护配置不被修改。
运行实例
编译后直接运行，会在指定的端口范围内尝试监听，无需指定端口。主线程中scanf阻塞获取输入。我们直接打开三个进程，在一个终端中输入数据发送，其他两个终端都能获取聊天数据。
以上就是Go语言结合grpc和protobuf实现去中心化的聊天室的详细内容，更多关于Go聊天室的资料请关注脚本之家其它相关文章！