网站开发和移动开发,中国环球贸易网,新闻热点事件素材,个人主页设计实验报告文章目录 文章概要需求分析技术栈准备工作封装日志框架封装Redis封装ETCD 业务逻辑开发定义模型定义服务实现服务接口服务端启动程序客户端测试程序 总结 文章概要
本篇文章分享在做一个IM(即时通讯)系统时#xff0c;设计一个管理用户在线状态的服务。
该系列文章将从以下几… 文章目录 文章概要需求分析技术栈准备工作封装日志框架封装Redis封装ETCD 业务逻辑开发定义模型定义服务实现服务接口服务端启动程序客户端测试程序 总结 文章概要
本篇文章分享在做一个IM(即时通讯)系统时设计一个管理用户在线状态的服务。
该系列文章将从以下几个方面进行介绍
状态服务的设计需求分析、采用哪些技术栈各个需求功能的设计和实现运行效果
需求分析
顾名思义状态服务器是用来记录和管理用户在线状态的该系统中简化了用户的在线状态分为两种:Online和Offline。 作为一个独立的服务应该为聊天服务器提供以下功能
实现多端设备同时在线查询某个用户的在线状态登记某个用户的在线状态(包括下线和上线)
技术栈
Redis 这里在设计时没有将用户的上下线记录进行持久化只需要将用户的在线数据使用redis进行缓存即可当然为了方便用户查看自己的登录历史记录该数据也具有一定的重要性后序可能会增加。 gRPC 该服务是可以独立运行的并且可以进行集群部署实现系统的高扩展性和高可用所以整体必然采用微服务架构使用RPC协议作为微服务之间的通讯协议满足实时性。gRPC框架是RPC的一种通用实现官网文档也很详细所以这里采用该框架来实现RPC通信。 ETCD 采用微服务的架构必然需要考虑服务的注册与发现的实现方案虽然gRPC框架支持服务的注册与发现并且可以进行负载均衡但是使用ETCD可以为系统提供高扩展性和可用性并且ETCD更加全面支持服务的健康状态检查和数据一致性所以为系统后序可以更好的扩展这里采用ETCD来实现服务的注册与发现。 Zap Zap是一个轻量、快速的日志框架。
准备工作
封装日志框架
本封装只是对zap进行简单的套壳目的是为了更方便的调用。像log一样
//pkg/logger/logger.go
var (logger *zap.Logger
)func init() {dev, err : zap.NewDevelopment()if err ! nil {panic(err)}logger dev
}func Info(msg string, fields ...zap.Field) {logger.Info(msg, fields...)
}func Error(msg string, field ...zap.Field) {logger.Error(msg, field...)
}func Fatal(msg string, field ...zap.Field) {logger.Fatal(msg, field...)
}func Warn(msg string, field ...zap.Field) {logger.Warn(msg, field...)
}func Debug(msg string, field ...zap.Field) {logger.Debug(msg, field...)
}func DPanic(msg string, field ...zap.Field) {logger.DPanic(msg, field...)
}func Sync() {logger.Sync()
}封装Redis
为了实现用户状态的缓存需要独立封装redis的相关操作使得实际的业务代码更加的专一和简洁。
连接Redis服务
//pkg/db/redis.go
var (StatusRDB *redis.Client
)func init() {StatusRDB InitRedis()
}// InitRedis 初始化Redis连接
func InitRedis() *redis.Client {rdb : redis.NewClient(redis.Options{Addr: config.RedisAddr,Password: config.RedisPass,DB: config.RedisDB,})_, err : rdb.Ping(context.Background()).Result()if err ! nil {logger.Fatal(Redis 初始化失败, zap.String(失败原因, err.Error()))}logger.Info(Redis 初始化成功)return rdb
}封装Redis操作 封装之前需要考虑一下需要哪些功能我们的目的是记录一个用户的在线状态该在线状态是一个对象我们需要做的是添加整个字段和修改用户的在线状态字段。所以value可选择的方案有String和Hash。使用字符串的话需要频繁的序列化和反序列化对象所以这里使用Hash作为value的容器。
因为要支持多端设备登录所以是多个设备对应同一个用户所以这里将用户的设备号作为key保证唯一性。
func RegisterUserStatus(context context.Context, deviceID string, status model.UserStatus) error; 后面将看到model.UserStatus的定义func UpdateUserStatus(context context.Context, deviceID string, status string, newStatus bool) errorfunc RemoveAfter(context context.Context, deviceID string, duration time.Duration) error;func CheckOnline(context context.Context, deviceID string) (bool, error)func GetUserStatus(context context.Context, deviceID string) (*model.UserStatus, error)
// RegisterUserStatus 注册用户的在线状态
func RegisterUserStatus(context context.Context, deviceID string, status model.UserStatus) error {_, err : db.StatusRDB.HSet(context, deviceID, map[string]interface{}{UserID: status.UserID,DeviceID: status.DeviceID,Status: status.Status,ServerAddr: status.ServerAddr,LoginTime: status.LoginTime,LogoutTime: status.LogoutTime,}).Result()return err
}// UpdateUserStatus 更新用户的在线状态
func UpdateUserStatus(context context.Context, deviceID string, status string, newStatus bool) error {_, err : db.StatusRDB.HSet(context, deviceID, map[string]interface{}{status: newStatus,}).Result()return err
}// CheckOnline 检查用户是否在线
func CheckOnline(context context.Context, deviceID string) (bool, error) {status, err : db.StatusRDB.HGet(context, deviceID, status).Result()if err redis.Nil {return false, nil}return status 1, err
}// RemoveAfter 设置键的过期时间
func RemoveAfter(context context.Context, deviceID string, duration time.Duration) error {_, err : db.StatusRDB.ExpireNX(context, deviceID, duration).Result()return err
}// GetUserStatus 获取用户的在线状态记录
func GetUserStatus(context context.Context, deviceID string) (*model.UserStatus, error) {// 查询该键值是否存在count, err : db.StatusRDB.Exists(context, deviceID).Result()if err ! nil {return nil, err}if count 0 {return nil, nil}result, err : db.StatusRDB.HGetAll(context, deviceID).Result()if err ! nil {return nil, err}loginTime, _ : strconv.Atoi(result[LoginTime])logoutTime, _ : strconv.Atoi(result[LogoutTime])return model.UserStatus{UserID: result[UserID],DeviceID: result[DeviceID],Status: result[Status] 1,ServerAddr: result[ServerAddr],LoginTime: int64(loginTime),LogoutTime: int64(logoutTime),}, nil
}封装ETCD
ETCD的作用是用于服务的注册与发现所以我们要封装注册服务和注销服务的方法。程序设计可以参考ETCD官网
func RegisterServerToEtcd(addr string) error;func UnRegisterFromEtcd(addr string) error;
const (serviceName im/state-serverttl 10
)var (etcdClient *clientv3.Client
)func init() {c, err : clientv3.NewFromURL(config.EtcdAddr)if err ! nil {logger.Error(连接ETCD服务器失败, zap.String(失败原因, err.Error()))panic(err)}etcdClient c
}// RegisterServerToEtcd 注册服务到ETCD
func RegisterServerToEtcd(addr string) error {manager, err : endpoints.NewManager(etcdClient, serviceName)if err ! nil {return err}lease, _ : etcdClient.Grant(context.TODO(), ttl)err manager.AddEndpoint(context.TODO(), fmt.Sprintf(%s/%s, serviceName, addr), endpoints.Endpoint{Addr: addr}, clientv3.WithLease(lease.ID))if err ! nil {return err}alive, err : etcdClient.KeepAlive(context.TODO(), lease.ID)if err ! nil {return err}go func() {for {-alive}}()return nil
}// UnRegisterFromEtcd 从ETCD注销服务
func UnRegisterFromEtcd(addr string) error {em, err : endpoints.NewManager(etcdClient, serviceName)if err ! nil {return err}err em.DeleteEndpoint(context.TODO(), fmt.Sprintf(%s/%s, serviceName, addr))return err
}业务逻辑开发
定义模型
// UserStatus 用户的在线状态表
type UserStatus struct {UserID string json:user_idDeviceID string json:device_idStatus bool json:status // (0:表示离线 1:表示在线)ServerAddr string json:server_addr //用户所在的服务器地址LoginTime int64 json:login_timeLogoutTime int64 json:logout_time
}定义服务
syntax proto3;package pb;
option go_package ./pb;// 当服务器查询某个用户的状态时需要提供该用户的ID和设备号
message QueryUserStatusRequest{string UserID 1;string DeviceID 2;
}// 状态服务器给出响应返回该用户的在线状态
message QueryUserStatusResponse{bool Status 1;string ServerAddr 2;
}// 当用户上线需要告诉状态服务器
message OnlineRequest{string UserID 1;string DeviceID 2;string ServerAddr 4;
}// 返回用户的在线状态是否设置成功
message OnlineResponse{
}// 当用户下线需要告诉状态服务器
message OfflineRequest{string UserID 1;string DeviceID 2;
}message OfflineResponse{
}service UserStatus{// 查询用户在线状态的服务rpc QueryUserStatus(QueryUserStatusRequest) returns(QueryUserStatusResponse);// 上线服务rpc Online(OnlineRequest) returns(OnlineResponse);// 下线服务rpc Offline(OfflineRequest) returns(OfflineResponse);
}通过protoc命令将上述文件生成一份客户端和服务端的代码。
实现服务接口
type UserStatusService struct {pb.UnimplementedUserStatusServer
}func NewUserStatusService() *UserStatusService {return UserStatusService{}
}// QueryUserStatus 查询用户在线状态的服务
func (u *UserStatusService) QueryUserStatus(c context.Context, req *pb.QueryUserStatusRequest) (*pb.QueryUserStatusResponse, error) {logger.Info(调用用户在线状态查询服务)// 查询用户的在线状态status, err : repository.GetUserStatus(c, req.DeviceID)if err ! nil {logger.Error(查询用户在线状态失败, zap.String(失败的原因, err.Error()))return nil, errors.New(查询用户在线状态失败)}if status nil || !status.Status {logger.Info(查询的用户不在缓存中)return pb.QueryUserStatusResponse{Status: false,}, nil}return pb.QueryUserStatusResponse{Status: status.Status,ServerAddr: status.ServerAddr,}, nil
}// Online 上线服务
func (u *UserStatusService) Online(c context.Context, request *pb.OnlineRequest) (*pb.OnlineResponse, error) {logger.Info(调用上线服务)status : model.UserStatus{UserID: request.UserID,DeviceID: request.DeviceID,Status: true,ServerAddr: request.ServerAddr,LoginTime: time.Now().Unix(),LogoutTime: 0,}err : repository.RegisterUserStatus(c, request.DeviceID, status)if err ! nil {logger.Error(将用户在线信息存入Redis失败, zap.String(失败原因, err.Error()))return nil, errors.New(记录用户的在线状态失败)}logger.Info(成功记录用户的在线状态)return pb.OnlineResponse{}, nil
}// Offline 下线服务
func (u *UserStatusService) Offline(c context.Context, request *pb.OfflineRequest) (*pb.OfflineResponse, error) {logger.Info(调用下线服务)err : repository.UpdateUserStatus(c, request.DeviceID, Status, false)if err ! nil {logger.Error(更新用户在线状态失败, zap.String(失败原因, err.Error()))return nil, errors.New(更新用户在线状态失败)}err repository.RemoveAfter(c, request.DeviceID, time.Minute*3)if err ! nil {logger.Error(设置用户在线状态过期时间失败, zap.String(失败原因, err.Error()))return nil, errors.New(更新用户在线状态失败)}return pb.OfflineResponse{}, nil
}服务端启动程序
var (ip stringport string
)func init() {const (defaultAddr 127.0.0.1defaultPort 8080)flag.StringVar(ip, addr, defaultAddr, IP地址)flag.StringVar(port, port, defaultPort, 服务端口)
}func main() {flag.Parse()defer logger.Sync()logger.Info(正在启动状态服务器, zap.String(日志框架, zap))// 注册服务ch : make(chan os.Signal, 1)signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGABRT)go func() {s : -cherr : etcd.UnRegisterFromEtcd(ip : port)if err ! nil {logger.Info(注销服务失败, zap.String(失败原因, err.Error()))} else {logger.Info(注销服务成功)}if i, ok : s.(syscall.Signal); ok {os.Exit(int(i))} else {os.Exit(0)}}()err : etcd.RegisterServerToEtcd(ip : port)if err ! nil {logger.Fatal(注册服务到ETCD失败, zap.String(失败原因, err.Error()), zap.String(程序状态, 即将退出))}// 添加选项var options []grpc.ServerOptionuserStatusRpcServer : grpc.NewServer(options...)pb.RegisterUserStatusServer(userStatusRpcServer, api.NewUserStatusService())listen, err : net.Listen(tcp, :8080)logger.Info(服务启动, zap.String(服务地址, fmt.Sprintf(%s:%s, ip, port)))if err ! nil {logger.Fatal(启动监听失败, zap.String(错误信息, err.Error()), zap.String(程序状态, 退出))}if userStatusRpcServer.Serve(listen) ! nil {logger.Fatal(启动监听失败, zap.String(错误信息, err.Error()), zap.String(程序状态, 退出))}
}客户端测试程序
func main() {cli, cerr : clientv3.NewFromURL(config.EtcdAddr)if cerr ! nil {panic(cerr)}etcdResolver, err : resolver.NewBuilder(cli)if err ! nil {panic(err)}conn, gerr : grpc.Dial(etcd:///im/state-server, grpc.WithResolvers(etcdResolver), grpc.WithTransportCredentials(insecure.NewCredentials()))if gerr ! nil {println(gerr.Error())panic(conn)}defer conn.Close()client : pb.NewUserStatusClient(conn)response, err : client.QueryUserStatus(context.Background(), pb.QueryUserStatusRequest{UserID: 1, DeviceID: 2})if err ! nil {println(err.Error())}fmt.Printf(%#v, response.Status)
}总结
本篇文章记录了一个IM系统中状态服务器的设计和实现完整代码可在github查看state-server
