SOFARegistry

https://www.sofastack.tech/projects/sofa-registry/
Admin
SOFARegistry is a production-grade, high-timeliness, and highly available service registration...
Tokens:30,069
Snippets:230
Trust Score:8.6
Update:4 months ago
Show doc for...
Tokens:
Raw
### Start SOFARegistry in Production Mode (MySQL)

Source: https://www.sofastack.tech/projects/sofa-registry/server-quick-start

Start the SOFARegistry in production mode using MySQL as the metadata store. Ensure the application.properties are configured correctly, and the JDBC_PASSWORD environment variable is set if needed.

```bash
sh bin/integration/start.sh
```

--------------------------------

### Start SOFARegistry in Development Mode (H2)

Source: https://www.sofastack.tech/projects/sofa-registry/server-quick-start

Start the SOFARegistry in development mode using the provided script. This mode utilizes an H2 database and is suitable for local testing.

```bash
sh bin/integration/start_dev.sh
```

--------------------------------

### Example Output After Subscribing

Source: https://www.sofastack.tech/projects/sofa-registry/client-quick-start

This output is shown in the console if the publisher program is run before the subscriber program, indicating successful data reception.

```text
receive data success, dataId: com.alipay.test.demo.service:1.0@DEFAULT, data: DefaultUserData{zoneData={DEFAULT_ZONE=[10.10.1.1:12200?xx=yy]}, localZone='DEFAULT_ZONE'}
```

--------------------------------

### Consumer Stored Service List Example

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/origins-ant-s-practical-path-to-application-level-service-discovery

This example shows how a consumer might store service lists in memory, mapping service names to provider addresses and their associated parameters. This can lead to high memory usage if many services are subscribed.

```text
com.alipay.sample.FooService:
        - 1.1.1.1:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
        - ...
        - 1.1.1.99:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0

com.alipay.sample.BarService:
        - 1.1.1.1:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
        - ...
        - 1.1.1.99:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0

//...中间省略

com.alipay.sample.Service07:
        - 1.1.1.1:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
        - ...
        - 1.1.1.99:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
```

--------------------------------

### Data Server Bootstrap Start Method

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-non-destructive-o-and-m

This method initiates the startup sequence for the Data Server, including opening various servers, renewing node information, and registering task metrics. It also sets up a shutdown hook for graceful termination.

```java
/** start dataserver */
public void start() {
  try {
    LOGGER.info("begin start server");
    LOGGER.info("release properties: {}", ServerEnv.getReleaseProps());
    LOGGER.info("the configuration items are as follows: " + dataServerConfig.toString());

    ReporterUtils.enablePrometheusDefaultExports();

    openDataServer();

    openDataSyncServer();

    openHttpServer();

    renewNode();
    fetchProviderData();

    systemPropertyProcessorManager.startFetchMetaSystemProperty();

    startScheduler();

    TaskMetrics.getInstance().registerBolt();

    postStart();
    Runtime.getRuntime().addShutdownHook(new Thread(this::doStop));

    LOGGER.info("start server success");
  } catch (Throwable e) {
    throw new RuntimeException("start server error", e);
  }
}
```

--------------------------------

### Provider Stored Publisher Data Example

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/origins-ant-s-practical-path-to-application-level-service-discovery

This example illustrates how a provider's data is stored in the registry, with each service name mapping to a list of provider addresses. This can lead to significant data redundancy if a provider offers multiple services with similar configurations.

```text
com.alipay.sample.FooService:
        - 1.1.1.1:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0

com.alipay.sample.BarService:
        - 1.1.1.1:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0

//...中间省略

com.alipay.sample.Service100:
        - 1.1.1.1:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
```

--------------------------------

### Step 1: Create Lock Record on Startup

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-registry-meta

SQL statement for inserting a new lock record when a service starts. Uses `ON DUPLICATE KEY UPDATE` to handle cases where the lock might already exist.

```sql
<insert id="competeLockOnInsert" parameterType="com.alipay.sofa.registry.jdbc.domain.DistributeLockDomain">
        <![CDATA[
       INSERT /*+ QUERY_TIMEOUT(2000000) */ INTO distribute_lock
       (
           data_center,
           lock_name,
           owner,
           duration,
           gmt_create,
           gmt_modified,
           `term`,
           `term_duration`
       )
       VALUES
       (
           #{dataCenter},
           #{lockName},
           #{owner},
           #{duration},
           NOW(3),
           NOW(3),
           1,
           0
       )
       ON DUPLICATE KEY UPDATE lock_name = #{lockName}
       ]]>
    </insert>

```

--------------------------------

### Extract SOFARegistry Installation Package

Source: https://www.sofastack.tech/projects/sofa-registry/server-quick-start

Extract the downloaded SOFARegistry installation package (registry-all.tgz) to a desired location. This is for users who prefer not to build from source.

```bash
tar -zxvf registry-all.tgz
cd registry-all
```

--------------------------------

### Application-Level Service Discovery: Registry Push Data

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/origins-ant-s-practical-path-to-application-level-service-discovery

This example shows the data pushed by the registry to a consumer after adopting application-level service discovery. The data is organized by application name, containing a list of provider addresses for that application.

```text
applicationB:
        - 1.1.1.1:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
        - ...
        - 1.1.1.100:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
```

--------------------------------

### Application-Level Routing: Consumer Data Storage

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/origins-ant-s-practical-path-to-application-level-service-discovery

