基于 SOFAJRaft 实现注册中心

文章目录

- 1.前言
- 2.git 示例地址
- 3.官网示例分析
- 3.SOFAJRAFT 注册中心实现（服务端）
- - 3.1 核心功能
  - 3.2 模块设计
  - 3.3 请求消息数据结构设计
  - - 3.3.1 Registration 注册消息
    - 3.3.2 GetServiceInstancesRequest 获取服务实例请求
    - 3.3.3 GetServiceInstancesResponse 获取服务实例返回对象
    - 3.3.4 Response 返回对象
    - 3.3.5 HeartBeat 心跳
- 4.实现步骤
- - 4.1 ServiceDiscoveryOuter 自动生成 proto 类
  - 4.2 服务注册发现接口
  - 4.3 ServiceDiscoveryServer 服务发现服务端
  - 4.4 ServiceDiscoveryStateMachine 状态机
  - - 功能点1
    - 功能点2
    - 功能点3
    - 功能点4
    - - 服务注册请求处理者
      - 服务注销请求处理者
  - 4.5 RpcProcessor 请求处理器
  - - GetServiceInstancesRequestRpcProcessor
    - RegistrationRpcProcessor
  - 4.6 ServiceDiscoveryClient 客户端
  - 4.7 实现 ServiceDiscovery 接口
  - 4.8 Spring Web 项目集成
- 5.测试

1.前言

通过本文的学习可以帮助大家了解 SOFAJRaft 的使用方式和集成步骤；参考本文的实现步骤，可以来完成自己工作中相关产品的一致性协议 raft 集成，从而实现应用的高可用。
学习本文的前提

需要了解 Raft 一致性协议
对 SOFAJRaft 框架有一定的了解
可以参考：SOFAJRaft 日志复制共识算法 https://blog.csdn.net/xiewenfeng520/article/details/133775751

本文主要讨论的内容如下

注册中心的实现
如何集成 Raft 协议来实现注册中心高可用

2.git 示例地址

本文完整代码地址：https://github.com/huajiexiewenfeng/eval-discovery

3.官网示例分析

截图中的报错，是对官网示例做了一定的改造，不影响，忽略即可；大家也可以自己来扩展官方的代码来加深对 JRaft 的理解。

代码分为以下几个部分

CounterClient 客户端
CounterServer 服务端
CounterStateMachine 状态机
CounterClosure 回调
CounterOperation 操作
- GET
- INCREMENT
- 扩展 …
Processor 操作对应的处理类
- IncrementAndGetRequestProcessor -> INCREMENT
- GetValueRequestProcessor -> GET
CounterSnapshotFile 快照存储
CounterService 计数服务接口
- CounterServiceImpl 计数服务实现

具体的实现细节可参考，我们这里就不做赘述了

https://www.sofastack.tech/projects/sofa-jraft/counter-example/

3.SOFAJRAFT 注册中心实现（服务端）

3.1 核心功能

服务注册（Service Registration）
服务订阅（Service Subscription）
数据同步（Replicas Sync）
- SOFAJRAFT 内建支持

3.2 模块设计

按照官网 counter 示例，注册中心模块设计如下：

ServiceDiscoveryClient 客户端
ServiceDiscoveryServer 服务端
ServiceDiscoveryStateMachine 状态机
ServiceDiscoveryClosure 回调
ServiceDiscoveryOperation 操作
- REGISTRATION 注册
- DEREGISTRATION 注销
- GET_SERVICE_INSTANCES 获取服务实例
- BEAT 心跳
Processor 操作对应的处理类
- RegistrationRpcProcessor -> REGISTRATION
- GetServiceInstancesRequestRpcProcessor -> GET_SERVICE_INSTANCES
- HeartBeatRpcProcessor -> BEAT
注册中心一般采用内存型存储即可
ServiceDiscovery 服务发现注册接口
- JRaftServiceDiscovery 基于 JRaft 服务发现注册实现

3.3 请求消息数据结构设计

3.3.1 Registration 注册消息

