基于 Redis 发布订阅实现服务注册与发现

写在前面

其实很少有公司会使用 Redis 来实现服务注册与发现，通常是ETCD、NACOS、ZOOKEEPER等等，但是也不妨碍我们了解。本文会先介绍 Redis 的发布/订阅模式，接着基于这个模式实现服务注册与发现。

Redis发布订阅流程图：
在这里插入图片描述

Redis 发布订阅

1. 简介

Redis的发布订阅功能主要由PUBLISH、SUBSCRIBE、PSUBSCRIBE 等命令组成的。

通过执行 SUBSCRIBE 命令，客户端可以订阅一个或多个频道，从而成为这个频道的订阅者。
在这里插入图片描述

每当有其他客户端向这个被订阅的频道发送消息的时候，频道的所有订阅者都会收到这条消息。

在这里插入图片描述
当然，客户端还可以通过PSUBSCRIBE订阅一个或多个模式，从而成为这些模式的订阅者，也就是模糊匹配

2. 订阅

每当一个客户端执行SUBSCRIBE命令订阅某个或某些频道的时候，这个客户端与被订阅者之间就会建立起一种订阅关系。而Redis会将这种订阅关系保存到pubsub_channels 这个字典中，这个字典的键是某个被订阅的频道，而值是一个链表，这个链表记录了所有订阅这个频道的客户端

在这里插入图片描述
每当有客户端执行了SUBSCRIBE命令订阅某个或某些频道的时候，服务器都会将客户端与被订阅的频道在 pubsub_channels字典中进行关联。

3. 退订

如果进行退订UNSUBSCRIBE，那么服务器会从pubsub_channels中接触客户端与被退订频道之间的关联。当这个key中，已经没有订阅者，那么会将这个key进行删除。例如下面的client7
在这里插入图片描述

4. 发布消息

当一个Redis客户端执行 PUBLISH channel message 命令将消息 message 发送给channel的时候，将消息发送给channel频道的所有订阅者（本文不讨论pattern模式）

服务注册与发现

我们了解完redis的发布订阅流程之后，我们来基于这个发布订阅来实现一个服务注册与发现的功能。

Redis服务发现与注册流程图：
在这里插入图片描述

1. 对象定义

redis 服务发现与注册的结构体

type RedisRegistryService struct {config *RedisConfig // the config about rediscli *redis.Client // client for redisrwLock *sync.RWMutex // rwLock lock groupList when update service instance// vgroupMapping to store the cluster group// eg: map[cluster_name_key]cluster_namevgroupMapping map[string]string// groupList store all addresses under this cluster// eg: map[cluster_name][]{service_instance1,service_instance2...}groupList map[string][]*ServiceInstancectx context.Context
}

订阅的消息内容，为key 以及 value ，而key就是服务的name，value就是服务的具体地址

type NotifyMessage struct {// key = registry.redis.${cluster}_ip:portKey   string `json:"key"`Value string `json:"value"`
}

2. 对象加载

新建一个redis 服务注册与发现对象，并且在创建的这个对象的时候，我们会做两件事情

将redis中所已存在的key都load一次，存到本地缓存中。
开启一些协程进行发布订阅，不断监听上游的注册消息

func newRedisRegisterService(config *ServiceConfig, redisConfig *RedisConfig) RegistryService {if redisConfig == nil {log.Fatalf("redis config is nil")panic("redis config is nil")}cfg := &redis.Options{Addr:     redisConfig.ServerAddr,Username: redisConfig.Username,Password: redisConfig.Password,DB:       redisConfig.DB,}cli := redis.NewClient(cfg)vgroupMapping := config.VgroupMappinggroupList := make(map[string][]*ServiceInstance)redisRegistryService := &RedisRegistryService{config:        redisConfig,cli:           cli,ctx:           context.Background(),rwLock:        &sync.RWMutex{},vgroupMapping: vgroupMapping,groupList:     groupList,}// loading all server at init timeredisRegistryService.load()// subscribe at real timego redisRegistryService.subscribe()return redisRegistryService
}

3. 服务加载

load 函数：将所有 key 都 scan 出来，再遍历所有的key，拿到对应的value，进行一次初始化操作，加载到本地缓存中

func (s *RedisRegistryService) load() {// find all the server list redis register by redisFileKeyPrefixkeys, _, err := s.cli.Scan(s.ctx, 0, fmt.Sprintf("%s*", redisFileKeyPrefix), 0).Result()if err != nil {log.Errorf("RedisRegistryService-Scan-Key-Error:%s", err)return}for _, key := range keys {clusterName := s.getClusterNameByKey(key)val, err := s.cli.Get(s.ctx, key).Result()if err != nil {log.Errorf("RedisRegistryService-Get-Key:%s, Err:%s", key, err)continue}ins, err := s.getServerInstance(val)if err != nil {log.Errorf("RedisRegistryService-getServerInstance-val:%s, Err:%s", val, err)continue}// put server instance list in group lists.rwLock.Lock()if s.groupList[clusterName] == nil {s.groupList[clusterName] = make([]*ServiceInstance, 0)}s.groupList[clusterName] = append(s.groupList[clusterName], ins)s.rwLock.Unlock()}
}

4.服务发现

通过 key 从 vgroupMapping 找到对应的 value

func (s *RedisRegistryService) Lookup(key string) (r []*ServiceInstance, err error) {s.rwLock.RLock()defer s.rwLock.RUnlock()cluster := s.vgroupMapping[key]if cluster == "" {err = fmt.Errorf("cluster doesnt exit")return}r = s.groupList[cluster]return
}

5. 服务注册

将 key 和 value set到 redis 中
将 key 和 value 通过 Channel 发布出去
另外开启一个协程将进行保活

func (s *RedisRegistryService) register(key, value string) (err error) {_, err = s.cli.HSet(s.ctx, key, value).Result()if err != nil {return}msg := &NotifyMessage{Key:   key,Value: value,}s.cli.Publish(s.ctx, redisRegisterChannel, msg)go func() {s.keepAlive(s.ctx, key)}()return
}

6. 服务订阅

订阅 Subscribe Channel 监听上游服务，并对服务的 key 和 value 进行更新操作。 注意这里对map进行读写的时候要加上读写锁，防止线程不安全。

func (s *RedisRegistryService) subscribe() {go func() {msgs := s.cli.Subscribe(s.ctx, redisRegisterChannel).Channel()for msg := range msgs {var data *NotifyMessageerr := json.Unmarshal([]byte(msg.Payload), &data)if err != nil {log.Errorf("RedisRegistryService-subscribe-Subscribe-Err:%+v", err)continue}// get cluster name by keyclusterName := s.getClusterNameByKey(data.Key)ins, err := s.getServerInstance(data.Value)if err != nil {log.Errorf("RedisRegistryService-subscribe-getServerInstance-value:%s, Err:%s", data.Value, err)continue}s.rwLock.Lock()if s.groupList[clusterName] == nil {s.groupList[clusterName] = make([]*ServiceInstance, 0)}s.groupList[clusterName] = append(s.groupList[clusterName], ins)s.rwLock.Unlock()}}()return
}