This example illustrates the data structure within a consumer after implementing application-level routing. It stores a mapping from application names to their respective services and provider addresses, reducing memory footprint.

```text
applicationB:
        - com.alipay.sample.FooService
  - com.alipay.sample.BarService
  - ...
        - com.alipay.sample.Service50

applicationB:
        - 1.1.1.1:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
        - ...
        - 1.1.1.100:12200?app=applicationB&_SERIALIZETYPE=hessian2&_TIMEOUT=3000&zone=zone1&version=1&_WARMUPTIME=0
```

--------------------------------

### Renew Node and Session Information

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-non-destructive-o-and-m

This method is responsible for renewing node information with the meta server, initializing session leases, and starting a background process to continuously refresh cluster information.

```java
  private void renewNode() {
    metaServerService.renewNode();
    // init session lease with first renew
    for (ProcessId processId : metaServerService.getSessionProcessIds()) {
      sessionLeaseManager.renewSession(processId);
    }
    metaServerService.startRenewer();
  }

```

--------------------------------

### Get Datum from Cache or Fetch and Update

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-publish-subscription-push

This method attempts to retrieve address list data (SubDatum) from the session cache. If the data is present and its version is greater than or equal to the expected version, it's returned directly (cache hit). Otherwise, the cache is invalidated, and the data is fetched anew and updated in the cache before being returned. Cache hit and miss counters are incremented accordingly.

```java
// 获取地址列表
SubDatum getDatum(String dataCenter, String dataInfoId, long expectVersion) {
  Key key = new Key(DatumKey.class.getName(), new DatumKey(dataInfoId, dataCenter));
  Value value = sessionCacheService.getValueIfPresent(key);
  if (value != null) {
    SubDatum datum = (SubDatum) value.getPayload();
    // 缓存中的数据更新，以缓存中的数据为准
    if (datum != null && datum.getVersion() >= expectVersion) {
      // the expect version got
      CACHE_HIT_COUNTER.inc();
      return datum;
    }
  }
  CACHE_MISS_COUNTER.inc();
  // invalidate缓存中的数据
  // the cache is too old
  sessionCacheService.invalidate(key);
  // 获取并更新缓存
  value = sessionCacheService.getValue(key);
  return value == null ? null : (SubDatum) value.getPayload();
}

```

--------------------------------

### Initialize SOFARegistry Client

Source: https://www.sofastack.tech/projects/sofa-registry/java-sdk

Demonstrates how to build and initialize a SOFARegistry client instance using a configuration builder.

```APIDOC
## Initialize SOFARegistry Client

### Description
Builds and initializes a SOFARegistry client instance. Configuration properties like registry endpoint and port are set using `DefaultRegistryClientConfigBuilder`.

### Code Example
```java
RegistryClientConfig config =  DefaultRegistryClientConfigBuilder.start().setRegistryEndpoint("127.0.0.1").setRegistryEndpointPort(9603).build();
DefaultRegistryClient registryClient = new DefaultRegistryClient(config);
registryClient.init();
```

### DefaultRegistryClientConfigBuilder Properties

| Property Name | Property Type | Description |
|---|---|---|
| instanceId | String | Instance ID, must be the same for publish and subscribe. Data unique identifier is composed of dataId+group+instanceId. Default value is `DEFAULT_INSTANCE_ID`. |
| zone | String | Zone to which the unit belongs. Default value is `DEFAULT_ZONE`. |
| registryEndpoint | String | Address of any Session node on the server. |
| registryEndpointPort | int | The session.server.httpServerPort configured for the Session node. Default value is 9603. |
| dataCenter | String | Data center. Default value is `DefaultDataCenter`. |
| appName | String | Application name. |
| connectTimeout | int | Connection timeout to the server. Default value is 3000ms. |
| socketTimeout | int | Timeout for accessing the server's REST interface. Default value is 3000ms. |
| invokeTimeout | int | Timeout for invoking server services. Default value is 1000ms. |
| recheckInterval | int | Interval for checking the task queue. Default value is 500ms. |
| observerThreadCoreSize | int | Core thread size of the thread pool for processing data pushed by the server. Default value is 5. |
| observerThreadMaxSize | int | Maximum thread size of the thread pool for processing data pushed by the server. Default value is 10. |
| observerThreadQueueLength | int | Queue size of the thread pool for processing data pushed by the server. Default value is 1000. |
| syncConfigRetryInterval | int | Interval for synchronizing the server list. Default value is 30000ms. |

```

--------------------------------

### List Entries Tail Recursively

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyza-data-table-listening

The listToTail method recursively lists stable entries starting from a given ID, processing each entry and updating the current start ID to the maximum ID encountered. It includes a small sleep to prevent overwhelming the system during high write loads.

```java
private long listToTail(EntryCallable<T> callable, final long start, final int page) {
  long curStart = start;
  while (true) {
    List<T> entries = listStableEntries(curStart, page);
    if (CollectionUtils.isEmpty(entries)) {
      break;
    }
    for (T entry : entries) {
      callable.onEntry(entry);
      curStart = Math.max(curStart, entry.getId());
    }
    ConcurrentUtils.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
  }
  return curStart;
}
```

--------------------------------

### Create MySQL Database and Table for SOFARegistry

Source: https://www.sofastack.tech/projects/sofa-registry/server-quick-start

Create the 'registrymetadb' database and import the necessary table structure using the provided SQL script. This is required for production environments using MySQL.