reversed 用于区分该消息是注册还是注销，true 表示注销，false 表示注册。

message Registration {
  string id = 1;
  string serviceName = 2;
  string host = 3;
  int32 port = 4;
  map<string, string> metadata = 5;
  optional bool reversed = 6;
}

3.3.2 GetServiceInstancesRequest 获取服务实例请求

通过 serviceName 获取服务实例集合

message GetServiceInstancesRequest {
  string serviceName = 1;
}

3.3.3 GetServiceInstancesResponse 获取服务实例返回对象

返回服务实例集合

message GetServiceInstancesResponse {
  repeated Registration value = 1;
}

3.3.4 Response 返回对象

message Response {
  int32 code = 1;
  optional string message = 2;
}

3.3.5 HeartBeat 心跳

message HeartBeat {
  string id = 1;
  string serviceName = 2;
  string host = 3;
  int32 port = 4;
}

4.实现步骤

我们按照上面的设计的模块，将类都创建好，然后将官网示例中对应的代码 copy 到类中，慢慢再根据我们的需求来进行改造。

4.1 ServiceDiscoveryOuter 自动生成 proto 类

1.proto 文件

syntax = "proto3";

package service.discovery;

option java_package = "com.csdn.eval.discovery.jraft.proto";
option java_outer_classname = "ServiceDiscoveryOuter";

message Registration {
  string id = 1;
  string serviceName = 2;
  string host = 3;
  int32 port = 4;
  map<string, string> metadata = 5;
}

message HeartBeat {
  string id = 1;
  string serviceName = 2;
  string host = 3;
  int32 port = 4;
}

message Response {
  int32 code = 1;
  string message = 2;
}

message GetServiceInstancesRequest {
  string serviceName = 1;
}

message GetServiceInstancesResponse {
  repeated Registration value = 1;
}

2.生成 proto 对应的 java 类
pom 中新增 proto 插件和依赖

  <properties>
    <jraft.version>1.3.12</jraft.version>
    <protobuf-java.version>3.22.4</protobuf-java.version>
  </properties>

 <dependencies>
    <!-- SOFA JRAFT -->
    <dependency>
      <groupId>com.alipay.sofa</groupId>
      <artifactId>jraft-core</artifactId>
      <version>${jraft.version}</version>
    </dependency>

    <dependency>
      <groupId>com.alipay.sofa</groupId>
      <artifactId>jraft-rheakv-core</artifactId>
      <version>${jraft.version}</version>
    </dependency>

    <dependency>
      <groupId>com.google.protobuf</groupId>
      <artifactId>protobuf-java</artifactId>
      <version>${protobuf-java.version}</version>
    </dependency>
 </dependencies>

  <build>
    <extensions>
      <extension>
        <groupId>kr.motd.maven</groupId>
        <artifactId>os-maven-plugin</artifactId>
        <version>1.7.1</version>
      </extension>
    </extensions>
    <plugins>
      <plugin>
        <groupId>org.xolstice.maven.plugins</groupId>
        <artifactId>protobuf-maven-plugin</artifactId>
        <version>0.6.1</version>
        <configuration>
          <protoSourceRoot>src/main/resources/proto</protoSourceRoot>
          <protocArtifact>com.google.protobuf:protoc:3.21.7:exe:${os.detected.classifier}</protocArtifact>
          <pluginId>grpc-java</pluginId>
          <pluginArtifact>io.grpc:protoc-gen-grpc-java:1.54.1:exe:${os.detected.classifier}</pluginArtifact>
        </configuration>
        <executions>
          <execution>
            <goals>
              <goal>compile</goal>
              <goal>compile-custom</goal>
            </goals>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>

3.编译完成之后，将类拷贝到 src 目录下面

4.2 服务注册发现接口

初始化
注册
注销
获取服务实例集合
关闭

public interface ServiceDiscovery {

    ServiceDiscovery DEFAULT = loadDefault(ServiceDiscovery.class);

    void initialize(Map<String, Object> config);

