个人技术分享

写在前面

其实很少有公司会使用 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 redis
	cli *redis.Client // client for redis
	rwLock *sync.RWMutex // rwLock lock groupList when update service instance
	// vgroupMapping to store the cluster group
	// eg: map[cluster_name_key]cluster_name
	vgroupMapping map[string]string
	// groupList store all addresses under this cluster
	// eg: map[cluster_name][]{service_instance1,service_instance2...}
	groupList map[string][]*ServiceInstance
	ctx context.Context
}

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

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

2. 对象加载

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

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

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.VgroupMapping
	groupList := 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 time
	redisRegistryService.load()
	// subscribe at real time
	go redisRegistryService.subscribe()
	return redisRegistryService
}

3. 服务加载

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

func (s *RedisRegistryService) load() {
	// find all the server list redis register by redisFileKeyPrefix
	keys, _, 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 list
		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()
	}
}

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. 服务注册

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

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 *NotifyMessage
			err := json.Unmarshal([]byte(msg.Payload), &data)
			if err != nil {
				log.Errorf("RedisRegistryService-subscribe-Subscribe-Err:%+v", err)
				continue
			}
			// get cluster name by key
			clusterName := 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
}

注意一点：redis的发布订阅的消息是不存储到日志的，也没有ack确认。 所以如果发生的消息的丢失，就需要业务自己承担了，比如自己实现一个ack，发送的时候进行消息日志的存储。

完整代码：
https://github.com/CocaineCong/incubator-seata-go/blob/discovery/redis/pkg/discovery/redis.go