```bash
echo "create database registrymetadb " | mysql -u username -p
mysql -u username -p registrymetadb < create_table.sql
```

--------------------------------

### Get Client Handlers

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-push-switch

Retrieves the collection of meta client handlers. These handlers are registered with Spring and process incoming messages.

```java
protected Collection<ChannelHandler> getClientHandlers() {
    return metaClientHandlers;
}
```

```java
@Resource(name = "metaClientHandlers")
private Collection<ChannelHandler> metaClientHandlers;
```

--------------------------------

### Check Meta Role Health Status

Source: https://www.sofastack.tech/projects/sofa-registry/server-quick-start

Verify the health of the meta role by sending a GET request to the /health/check endpoint. This helps confirm the registry is operational.

```bash
$ curl http://localhost:9615/health/check
{"success":true,"message":"..."}
```

--------------------------------

### Publish Data with SOFARegistry Client

Source: https://www.sofastack.tech/projects/sofa-registry/client-quick-start

This snippet demonstrates how to initialize the SOFARegistry client, register a publisher, and publish data. Ensure the registry endpoint and port are correctly configured.

```java
// 构建客户端实例
RegistryClientConfig config = DefaultRegistryClientConfigBuilder.start().setRegistryEndpoint("127.0.0.1").setRegistryEndpointPort(9603).build();
DefaultRegistryClient registryClient = new DefaultRegistryClient(config);
registryClient.init();

// 构造发布者注册表
String dataId = "com.alipay.test.demo.service:1.0@DEFAULT";
PublisherRegistration registration = new PublisherRegistration(dataId);

// 将注册表注册进客户端并发布数据
registryClient.register(registration, "10.10.1.1:12200?xx=yy");
```

--------------------------------

### Subscribe to Data with SOFARegistry Client

Source: https://www.sofastack.tech/projects/sofa-registry/client-quick-start

This snippet shows how to initialize the SOFARegistry client, create a data observer, register a subscriber, and subscribe to data. Data is received via callbacks.

```java
// 构建客户端实例
RegistryClientConfig config = DefaultRegistryClientConfigBuilder.start().setRegistryEndpoint("127.0.0.1").setRegistryEndpointPort(9603).build();
DefaultRegistryClient registryClient = new DefaultRegistryClient(config);
registryClient.init();

// 创建 SubscriberDataObserver 
SubscriberDataObserver subscriberDataObserver = new SubscriberDataObserver() {
    public void handleData(String dataId, UserData userData) {
            System.out.println("receive data success, dataId: " + dataId + ", data: " + userData);
    }
};

// 构造订阅者注册表，设置订阅维度，ScopeEnum 共有三种级别 zone, dataCenter, global
String dataId = "com.alipay.test.demo.service:1.0@DEFAULT";
SubscriberRegistration registration = new SubscriberRegistration(dataId, subscriberDataObserver);
registration.setScopeEnum(ScopeEnum.global);

// 将注册表注册进客户端并订阅数据，订阅到的数据会以回调的方式通知 SubscriberDataObserver
registryClient.register(registration);
```

--------------------------------

### Clone SOFARegistry Repository and Build

Source: https://www.sofastack.tech/projects/sofa-registry/server-quick-start

Clone the SOFARegistry source code from GitHub and build the server release package using Maven. This is the first step for source-based deployment.

```bash
git clone https://github.com/sofastack/sofa-registry.git
cd sofa-registry
mvn clean package -Pserver-release -Dmaven.test.skip=true
cp ./server/distribution/all/target/registry-all.tgz <somewhere>
cd <somewhere>
tar -zxvf registry-all.tgz
cd registry-all
```

--------------------------------

### CompressorMap Initialization in CompressUtils

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-communication-data-compression

This snippet shows how the CompressUtils class loads compressor implementations into a static map, keyed by encoding type, for efficient retrieval during compression operations.

```java
public class CompressUtils {
    private static final Map<String, Compressor> compressorMap = new HashMap<>();

    static {
        // Load compressors on startup
        compressorMap.put(GzipCompressor.ENCODING, new GzipCompressor());
        compressorMap.put(ZstdCompressor.ENCODING, new ZstdCompressor());
    }

    public static Compressor findCompressor(String encoding) {
        return compressorMap.get(encoding);
    }

    // Other methods for compression and decompression...
}
```

--------------------------------

### Connect and Request with URL

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-push-switch

Establishes a connection to a server URL and sends a request. This is used in the message exchange process.

```java
public Response request(Request request) throws RequestException {
    final URL url = request.getRequestUrl();
    ......
    connect(url);
    ......
}
```

```java
public Channel connect(URL url) {
    Client client = getClient();
    ......
    client = boltExchange.connect(serverType,getConnNum(),url,
                getClientHandlers().toArray(new ChannelHandler[0]));
    ......
}
```

--------------------------------

### Find Candidate Data Server for Leader

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-slottable

Identifies a suitable data server to become a new leader for a given slot by selecting the least busy server among available followers, excluding specified servers.