    void register(ServiceInstance serviceInstance);

    void deregister(ServiceInstance serviceInstance);

    List<ServiceInstance> getServiceInstances(String serviceName);

    void close();

}

4.3 ServiceDiscoveryServer 服务发现服务端

将示例代码 com.alipay.sofa.jraft.example.counter.CounterServer 整个 copy 到我们的类中，再将下图中所有的报错处理完

将相关的组件替换成我们自己的组件
- CounterStateMachine -> ServiceDiscoveryStateMachine
- GetValueRequestProcessor -> RegistrationRpcProcessor
- IncrementAndGetRequestProcessor -> GetServiceInstancesRequestRpcProcessor
- 新增 HeartBeatRpcProcessor
将没有用的代码注释掉或者删除掉

4.4 ServiceDiscoveryStateMachine 状态机

按照上面的操作将代码复制，我在重要代码中增加了中文注释，从下图中可以看到大概有三个改造点

我们要实现功能点如下

功能点1

从 ServiceDiscoveryClosure 获取到当前请求的类型，这里我们可以建一个枚举类，类型如下：
- REGISTRATION 注册
- DEREGISTRATION 注销
- GET_SERVICE_INSTANCES 获取实例
- BEAT 心跳

    public enum Kind {

        REGISTRATION,

        DEREGISTRATION,

        GET_SERVICE_INSTANCES,

        BEAT;
    }

功能点2

ServiceDiscoveryOperation 序列化和反序列化的实现，也就是改造点2 的部分，移到 ServiceDiscoveryOperation 类中来实现，减少主流程的冗余代码。

功能点3

通过请求类型来完成对应请求方法的调用，这里我们可以采用策略模式来实现

创建策略工厂类
ServiceDiscoveryRequestHandlerFactory

RegistrationRequestHandler 注册处理者
DeRegistrationRequestHandler 注销处理者
GetServiceInstancesRequestHandler 获取服务实例处理者
HeartBeatRequestHandler 心跳处理者

public class ServiceDiscoveryRequestHandlerFactory {

  /**
   * 服务名称与服务实例列表（List）映射
   */
  private final Map<String, Map<String, ServiceInstance>> serviceNameToInstancesStorage = new ConcurrentHashMap<>();

  private final Map<Kind, ServiceDiscoveryRequestHandler> handlers = new HashMap<>();

  public void init() {
    handlers.put(Kind.REGISTRATION, new RegistrationRequestHandler(this));
    handlers.put(Kind.DEREGISTRATION, new DeRegistrationRequestHandler(this));
    handlers.put(Kind.GET_SERVICE_INSTANCES, new GetServiceInstancesRequestHandler(this));
    handlers.put(Kind.BEAT, new HeartBeatRequestHandler(this));
  }

  public ServiceDiscoveryRequestHandlerFactory() {

  }

  public ServiceDiscoveryRequestHandler getHandler(Kind kind) {
    return handlers.get(kind);
  }

  public synchronized void storage(String id, String serviceName, ServiceInstance serviceInstance) {
    Map<String, ServiceInstance> serviceInstancesMap = serviceNameToInstancesStorage
        .computeIfAbsent(serviceName, n -> new LinkedHashMap<>());
    serviceInstancesMap.put(id, serviceInstance);
  }

  public synchronized void delete(String id, String serviceName) {
    Map<String, ServiceInstance> instanceMap = getServiceInstancesMap(serviceName);
    instanceMap.remove(id);
  }

  public Map<String, ServiceInstance> getServiceInstancesMap(String serviceName) {
    return serviceNameToInstancesStorage.computeIfAbsent(serviceName, n -> new LinkedHashMap<>());
  }

}

在 ServiceDiscoveryServer 创建的过程中，通过构造函数来创建 Factory，并传入到 Fsm 状态机中
改造完成之后状态机代码如下

  @Override
  public void onApply(Iterator iter) {
    while (iter.hasNext()) {
      long current = 0;
      ServiceDiscoveryOperation operation = null;

      ServiceDiscoveryClosure closure = null;
      if (iter.done() != null) {
        // 从当前 Leader 节点获取 Closure
        closure = (ServiceDiscoveryClosure) iter.done();
        // 从 Closure 获取服务操作的类型
        operation = closure.getServiceDiscoveryOperation();
        logger.info("The closure with operation[{}] at the Leader node[{}]", operation, node);
      } else {
        // 在 Follower 节点通过 日志反序列化得到 ServiceDiscoveryOperation
        final ByteBuffer data = iter.getData();
        operation = ServiceDiscoveryOperation.deserialize(data);
        logger.info("The closure with operation[{}] at the Follower node[{}]", operation, node);
      }
      // 根据服务操作类型的不同来进行不同的业务操作
      if (operation != null) {
        ServiceDiscoveryRequestHandlerFactory instanceFactory = ServiceDiscoveryRequestHandlerFactory
            .getInstance();
        instanceFactory.init();
        instanceFactory.getHandler(kind)
            .doHandle(closure, (ServiceInstance) operation.getData());
        if (closure != null) {
          closure.run(Status.OK());
        }
      }
      iter.next();
    }
  }

关键代码：
instanceFactory.getHandler(kind).doHandle(closure, (ServiceInstance) operation.getData());
通过 kind 获取到对应的 handler，处理对应的请求方法。

功能点4

分别实现注册，注销，获取实例，心跳方法

服务注册请求处理者

public class RegistrationRequestHandler implements ServiceDiscoveryRequestHandler {

  private static final Logger logger = LoggerFactory.getLogger(RegistrationRequestHandler.class);

  private ServiceDiscoveryRequestHandlerFactory factory;

  public RegistrationRequestHandler(
      ServiceDiscoveryRequestHandlerFactory factory) {
    this.factory = factory;
  }

  @Override
  public void doHandle(ServiceDiscoveryClosure closure, ServiceInstance serviceInstance) {
    if (null == serviceInstance) {
      return;
    }
    String serviceName = serviceInstance.getServiceName();
    String id = serviceInstance.getId();

    factory.storage(id, serviceName, serviceInstance);

    logger.info("{} has been registered at the node", serviceInstance);
  }
}

服务注销请求处理者

public class DeRegistrationRequestHandler implements ServiceDiscoveryRequestHandler {

  private static final Logger logger = LoggerFactory.getLogger(DeRegistrationRequestHandler.class);

  private ServiceDiscoveryRequestHandlerFactory factory;

  public DeRegistrationRequestHandler(
      ServiceDiscoveryRequestHandlerFactory factory) {
    this.factory = factory;
  }

  @Override
  public void doHandle(ServiceDiscoveryClosure closure, ServiceInstance serviceInstance) {
    if (null == serviceInstance) {
      return;
    }
    String serviceName = serviceInstance.getServiceName();
    String id = serviceInstance.getId();

    factory.delete(id, serviceName);

    logger.info("{} has been deregistered at the node", serviceInstance);
  }
}

4.5 RpcProcessor 请求处理器

调用流程

RpcProcessor 实现的是 client -> apply(task) 这一部分的功能，具体的调用流程可以看官网 Counter 示例中的
com.alipay.sofa.jraft.example.counter.CounterServiceImpl#incrementAndGet 方法

此方法是在 IncrementAndGetRequestProcessor 中来调用的

GetServiceInstancesRequestRpcProcessor

参考 com.alipay.sofa.jraft.example.counter.rpc.IncrementAndGetRequestProcessor 示例来看需要完成两步

构造 Closure 回调
构造 task 对象，调用 Node().apply(task);

将对应的代码 copy 到 GetServiceInstancesRequestRpcProcessor 中，核心代码如下：