```java
Set<Integer> leaderSlots = dataNodeSlot.getLeaders();
Map<String, List<Integer>> dataServers2Followers = Maps.newHashMap();
//1. 从 dataNodeSlot 获取 leaderDataServer 节点leader的所有slotId: leaderSlots
for (int slot : leaderSlots) {
  //2. 从slotTableBuilder 中找出当前 slot 的follow
  List<String> followerDataServers = slotTableBuilder.getDataServersOwnsFollower(slot);
  //3. 去掉 excludes ，得到候选节点，因为 excludes 肯定不会是新的 leader 节点
  followerDataServers = getCandidateDataServers(excludes, null, followerDataServers);
  //4. 构建 候选节点到 slotId 集合的映射关系。
  for (String followerDataServer : followerDataServers) {
    List<Integer> followerSlots = 
        dataServers2Followers.computeIfAbsent(followerDataServer, k -> Lists.newArrayList());
    followerSlots.add(slot);
  }
}
if (dataServers2Followers.isEmpty()) {
  //当 leaderDataServer 节点的follow 都是 excludes 中的成员时候、那么就有可能是空的。
  return null;
}
List<String> dataServers = Lists.newArrayList(dataServers2Followers.keySet());
//按照 候选节点的 leader的 slot 个数升序排序、也就是也就是找到那个最不忙的，感兴趣可以查看 leastLeadersFirst 方法内部实现。
dataServers.sort(Comparators.leastLeadersFirst(slotTableBuilder));
final String selectedDataServer = dataServers.get(0);
List<Integer> followers = dataServers2Followers.get(selectedDataServer);
return Tuple.of(selectedDataServer, followers.get(0));
```

--------------------------------

### Guava Cache for Compression Results

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-communication-data-compression

Illustrates the use of Guava's Cache for storing compression results to reduce CPU load. It shows how to retrieve from cache first, and if not found, perform compression and then cache the result.

```java
public class CompressionService {
    private final LoadingCache<String, byte[]> compressionCache;

    public CompressionService() {
        compressionCache = CacheBuilder.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(10, TimeUnit.MINUTES)
            .build(
                new CacheLoader<String, byte[]>() {
                    @Override
                    public byte[] load(String key) throws IOException {
                        // Logic to perform actual compression based on key
                        // For example, retrieve original data, compress it, and return
                        byte[] originalData = getOriginalData(key);
                        Compressor compressor = CompressUtils.findCompressor("gzip"); // Example encoding
                        return compressor.compress(originalData);
                    }
                });
    }

    public byte[] getCompressedData(String key) throws ExecutionException, IOException {
        return compressionCache.get(key);
    }

    private byte[] getOriginalData(String key) { 
        // Placeholder for fetching original data
        return new byte[0]; 
    }
}
```

--------------------------------

### Initialize SOFARegistry Client

Source: https://www.sofastack.tech/projects/sofa-registry/java-sdk

Initialize the SOFARegistry client with configuration, including the registry endpoint and port. The DefaultRegistryClientConfigBuilder allows setting various connection and timeout properties.

```java
RegistryClientConfig config =  DefaultRegistryClientConfigBuilder.start().setRegistryEndpoint("127.0.0.1").setRegistryEndpointPort(9603).build();
DefaultRegistryClient registryClient = new DefaultRegistryClient(config);
registryClient.init();
```

```java
public class DefaultRegistryClientConfigBuilder {
    private String  instanceId;
    private String  zone                      = DEFAULT_ZONE;
    private String  registryEndpoint;
    private int     registryEndpointPort      = 9603;
    private String  dataCenter                = DEFAULT_DATA_CENTER;
    private String  appName;
    private int     connectTimeout            = 3000;
    private int     socketTimeout             = 3000;
    private int     invokeTimeout             = 1000;
    private int     recheckInterval           = 500;
    private int     observerThreadCoreSize    = 5;
    private int     observerThreadMaxSize     = 10;
    private int     observerThreadQueueLength = 1000;
    private int     syncConfigRetryInterval   = 30000;
}
```

--------------------------------

### Gzip Compression Implementation

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-communication-data-compression

Provides the core implementation for Gzip compression, including the encoding constant and the compression/decompression logic.

```java
public class GzipCompressor implements Compressor {
    public static final String ENCODING = "gzip";

    @Override
    public byte[] compress(byte[] data) throws IOException {
        // Gzip compression logic
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        try (GZIPOutputStream gzipStream = new GZIPOutputStream(out)) {
            gzipStream.write(data);
        }
        return out.toByteArray();
    }

    @Override
    public byte[] decompress(byte[] data) throws IOException {
        // Gzip decompression logic
        ByteArrayInputStream in = new ByteArrayInputStream(data);
        try (GZIPInputStream gzipStream = new GZIPInputStream(in))
        {
            return gzipStream.readAllBytes();
        }
    }

    @Override
    public String getEncoding() {
        return ENCODING;
    }
}
```

--------------------------------

### balanceLeaderSlots Method

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-slottable

Handles the balancing of leader slots across data servers. It first attempts to upgrade high leaders and then migrates high leaders if necessary.

```java
  private boolean balanceLeaderSlots() {
    //这里就是找到每一个节点 dataServer 作为leader的 slot 个数的最大天花板值----> 
    //容易想到的方案肯定是平均方式、一共有 slotNum 个slot、
    //将这些slot的leader归属平均分配给 currentDataServer
    final int leaderCeilAvg = MathUtils.divideCeil(slotNum, currentDataServers.size());
    if (upgradeHighLeaders(leaderCeilAvg)) {
      //如果有替换过 leader、那么就直接返回、不用进行 migrateHighLeaders 操作
      return true;
    }
    if (migrateHighLeaders(leaderCeilAvg)) {
      //经过上面的 upgradeHighLeaders 操作
      //不能找到 follow 进行迁移、因为所有的follow也都很忙、在 exclude 当中、
      //所以没法找到一个follow进行迁移。那么我们尝试迁移 follow。
      //因为 highLeader 的所有 follower 都是比较忙、所以需要将这些忙的节点进行迁移、期待给这些 highLeader 所负责的 slot 替换一些比较清闲的 follow
      return true;
    }
    return false;
  }
```