构造 Closure 回调，还有一个 getValueResponse() 方法需要实现，我们拿到返回结果；
getValueResponse() ->ServiceDiscoveryClosure#getResult()

  @Override
  public void handleRequest(RpcContext rpcContext, GetServiceInstancesRequest request) {
    String serviceName = request.getServiceName();

    ServiceDiscoveryOperation op = new ServiceDiscoveryOperation(Kind.GET_SERVICE_INSTANCES,
        serviceName);

    final ServiceDiscoveryClosure closure = new ServiceDiscoveryClosure(op) {
      @Override
      public void run(Status status) {
        rpcContext.sendResponse(getResult());
      }
    };

    if (!isLeader()) {
      handlerNotLeaderError(closure);
      return;
    }

    final Task task = new Task();
    task.setData(op.serialize());
    task.setDone(closure);
    this.serviceDiscoveryServer.getNode().apply(task);

  }

与 Counter 示例对比，发现 rpcContext.sendResponse(getResult()) 与原示例语义不同，rpcContext.sendResponse 需要的是 Response 对象，而我们这里返回的是 result 对象，是服务实例集合。

我们需要实现一个类似于 ValueResponse 的封装，返回 ServiceDiscoveryOuter.GetServiceInstancesResponse 对象。response 方法将 result 对象转换成 GetServiceInstancesResponse 对象，核心代码如下：

  private ServiceDiscoveryOuter.GetServiceInstancesResponse response(Object result) {
    Collection<ServiceInstance> serviceInstances = (Collection<ServiceInstance>) result;
    GetServiceInstancesResponse response = GetServiceInstancesResponse.newBuilder()
        .addAllValue(convertRegistrations(serviceInstances)).build();
    return response;
  }

RegistrationRpcProcessor

按照上面的方式我们再实现注册功能，核心代码如下：

public class RegistrationRpcProcessor implements RpcProcessor<ServiceDiscoveryOuter.Registration> {
  ...
  @Override
  public void handleRequest(RpcContext rpcContext, Registration registration) {
    ServiceInstance serviceInstance = convertServiceInstance(registration);

    String serviceName = registration.getServiceName();

    final Kind kind = Kind.REGISTRATION;

    ServiceDiscoveryOperation op = new ServiceDiscoveryOperation(kind, serviceInstance);

    final ServiceDiscoveryClosure closure = new ServiceDiscoveryClosure(op) {
      @Override
      public void run(Status status) {
        if (!status.isOk()) {
          logger.warn("Closure status is : {} at the {}", status, serviceDiscoveryServer.getNode());
          return;
        }
        rpcContext.sendResponse(response(status));
        logger.info("'{}' has been handled ,serviceName : '{}' , result : {} , status : {}",
            kind, serviceName, getResult(), status);
      }
    };

    if (!isLeader()) {
      handlerNotLeaderError(closure);
      return;
    }

    final Task task = new Task();
    task.setData(op.serialize());
    task.setDone(closure);
    this.serviceDiscoveryServer.getNode().apply(task);

  }
  ...
}

里面有很多重复代码，我们再进行重构。
将通用的代码抽取到 RpcProcessorImpl 中

public class RpcProcessorImpl implements RpcProcessorService {

  private static final Logger logger = LoggerFactory.getLogger(RpcProcessorImpl.class);

  private final ServiceDiscoveryServer serviceDiscoveryServer;

  public RpcProcessorImpl(ServiceDiscoveryServer serviceDiscoveryServer) {
    this.serviceDiscoveryServer = serviceDiscoveryServer;
  }

  @Override
  public Node getNode() {
    return this.serviceDiscoveryServer.getNode();
  }

  @Override
  public void applyOperation(ServiceDiscoveryClosure closure) {
    if (!isLeader()) {
      handlerNotLeaderError(closure);
      return;
    }
    final Task task = new Task();
    // 写入本地日志，将作为 AppendEntries RPC 请求的来源 -> Followers
    task.setData(closure.getServiceDiscoveryOperation().serialize());
    // 触发 Leader 节点上的状态机 onApply 方法
    task.setDone(closure);
    this.serviceDiscoveryServer.getNode().apply(task);
  }