--------------------------------

### Generate Verification Data by ClientId (v2)

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/sofaregistry-talk-about-the-data-consistency-of-service-discovery

In v2, this method generates verification data based on ClientId. It iterates through all responsible clients, creates a DistroClientVerifyInfo, serializes it, and returns it as DistroData objects.

```java
@Overridepublic List<DistroData> getVerifyData() {
List<DistroData> result = new LinkedList<>();
for (String each : clientManager.allClientId()) 
{
Client client = clientManager.getClient(each);
if (null == client || !client.isEphemeral()) 
{
continue;
}
if (clientManager.isResponsibleClient(client)) {
// TODO add revision for client.
DistroClientVerifyInfo verifyData = new DistroClientVerifyInfo(client.getClientId(), 0);
DistroKey distroKey = new DistroKey(client.getClientId(), TYPE);
DistroData data = new DistroData(distroKey,
ApplicationUtils.getBean(Serializer.class).serialize(verifyData));
data.setType(DataOperation.VERIFY);
result.add(data);
}
}
return result;
}
```

--------------------------------

### Subscribe to Data with SOFARegistry

Source: https://www.sofastack.tech/projects/sofa-registry/java-sdk

Subscribe to data by implementing SubscriberDataObserver to handle incoming data and creating a SubscriberRegistration specifying the data ID, app name, and scope. The received data is delivered via the observer's handleData method.

```java
// 创建 SubscriberDataObserver 
SubscriberDataObserver subscriberDataObserver = new SubscriberDataObserver() {
    @Override
    public void handleData(String dataId, UserData userData) {
        System.out.println("receive data success, dataId: " + dataId + ", data: " + userData);
    }
};

// 构造订阅者注册表，设置订阅维度，ScopeEnum 共有三种级别 zone, dataCenter, global
String dataId = "com.alipay.test.demo.service:1.0@DEFAULT";
SubscriberRegistration registration = new SubscriberRegistration(dataId, subscriberDataObserver);
registration.setGroup("TEST_GROUP");
registration.setAppName("TEST_APP");
registration.setScopeEnum(ScopeEnum.global);

// 将注册表注册进客户端并订阅数据，订阅到的数据会以回调的方式通知 SubscriberDataObserver
Subscriber subscriber = registryClient.register(registration);
```

```java
void handleData(String dataId, UserData data);
```

```java
public interface UserData {
    Map<String, List<String>> getZoneData();
    String getLocalZone();
}
```

--------------------------------

### Publish Data with SOFARegistry

Source: https://www.sofastack.tech/projects/sofa-registry/java-sdk

Publish data by creating a PublisherRegistration with data ID, group, and app name, then registering it with the client. You can also re-publish data to overwrite previous entries.

```java
// 构造发布者注册表
PublisherRegistration registration = new PublisherRegistration("com.alipay.test.demo.service:1.0@DEFAULT");
registration.setGroup("TEST_GROUP");
registration.setAppName("TEST_APP");

// 将注册表注册进客户端并发布数据
Publisher publisher = registryClient.register(registration, "10.10.1.1:12200?xx=yy");

// 如需覆盖上次发布的数据可以使用发布者模型重新发布数据
publisher.republish("10.10.1.1:12200?xx=zz");
```

--------------------------------

### Subscribe Data

Source: https://www.sofastack.tech/projects/sofa-registry/java-sdk

Explains how to subscribe to data using the SOFARegistry Java SDK, including the `SubscriberDataObserver` callback and `SubscriberRegistration`.

```APIDOC
## Subscribe Data

### Description
Subscribes to data from SOFARegistry. This requires implementing the `SubscriberDataObserver` interface for handling incoming data and creating a `SubscriberRegistration` to specify subscription details and the observer.

### Code Example
```java
// Create SubscriberDataObserver
SubscriberDataObserver subscriberDataObserver = new SubscriberDataObserver() {
    @Override
    public void handleData(String dataId, UserData userData) {
        System.out.println("receive data success, dataId: " + dataId + ", data: " + userData);
    }
};

// Construct subscriber registration, set subscription dimension. ScopeEnum has three levels: zone, dataCenter, global
String dataId = "com.alipay.test.demo.service:1.0@DEFAULT";
SubscriberRegistration registration = new SubscriberRegistration(dataId, subscriberDataObserver);
registration.setGroup("TEST_GROUP");
registration.setAppName("TEST_APP");
registration.setScopeEnum(ScopeEnum.global);

// Register the registration into the client and subscribe to data. Subscribed data will be notified to SubscriberDataObserver via callback.
Subscriber subscriber = registryClient.register(registration);
```

### SubscriberDataObserver Interface

```java
void handleData(String dataId, UserData data);
```

#### Parameters

| Name | Type | Description |
|---|---|---|
| dataId | String | Data ID |
| data | UserData | Data content |

### UserData Interface

```java
Map<String, List<String>> getZoneData();
String getLocalZone();
```

* `getLocalZone()`: Returns the current zone.
* `getZoneData()`: Returns data with zone as key and data for each zone as value.

### SubscriberRegistration Properties

| Property Name | Property Type | Description |
|---|---|---|
| dataId | String | Data ID. Must be the same for publish and subscribe. The unique data identifier is composed of `dataId` + `group` + `instanceId`. |
| group | String | Data group. Must be the same for publish and subscribe. The unique data identifier is composed of `dataId` + `group` + `instanceId`. Default value is `DEFAULT_GROUP`. |
| appName | String | Application appName. |
| scopeEnum | ScopeEnum | Enumeration value, representing one of the three dimensions: Zone, DataCenter, Global. |
| subscriberDataObserver | SubscriberDataObserver | Callback interface after receiving server data. |

```