  private ServiceDiscoveryStateMachine getFsm() {
    return this.serviceDiscoveryServer.getFsm();
  }

  private boolean isLeader() {
    return getFsm().isLeader();
  }

  private void handlerNotLeaderError(final Closure closure) {
    logger.error("No Leader node : {}", getNode().getNodeId());
    closure.run(new Status(RaftError.EPERM, "Not leader"));
  }

}

RegistrationRpcProcessor 核心代码如下：

public class RegistrationRpcProcessor implements RpcProcessor<ServiceDiscoveryOuter.Registration> {
  ...
  @Override
  public void handleRequest(RpcContext rpcContext, Registration registration) {
    ServiceInstance serviceInstance = convertServiceInstance(registration);

    String serviceName = registration.getServiceName();

    final Kind kind = Kind.REGISTRATION;

    ServiceDiscoveryOperation op = new ServiceDiscoveryOperation(kind, serviceInstance);

    final ServiceDiscoveryClosure closure = new ServiceDiscoveryClosure(op) {
      @Override
      public void run(Status status) {
        if (!status.isOk()) {
          logger.warn("Closure status is : {} at the {}", status, rpcProcessorService.getNode());
          return;
        }
        rpcContext.sendResponse(response(status));
        logger.info("'{}' has been handled ,serviceName : '{}' , result : {} , status : {}",
            kind, serviceName, getResult(), status);
      }
    };

    this.rpcProcessorService.applyOperation(closure);
  }
  ...
}

4.6 ServiceDiscoveryClient 客户端

同样我们先将官网的示例 copy。
官网的示例大概做了以下动作

1.更新 Configuration
2.初始化 CliClientService
3.通过 RouteTable 获取到 Leader 节点
3.执行对应的请求方法

我们可以将 main 方法改造成 init 方法，因为我们后面会采用 spring web 的方式来进行注册的测试，而不是采用 main 方式来启动 client。
将两个配置参数 groupId，registerAddress 以构造函数的方式进行注入。

public class ServiceDiscoveryClient {

  private String groupId = "service-discovery";

  /**
   * 127.0.0.1:8083
   */
  private String registerAddress;

  private CliClientService cliClientService;

  private RpcClient rpcClient;

  public ServiceDiscoveryClient(String groupId, String registerAddress) {
    this.groupId = groupId;
    this.registerAddress = registerAddress;
  }

  public void init(final String[] args) {
    ServiceDiscoveryGrpcHelper.initGRpc();

    final Configuration conf = new Configuration();
    if (!conf.parse(registerAddress)) {
      throw new IllegalArgumentException("Fail to parse conf:" + registerAddress);
    }
    RouteTable.getInstance().updateConfiguration(groupId, conf);

    final CliClientServiceImpl cliClientService = new CliClientServiceImpl();
    cliClientService.init(new CliOptions());
    this.cliClientService = cliClientService;
    this.rpcClient = cliClientService.getRpcClient();
  }
  ...
}

下面我们将需求 3 和 4 重构成一个通用的方法

  public <R> R invoke(Object request) throws Throwable {
    if (!RouteTable.getInstance().refreshLeader(cliClientService, groupId, 1000).isOk()) {
      throw new IllegalStateException("Refresh leader failed");
    }

    PeerId leader = RouteTable.getInstance().selectLeader(groupId);
    return (R) rpcClient.invokeSync(leader.getEndpoint(), request, TimeUnit.SECONDS.toMillis(5));
  }

4.7 实现 ServiceDiscovery 接口

以 register 注册方法为例
我们只需要实现两步
1.构造 Registration 参数对象
2.采用 RPC 的方式调用注册方法
核心代码如下：

  @Override
  public void register(ServiceInstance serviceInstance) {
    // 调用 RPC
    ServiceDiscoveryOuter.Registration registration = buildRegistration(serviceInstance, false);
    try {
      serviceDiscoveryClient.invoke(registration);
    } catch (Throwable e) {
      e.printStackTrace();
    }
  }