--------------------------------

### Data Server Graceful Shutdown - Blacklist and Slot Migration

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-non-destructive-o-and-m

This snippet demonstrates the core logic within DataServer's graceful shutdown, focusing on adding the server to the meta-server's blacklist and waiting for all data slots to be migrated away. It ensures data integrity by preventing shutdown until all slots are reassigned.

```java
addBlacklistRetryer.call(
    () -> {
      LOGGER.info("[GracefulShutdown] add self to blacklist");
      metaServerService.addSelfToMetaBlacklist();
      return true;
    });
addBlacklistRetryer.call(
    () -> {
      if (fetchStopPushService.isStopPushSwitch()) {
        return true;
      }
      SlotTableStatusResponse statusResponse = metaServerService.getSlotTableStatus();
      if (statusResponse.isProtectionMode()) {
        return true;
      }
      LOGGER.info("[GracefulShutdown] wait no slot");
      if (slotManager.hasSlot()) {
        throw new RuntimeException("current data server still own slot, waiting...");
      }
      return true;
    });
LOGGER.info("add data self to blacklist successfully");
```

--------------------------------

### Zstd Compression Implementation

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-communication-data-compression

Details the implementation for Zstandard (zstd) compression, including its encoding constant and the methods for compressing and decompressing data.

```java
public class ZstdCompressor implements Compressor {
    public static final String ENCODING = "zstd";

    @Override
    public byte[] compress(byte[] data) throws IOException {
        // Zstd compression logic using a library like zstd-jni
        return Zstd.compress(data);
    }

    @Override
    public byte[] decompress(byte[] data) throws IOException {
        // Zstd decompression logic
        return Zstd.decompress(data);
    }

    @Override
    public String getEncoding() {
        return ENCODING;
    }
}
```

--------------------------------

### Perform Full Data List and Comparison

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyza-data-table-listening

The list method performs a full data retrieval and comparison. It fetches all entries, updates the container, and sets the lastLoadId. This is used for periodic full data correction and to compensate for potential gaps in the watch mechanism.

```java
private void list() {
  syncStart();
  try {
    C newContainer = containerFactory();
    long maxId = listToTail(newContainer::onEntry, 0, 1000);
    logger.info("end list to {}", maxId);
    preList(newContainer);
    this.container = newContainer;
    lastLoadId = maxId;
  } finally {
    syncEnd();
  }
}
```

--------------------------------

### Fetching and Comparing Configuration Data

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-push-switch

The doFetchData method retrieves current and persistent configuration data, compares versions, and returns a boolean indicating the success of the operation or if an update is needed.

```java
protected boolean doFetchData() {
    T expect = storage.get();
    E fetchData = fetchFromPersistence();

    if (fetchData == null) {
      ......
      return false;
    }

    if (fetchData.getVersion() < expect.getVersion()) {
      ......
      return false;
    } else if (fetchData.getVersion() == expect.getVersion()) {
      return true;
    }
    // do compare and set
    return doProcess(expect, fetchData);
  }
```

--------------------------------

### Blocking Queue Operations in WakeUpLoopRunnable

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-push-switch

Demonstrates the usage of a blocking queue (`bell`) for producer-consumer patterns within the WakeUpLoopRunnable. Includes methods for waiting and waking up.

```java
  // 阻塞队列
  private final ArrayBlockingQueue<Object> bell = new ArrayBlockingQueue<>(1);

  // 从队列中取出
  @Override
  public void waitingUnthrowable() {
    ConcurrentUtils.pollUninterruptibly(bell, getWaitingMillis(), TimeUnit.MILLISECONDS);
  }

  // 设置最长等待时间
  public abstract int getWaitingMillis();

  // 向队列中添加
  public void wakeup() {
    bell.offer(this);
  }
```

--------------------------------

### Execute Change and Fetch Address List

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-publish-subscription-push

This method is responsible for executing a data change event on the session server. It fetches the latest address list based on the expected version and performs version checks to ensure data consistency. If the fetched data is valid, it proceeds to update the local cache and notify relevant parties.

```java
boolean doExecuteOnChange(String changeDataInfoId, TriggerPushContext changeCtx) {
  final long expectVersion = changeCtx.getExpectDatumVersion();
  // 获取地址列表
  final SubDatum datum = getDatum(changeCtx.dataCenter, changeDataInfoId, expectVersion);
  if (datum == null) {
    // datum change, but get null datum, should not happen
    LOGGER.error("[changeNil] {},{},{!}", changeCtx.dataCenter, changeDataInfoId, expectVersion);
    return false;
  }
  // 判断版本号，获取到的最新版本号不应该比预期的小
  if (datum.getVersion() < expectVersion) {
    LOGGER.error(
        "[changeLessVer] {},{},{!}<{!}",
        changeCtx.dataCenter,
        changeDataInfoId,
        datum.getVersion(),
        expectVersion);
    return false;
  }
  onDatumChange(changeCtx, datum);
  return true;
}

```

--------------------------------

### Publish Data

Source: https://www.sofastack.tech/projects/sofa-registry/java-sdk

Details on how to publish data using the SOFARegistry Java SDK, including constructing the `PublisherRegistration` and re-publishing data.

```APIDOC
## Publish Data

### Description
Publishes data to SOFARegistry. This involves creating a `PublisherRegistration` object with data details and then registering it with the `registryClient`. Existing published data can be updated using the `republish` method.

### Code Example
```java
// Construct publisher registration
PublisherRegistration registration = new PublisherRegistration("com.alipay.test.demo.service:1.0@DEFAULT");
registration.setGroup("TEST_GROUP");
registration.setAppName("TEST_APP");

// Register the registration into the client and publish data
Publisher publisher = registryClient.register(registration, "10.10.1.1:12200?xx=yy");

// To overwrite previously published data, use the publisher model to re-publish data
publisher.republish("10.10.1.1:12200?xx=zz");
```

### PublisherRegistration Properties

| Property Name | Property Type | Description |
|---|---|---|
| dataId | String | Data ID. Must be the same for publish and subscribe. The unique data identifier is composed of `dataId` + `group` + `instanceId`. |
| group | String | Data group. Must be the same for publish and subscribe. The unique data identifier is composed of `dataId` + `group` + `instanceId`. Default value is `DEFAULT_GROUP`. |
| appName | String | Application appName. |

```

--------------------------------

### DataServer Leader/Follower Synchronization Logic

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-data-synchronization

This code snippet shows the logic for DataServers to determine whether to act as a leader or synchronize with the leader based on the slot's leadership status.

```Java
if (localIsLeader(slot)) {
   // 如果当前是 leader，则执行 session 同步或者 migrating
} else {
    // 如果当前不是 leader，则和 leader 同步数据
    syncLeader(slotState, syncLeaderIntervalMs, slotTableEpoch);
}
```

--------------------------------

### Assign Leader Slots in SOFARegistry

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-slottable

This method attempts to assign leader slots based on a 'fewer followers first' strategy. It prioritizes slots with fewer followers to prevent data loss. It also includes logic to promote a follower to a leader if the current leader is unavailable.

```java
private boolean tryAssignLeaderSlots(int highWatermark) {
    //按照 follows 节点的数量 从大到小排序 0比较特殊排在最后面,0 为什么比较特殊呢、因为无论怎么分配、
    //最终选择出来的leader一定不是该slot的follow、因为该slot的follow为空
    //优先安排 follow节点比较少的 Slot
    //其实这点也可以想明白的。这些没有 leader 的 slot 分配顺序肯定是要根据 follow节点越少的优先分配最好
    //以防止这个 follow 也挂了、那么数据就有可能会丢失了。
    List<Integer> leaders = 
        migrateSlotGroup.getLeadersByScore(new FewerFollowerFirstStrategy(slotTableBuilder));
    for (int slotId : leaders) {
      List<String> currentDataNodes = Lists.newArrayList(currentDataServers);
       //选择 nextLeader 节点算法?
      String nextLeader = 
          Selectors.slotLeaderSelector(highWatermark, slotTableBuilder, slotId) 
              .select(currentDataNodes);
      //判断nextLeader是否是当前slot的follow节点 将follow节点提升为主节点的。 
      boolean nextLeaderWasFollower = isNextLeaderFollowerOfSlot(slotId, nextLeader);
      // 将当前 slot 的 leader 节点用选择出来的 nextLeader 替换
      slotTableBuilder.replaceLeader(slotId, nextLeader);
      if (nextLeaderWasFollower) {
        //因为当前 Slot 将 follow节点提升为leader节点了、那么该 Slot 肯定 follows 个数又不够了、需要再次分配 follow 节点
        migrateSlotGroup.addFollower(slotId);
      }
    }
    return true;
  }
```

--------------------------------

### Fetch System Property Service

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-push-switch

Iterates through registered system data processors to find one that supports the given dataInfoId and then fetches the property.

```java
public boolean doFetch(String dataInfoId) {
    private Collection<AbstractFetchPersistenceSystemProperty> systemDataPersistenceProcessors = 
      new ArrayList<>();

    ......
    for (FetchSystemPropertyService systemDataProcessor : systemDataPersistenceProcessors) {
      if (systemDataProcessor.support(dataInfoId)) {
        return systemDataProcessor.doFetch();
      }
    }
    ......
  }
```

--------------------------------

### SlotTableBuilder Initialization

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-slottable

Initializes the SlotBuilder for each slot, copying data from the old SlotTable and handling leader and follower nodes. It also initializes a reverse map for efficient node-to-slot lookups.

```java
public void init(List<String> dataServers) {
    for (int slotId = 0; slotId < slotNums; slotId++) {
      Slot slot = initSlotTable == null ? null : initSlotTable.getSlot(slotId);
      if (slot == null) {
        getOrCreate(slotId);
        continue;
      }
     //1. 重新新建一个 SlotBuilder 将原来的 Slot 里面的数据拷贝过来
     //2. 拷贝 leader节点
      SlotBuilder slotBuilder = 
          new SlotBuilder(slotId, followerNums, slot.getLeader(), slot.getEpoch());
     //3. 拷贝 follow 节点。
      slotBuilder.addFollower(initSlotTable.getSlot(slotId).getFollowers())
      buildingSlots.put(slotId, slotBuilder);
    }
     //4. 初始化反向查询索引数据、通过 节点查询该节点目前管理哪些 slot
    initReverseMap(dataServers);
  }
```