4.8 Spring Web 项目集成

构建一个 Spring Web 的工程，在 Spring Application 启动的时候，初始化 JRaftServiceDiscovery，将 web 服务信息注册到注册中心即可。
pom 文件引入依赖

  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.3.0.RELEASE</version>
  </parent>

  <dependencies>
    ...
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
     ...
 </dependencies>

启动类 MyApplication，这里为了测试方便，配置都是写死的，实际可以从 Spring Environment 获取 Spring 中的环境变量

在 Spring Application 启动中，初始化 JRaftServiceDiscovery
再调用 register 方法注册服务实例即可

@RestController
@SpringBootApplication
public class MyApplication implements ApplicationListener<ApplicationReadyEvent> {

    @RequestMapping("/")
    String home() {
        return "Hello World!";
    }

    public static void main(String[] args) {
        SpringApplication.run(MyApplication.class, args);
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        ConfigurableApplicationContext applicationContext = event.getApplicationContext();
        ConfigurableEnvironment environment = applicationContext.getEnvironment();
        JRaftServiceDiscovery jRaftServiceDiscovery = new JRaftServiceDiscovery();
        environment.getSystemProperties().put("service.discovery.jraft.registry.address","127.0.0.1:9081,127.0.0.1:9082,127.0.0.1:9083");
        jRaftServiceDiscovery.initialize(environment.getSystemProperties());
        DefaultServiceInstance serviceInstance = new DefaultServiceInstance();
        serviceInstance.setHost("127.0.0.1");
        serviceInstance.setId("1");
        serviceInstance.setPort(8080);
        serviceInstance.setServiceName("test1");
        jRaftServiceDiscovery.register(serviceInstance);
    }
}

5.测试

为了日志观察方便，我们可以在 ServiceDiscoveryRequestHandlerFactory 类中增加一个打印方法，每次操作之后打印 serviceNameToInstancesStorage 集合的元素

public class ServiceDiscoveryRequestHandlerFactory {
 ...
 public synchronized void storage(String id, String serviceName, ServiceInstance serviceInstance) {
    Map<String, ServiceInstance> serviceInstancesMap = serviceNameToInstancesStorage
        .computeIfAbsent(serviceName, n -> new LinkedHashMap<>());
    serviceInstancesMap.put(id, serviceInstance);
    print();
  }

  private void print() {
    serviceNameToInstancesStorage.forEach((k, v) -> {
      logger.info(" key :{}", k);
      v.forEach((nk, nv) -> {
        logger.info(" n_key :{} + n_value:{}", nk, nv);
      });
    });
  }
...
}

编辑服务端启动参数三个服务端参数分别为

/tmp/server1 service-discovery 127.0.0.1:9081 127.0.0.1:9081,127.0.0.1:9082,127.0.0.1:9083
/tmp/server2 service-discovery 127.0.0.1:9082 127.0.0.1:9081,127.0.0.1:9082,127.0.0.1:9083
/tmp/server3 service-discovery 127.0.0.1:9083 127.0.0.1:9081,127.0.0.1:9082,127.0.0.1:9083

依次启动 ServiceDiscoveryServer 服务。
再启动 Spring MyApplication，为了测试方便，我们可以直接采用写死配置的方式来进行调试，利用 debug 模式，来模拟多个 web 服务的注册

每次执行完，再替换参数，再次执行即可
核心代码如下：

DefaultServiceInstance serviceInstance = new DefaultServiceInstance();
serviceInstance.setHost("127.0.0.1");
serviceInstance.setId("2");
serviceInstance.setPort(8082);
serviceInstance.setServiceName("test2");
jRaftServiceDiscovery.register(serviceInstance);

观察每个注册中心服务端的打印信息
第一次注册
ServiceDiscoveryServer1：

ServiceDiscoveryServer2：

第二次注册
ServiceDiscoveryServer1：

ServiceDiscoveryServer2：

ServiceDiscoveryServer3：

可以看到每次服务注册的信息，都被同步到了三个服务端节点中。