--------------------------------

### Calculate Checksum for Service Data (v1)

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/sofaregistry-talk-about-the-data-consistency-of-service-discovery

This method calculates the MD5 checksum for all instances of a service based on their IP, port, weight, health status, and cluster name. It's used in the v1 version of node data verification.

```java
public synchronized void recalculateChecksum() {
List<Instance> ips = allIPs();
    StringBuilder ipsString = new StringBuilder();
    String serviceString = getServiceString();
    ipsString.append(serviceString);
    for (Instance ip : ips) {
    String string = ip.getIp() + ":" + ip.getPort() + "_" + ip.getWeight() + "_" + ip.isHealthy() + "_" + ip
    .getClusterName();
    ipsString.append(string);
    ipsString.append(',');
    }
    checksum = MD5Utils.md5Hex(ipsString.toString(), Constants.ENCODE);
    }
```

--------------------------------

### Save Provide Data with Version Check

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-push-switch

This Java code snippet shows the `saveProvideData` method, which is responsible for updating configuration data. It includes version checking to ensure only newer data overwrites existing data, database updates, and in-memory cache updates.

```java
public boolean saveProvideData(PersistenceData persistenceData, long expectVersion) {
    // 1.比较版本号
    if (persistenceData.getVersion() <= expectVersion) {
      ......
      return false;
    }

    // 2.更新数据库
    boolean success = provideDataRepository.put(persistenceData, expectVersion);

    if (success) {
      lock.writeLock().lock();
      try {
        // 3.更新内存
        provideDataCache.put(
            PersistenceDataBuilder.getDataInfoId(persistenceData), persistenceData);
      } catch (Throwable t) {
        ......
        return false;
      } finally {
        lock.writeLock().unlock();
      }
    }
    return success;
  }
```

--------------------------------

### Request Handling and Routing in Session Nodes

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-slottable

The request method prepares and sends batched requests to data nodes. It retrieves the appropriate Slot, sets metadata like epoch information, and determines the target URL using the SlotTable.

```java
private boolean request(BatchRequest batch) {
  final Slot slot = getSlot(batch.getSlotId());
  batch.setSlotTableEpoch(slotTableCache.getEpoch());
  batch.setSlotLeaderEpoch(slot.getLeaderEpoch());
  sendRequest(
      new Request() {
        @Override
        public Object getRequestBody() {
          return batch;
        }

        @Override
        public URL getRequestUrl() {
          //通过 slot 路由表找到对应的 leader data节点，这
          //个路由表是 心跳中从 Meta 节点获取来的。
          return getUrl(slot);
        }
      });
  return true;
}
```

--------------------------------

### Add SOFARegistry Client Dependency

Source: https://www.sofastack.tech/projects/sofa-registry/client-quick-start

Include the SOFARegistry client dependency in your Maven project's pom.xml file.

```xml
<dependency>
    <groupId>com.alipay.sofa</groupId>
    <artifactId>registry-client-all</artifactId>
    <version>${registry.client.version}</version>
</dependency>
```

--------------------------------

### Assign Follower Slots in SOFARegistry

Source: https://www.sofastack.tech/projects/sofa-registry/code-analyze/code-analyze-slottable

This method assigns follower slots to data nodes. It prioritizes slots that are missing the most followers. For each slot, it iterates through available candidates, sorting them by who is least busy (fewest followers and then fewest leaders), to find a suitable node to add as a follower.

```java
private boolean assignFollowerSlots() {
    //使用 FollowerEmergentScoreJury 排序得分策略表明
    // 某一个 slot 缺少越多 follow、排序越靠前。  
    List<MigrateSlotGroup.FollowerToAssign> followerToAssigns = 
        migrateSlotGroup.getFollowersByScore(new FollowerEmergentScoreJury());
    int assignCount = 0;
    for (MigrateSlotGroup.FollowerToAssign followerToAssign : followerToAssigns) {
      // 当前待分配的 slotId
      final int slotId = followerToAssign.getSlotId();
      // 当前 slotId 槽中还有多少待分配的 follow 从节点。依次迭代分配。
      for (int i = 0; i < followerToAssign.getAssigneeNums(); i++) {
        final List<String> candidates = Lists.newArrayList(currentDataServers);
        // 根据上文中的 DataNodeSlot 结构、依据 节点被作为follow 的slot的个数从小到大排序。
        // follows 个数一样、按照最少作为 leader 节点进行排序。
        // 其实最终目的就是找到最 "闲" 的那一台机器。
        candidates.sort(Comparators.leastFollowersFirst(slotTableBuilder));
        boolean assigned = false;
        for (String candidate : candidates) {
          DataNodeSlot dataNodeSlot = slotTableBuilder.getDataNodeSlot(candidate);
          //跳过已经是它的 follow 或者 leader 节点的Node节点
          if (dataNodeSlot.containsFollower(slotId) || dataNodeSlot.containsLeader(slotId)) {
            continue;
          }
          //给当前 slotId 添加候选 follow 节点。
          slotTableBuilder.addFollower(slotId, candidate);
          assigned = true;
          assignCount++;
          break;
        }
      }
    }
    return assignCount != 0;
  }
```
Tokens:
Raw