0 x00 the

We have made a preliminary analysis of the basic architecture of MetaServer through three articles, which have laid a solid foundation for our subsequent work.

In this series, we’ll continue with Data Servers, including some Session Servers. Because DataServer and MetaServer code implementation and architecture have similar basic routines, we are primarily concerned with the differences and characteristics of DataServer.

This article looks at the basic architecture of the DataServer program.

0 x01 ideas

The previous article focused on the system business itself, but this series of articles will focus on the analysis of design and architecture. That is, we will use multiple articles to summarize the implementation mechanism and architecture ideas of DataServer or SOFARegistry from multiple perspectives, so that we can learn how alibaba designs.

The specific learning methods are:

  • First, think of your own business scenarios, problems and solutions. Be sure to keep close contact with business, all from business to talk about design, are playing rogue;
  • Then look at ant Financial’s source code to see how they solve the problem and what the difference is with their own solution. The current code of Ant Financial may not be the most ideal solution (such as hash solution) in the current application scenario, but it is definitely the product of various power games and the best practice tempered by financial level.
  • After studying the code, I need to adjust my thinking and think again.
  • Then see if Ali’s solution can be improved in the future;
  • Finally, see if common ground can be extracted from the source code or if there are methods or patterns that can be reused.

When learning, pay attention to:

  • One of the essence of architectural design is balance and compromise. A system will choose different architectures in different periods, different data environments and different application scenarios, and the selection is essentially balancing some important points.
  • Focus on the temporal and spatial relationship between algorithms and other related logic — such a logical architectural relationship. In a system, these dimensions (space and time) criss-cross each other, making complexity very high. The purpose of our study is to separate these dimensions and simplify the interaction between them.
  • Look not only at the surface, but also at the underlying thinking and logic. Not only look at what is mentioned in the code comment, but also explore what is not mentioned in the code comment;
  • We should not only study isolated functions/components/modules in depth, but also re-examine these modules from the perspective of architecture and business scenarios, so that we can have a deeper understanding of the relationship between components and look at the system from a global perspective.
  • The change of mindset is the part you should care about most;

Because from multiple dimensions to analysis and design, such as business dimension and structure dimension, therefore in this series, may be some articles are concentrated in the model summary extraction, some articles are concentrated in the business implementation, the use of some articles will focus on the specific knowledge, also can appear a certain business module or code since the business and the phenomenon of realization in different article mentioned, I want you to know that.

0x02 Basic Architecture & Guidelines

2.1 SOFARegistry Overall architecture

First, let’s recall SOFARegistry’s overall architecture

  • The Client layer

Application server clusters. The Client layer is the application layer, where each application system programmatically uses the service publishing and service subscription capabilities of the service registry by relying on registrie-related Client JAR packages.

  • The Session layer

Session Server cluster. As the name implies, the Session layer is the Session layer, which communicates with the application server of the Client layer through a long connection and is responsible for receiving the service publication and service subscription requests from the Client. This layer only stores the publishing and subscription relationship of each service in memory. For specific service information, it only passes through and forwards between Client layer and Data layer. The Session layer is stateless and can be expanded as the application scale of the Client layer grows.

  • Data layer

Data server clusters. The Data layer stores the service registration Data of the application by means of fragment storage. Data is Hash sharded consistently based on the dataInfoId (unique identifier of each service data) and backed up with multiple copies to ensure high data availability. The following focuses on how the Data layer can smoothly expand and shrink as the Data scale grows without affecting the business.

  • Meta layer

Metadata server cluster. SOFARegistry acts as a service registry within the SOFARegistry architecture. SOFARegistry is a service registry that serves a wide range of application services. Meta clusters are Session and Data clusters that serve SOFARegistry. The Meta layer senses changes in Session and Data nodes and notifies other nodes in the cluster.

2.2 standard

There is no one-size-fits-all approach to what makes a good architecture for a program, and there are many books and articles about it, so let’s think about it in terms of the simplest and most straightforward way: in terms of results: static and dynamic.

  • Static: This perspective is that when you get new code, the first thing you look at is its directory structure. If the program’s directory structure is clear, see the directory structure can only make you can sort out this code logic, only from the file name will know what directory, it should belong to what modules, there will not be a single file should make you feel really in another directory, then the program from the static perspective, its architecture is excellent.
  • Dynamic: this point is when you just browse through the code, probably after you close your eyes, I can be constructed the application module in the brain, to know program is divided into several functional modules, clearly know the entry of the program, can frame its basic function of process and internal logic module interaction, then the program from the dynamic point of view, its architecture is excellent.

For example, if your application is based on SpringBoot, Bean construction and classification is very important, and if beans are handled well, it will be very helpful to clean up your dynamic architecture.

Let’s take a look at the basic architecture of DataServer.

0x03 Directory structure

The directory structure is as follows, and we can get an idea of what SOFAReistry is thinking about. Of course, because of business and architecture coupling, my categorization may not be entirely appropriate, but there are other ways of categorization, depending on how you think about it.

  • Program body: DataApplication;
  • Program entry and Bean: bootstrap;

Program basic business functions:

  • Network: remoting;
  • Auxiliary: utils;
  • HTTP: resource;
  • Cache: cache;
  • Thread: executor;

Business functions:

  • Renew;
  • Datasync;
  • Change;
  • The event;
  • The node;

The specific directories are as follows:

.├ ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ── ─ └ │ ├─ All exercises, all exercises, all exercises, all exercises, all exercises, all exercises Handler │ │ ├─ ├─ task │ ├─ disconnect │ ├─ forward │ ├─ Handler ├── renew ├─ resource ├─Copy the code

0x04 Basic Architecture

It is still similar to MetaServer, using the SpringBoot framework to build the overall.

@EnableDataServer
@SpringBootApplication
public class DataApplication {
    public static void main(String[] args) { SpringApplication.run(DataApplication.class, args); }}Copy the code

DataServerBeanConfiguration EnableDataServer this annotation will introduce the basic configuration.

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(DataServerBeanConfiguration.class)
public @interface EnableDataServer {
}
Copy the code

0 x05 configuration Bean

DataServer is a SpringBoot program. So use beans a lot.

DataServerBeanConfiguration role is to build all kinds of related configuration, from which you can see DataServer related modules and functions.

System initialization when the beans inside the DataServerBeanConfiguration by JavaConfig to register, mainly in the following several configuration class (configuration class will have change, specific content can be reference to the source code implementation) :

  • DataServerBootstrap
  • DataServerBootstrapConfigConfiguration,
    • CommonConfig
    • DataServerConfig
    • DataNodeStatus
    • PropertySplitter
  • DataServerStorageConfiguration
    • DatumCache
    • LocalDatumStorage
  • LogTaskConfigConfiguration
    • CacheDigestTask
  • SessionRemotingConfiguration
    • jerseyExchange
    • boltExchange
    • MetaNodeExchanger
    • DataNodeExchanger
    • DataServerCache
    • ForwardService
    • SessionServerConnectionFactory
    • DataServerConnectionFactory
    • MetaServerConnectionFactory
    • serverHandlers
    • serverSyncHandlers
    • dataClientHandlers
    • metaClientHandlers
    • AfterWorkingProcessHandler
    • DatumLeaseManager
    • DisconnectEventHandler
  • DataServerNotifyBeanConfiguration
    • DataChangeHandler
    • SessionServerNotifier
    • TempPublisherNotifier
    • BackUpNotifier
    • SnapshotBackUpNotifier
  • DataServerSyncBeanConfiguration
    • SyncDataService
    • LocalAcceptorStore
    • syncDataScheduler
    • StoreServiceFactory
  • DataServerEventBeanConfiguration
    • DataServerChangeEventHandler
    • LocalDataServerChangeEventHandler
    • MetaServerChangeEventHandler
    • StartTaskEventHandler
    • LocalDataServerCleanHandler
    • GetSyncDataHandler
    • EventCenter
    • DataChangeEventCenter
  • DataServerRemotingBeanConfiguration
    • ConnectionRefreshTask
    • ConnectionRefreshMetaTask
    • RenewNodeTask
    • List Tasks, including the above three beans
    • DefaultMetaServiceImpl
  • ResourceConfiguration
    • jerseyResourceConfig
    • HealthResource
    • DataDigestResource
  • ExecutorConfiguration
    • publishProcessorExecutor
    • renewDatumProcessorExecutor
    • getDataProcessorExecutor
  • DataProvideDataConfiguration
    • ProvideDataProcessorManager
    • datumExpireProvideDataProcessor

Partial beans have the following functions:

  • DataServerBootstrapConfigConfiguration: The configuration class mainly provides some basic beans for starting the DataServer service, such as the DataServerConfig basic configuration Bean, DataNodeStatus node status Bean, and DatumCache cache Bean.
  • LogTaskConfigConfiguration: the configuration class is mainly used to provide some log processing related Bean;
  • SessionRemotingConfiguration: the configuration class main role is to provide some beans and SessionServer mutual communication, and some links in the process of request processing Bean. For example, BoltExchange, JerseyExchange and other beans used to start services, as well as the beans used for node up and down, data publishing, etc., are key configuration classes.
  • DataServerNotifyBeanConfiguration: the configuration class configured in Bean is mainly used for event notification, such as used for processing data changes DataChangeHandler;
  • DataServerSyncBeanConfiguration: the configuration class configured in Bean is mainly used for data synchronization;
  • DataServerEventBeanConfiguration: the configuration class configured in Bean is mainly used for processing data node related events, such as center EventCenter, data change events center DataChangeEventCenter etc.
  • DataServerRemotingBeanConfiguration: the configuration class configured in Bean is mainly used for DataServer connection management;
  • ResourceConfiguration: Beans configured in this configuration class provide Rest interface resources;
  • AfterWorkingProcessConfiguration: the configuration class configure some post-processing Handler beans, used for handling some business logic after the post-processing actions;
  • ExecutorConfiguration: This configuration class configates thread pool beans for performing different tasks.

The code for the reduced version is as follows:

@Configuration
@Import(DataServerInitializer.class)
@EnableConfigurationProperties
public class DataServerBeanConfiguration {

    @Bean
    @ConditionalOnMissingBean
    public DataServerBootstrap dataServerBootstrap(a) {}

    @Configuration
    protected static class DataServerBootstrapConfigConfiguration {}

    @Configuration
    public static class DataServerStorageConfiguration {}

    @Configuration
    public static class LogTaskConfigConfiguration {}

    @Configuration
    public static class SessionRemotingConfiguration {}

    @Configuration
    public static class DataServerNotifyBeanConfiguration {}

    @Configuration
    public static class DataServerSyncBeanConfiguration {}

    @Configuration
    public static class DataServerEventBeanConfiguration {}

    @Configuration
    public static class DataServerRemotingBeanConfiguration {}

    @Configuration
    public static class ResourceConfiguration {}

    @Configuration
    public static class ExecutorConfiguration {}

    @Configuration
    public static class DataProvideDataConfiguration {}}Copy the code

0 x06 start

6.1 the entrance

The startup entry class of the DataServer module is DataServerInitializer. This class is not managed and configured by JavaConfig, but inherits the SmartLifecycle interface. The Spring framework calls its start method at startup time. The abbreviated version of the code is as follows:

public class DataServerInitializer implements SmartLifecycle {

    @Autowired
    private DataServerBootstrap dataServerBootstrap;

    @Override
    public void start(a) {
        dataServerBootstrap.start();
        this.isRunning = true; }}Copy the code

The DataServerBootstrap#start method is called to start a series of initialization services.

public void start(a) {
    try {
        openDataServer();
        openDataSyncServer();
        openHttpServer();
        startRaftClient();
        fetchProviderData();
        startScheduler();
        Runtime.getRuntime().addShutdownHook(new Thread(this::doStop)); }}Copy the code

6.2 Starting Services

DataServerBootstrap is responsible for starting the program as follows:

@EnableConfigurationProperties
public class DataServerBootstrap {

    // Bolt communication components between nodes and their configuration
    @Autowired
    private DataServerConfig                  dataServerConfig;
  
    @Resource(name = "serverHandlers")
    private Collection<AbstractServerHandler> serverHandlers;

    @Resource(name = "serverSyncHandlers")
    private Collection<AbstractServerHandler> serverSyncHandlers;  
  
    @Autowired
    private Exchange                          boltExchange;

    private Server                            server;

    private Server                            dataSyncServer;  
  
    // The Http communication component for control and its configuration
    @Autowired
    private ApplicationContext                applicationContext;    
  
    @Autowired
    private ResourceConfig                    jerseyResourceConfig;

    @Autowired
    private Exchange                          jerseyExchange;
 
    private Server                            httpServer; 
  
    // The event communication component within the JVM and its configuration
    @Autowired
    private EventCenter                       eventCenter;
  
    // MetaServer Raft related components
    @Autowired
    private IMetaServerService                metaServerService;
  
    @Autowired
    private DatumLeaseManager                 datumLeaseManager;
  
    // The timer component and its configuration
    @Autowired
    private Scheduler                         syncDataScheduler;

    @Autowired
    private CacheDigestTask                   cacheDigestTask;

    /** * start dataserver */
    public void start(a) {
            openDataServer(); // Bolt communication components between nodes and their configuration
            openDataSyncServer();

            openHttpServer(); // The Http communication component for control and its configuration

            startRaftClient(); // MetaServer Raft related components

            fetchProviderData(); 

            startScheduler(); // The timer component and its configuration

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

    // Bolt communication components between nodes and their configuration
    private void openDataServer(a) {
            if (serverForSessionStarted.compareAndSet(false.true)) {
                server = boltExchange.open(new URL(NetUtil.getLocalAddress().getHostAddress(),
                    dataServerConfig.getPort()), serverHandlers
                    .toArray(newChannelHandler[serverHandlers.size()])); }}private void openDataSyncServer(a) {
            if (serverForDataSyncStarted.compareAndSet(false.true)) {
                dataSyncServer = boltExchange.open(new URL(NetUtil.getLocalAddress()
                    .getHostAddress(), dataServerConfig.getSyncDataPort()), serverSyncHandlers
                    .toArray(newChannelHandler[serverSyncHandlers.size()])); }}// The Http communication component for control and its configuration
    private void openHttpServer(a) {
            if (httpServerStarted.compareAndSet(false.true)) {
                bindResourceConfig();
                httpServer = jerseyExchange.open(
                    new URL(NetUtil.getLocalAddress().getHostAddress(), dataServerConfig
                        .getHttpServerPort()), newResourceConfig[] { jerseyResourceConfig }); }}// MetaServer Raft related components
    private void startRaftClient(a) { 
        metaServerService.startRaftClient();
        eventCenter.post(new MetaServerChangeEvent(metaServerService.getMetaServerMap()));
    }

    private void fetchProviderData(a) {
        ProvideData provideData = metaServerService
            .fetchData(ValueConstants.ENABLE_DATA_DATUM_EXPIRE);
        boolean enableDataDatumExpire = Boolean.parseBoolean((String) provideData.getProvideData()
            .getObject());
        datumLeaseManager.setRenewEnable(enableDataDatumExpire);
    }

    // The timer component and its configuration
    private void startScheduler(a) {
            if (schedulerStarted.compareAndSet(false.true)) {
                syncDataScheduler.startScheduler();
                // start all startTask except correction task
                eventCenter.post(newStartTaskEvent( Arrays.stream(StartTaskTypeEnum.values()).filter(type -> type ! = StartTaskTypeEnum.RENEW) .collect(Collectors.toSet())));//start dump logcacheDigestTask.start(); }}}Copy the code

6.2 Core Components

The core startup class of DataServer is DataServerBootstrap. The core startup class of DataServer is DataServerBootstrap.

This class mainly includes three components: Bolt communication component between nodes, event communication component within JVM, and timer component.

I’m going to be more specific here, dividing the components into the following:

  • Bolt communication components between nodes and their configuration
    • DataServerConfig. configuration
    • BoltExchange. The BOLT component is used to provide communication services for the server and dataSyncServer.
    • Server. DataServer is responsible for data-related services, such as data services, notifications for retrieving data, and service up-and-down notifications.
    • DataSyncServer. DataSyncServer handles data synchronization services.
    • ServerHandlers. Service response function
    • ServerSyncHandlers. DataSyncServer and dataSever have overlapping responsibilities in terms of their registered handlers.
  • The Http communication component for control and its configuration provides a series of REST interfaces for dashboard management and data query.
    • JerseyResourceConfig. configuration
    • JerseyExchange. Jersey component Communication component, providing services;
    • ApplicationContext. Required for registration services;
    • The httpServer provides a series of HTTP interfaces for dashboard management and data query.
  • MetaServer related components
    • MetaServerService, for interacting with MetaServer, based on raft and Bolt;
    • DatumLeaseManager, used to maintain specific data;
  • The event communication components within the JVM and their configuration
    • EventCenter. The internal logic of DataServer is mainly implemented by event-driven mechanism. An event often has multiple delivery sources, so EventCenter is suitable to decouple the logic between event delivery and event processing.
  • The timer component and its configuration
    • SyncDataScheduler: starts expireCheckExecutor, versionCheckExecutor, for example, periodic check node information and periodic check data version information.
    • CacheDigestTask, used for timed analysis;

6.3 Server component

6.3.1 DataServer

DataServer is responsible for data-related services, such as data services, notifications for retrieving data, and online and offline notifications.

DataServer is based on Bolt to communicate.

private void openDataServer(a) {
    try {
        if (serverForSessionStarted.compareAndSet(false.true)) {
            server = boltExchange.open(new URL(NetUtil.getLocalAddress().getHostAddress(),
                dataServerConfig.getPort()), serverHandlers
                .toArray(newChannelHandler[serverHandlers.size()])); }}}Copy the code

The response function is serverHandlers

@Bean(name = "serverHandlers")
public Collection<AbstractServerHandler> serverHandlers(a) {
    Collection<AbstractServerHandler> list = new ArrayList<>();
    list.add(getDataHandler());
    list.add(clientOffHandler());
    list.add(getDataVersionsHandler());
    list.add(publishDataProcessor());
    list.add(sessionServerRegisterHandler());
    list.add(unPublishDataHandler());
    list.add(dataServerConnectionHandler());
    list.add(renewDatumHandler());
    list.add(datumSnapshotHandler());
    return list;
}
Copy the code

Its specific functions are as follows:

  • GetDataHandler: Obtains registration information Data from the current Data node. If the current node is not in working state, change it to the next node.
  • ClientOffHandler: Service subscriber offline;
  • GetDataVersionsHandler: Get the data version number;
  • PublishDataProcessor: Publishing service registration information;
  • SessionServerRegisterHandler: sessionServer session registration;
  • UnPublishDataHandler: Service offline processing;
  • DataServerConnectionHandler: connection management;
  • RenewDatumHandler: Data renewal management;
  • DatumSnapshotHandler: Data snapshot management.

6.3.2 DataSyncServer

DataSyncServer handles data synchronization services. Bolt is also used to communicate.

private void openDataSyncServer(a) {
    try {
        if (serverForDataSyncStarted.compareAndSet(false.true)) {
            dataSyncServer = boltExchange.open(new URL(NetUtil.getLocalAddress()
                .getHostAddress(), dataServerConfig.getSyncDataPort()), serverSyncHandlers
                .toArray(newChannelHandler[serverSyncHandlers.size()])); }}}Copy the code

Its response function is serverSyncHandlers.

@Bean(name = "serverSyncHandlers")
public Collection<AbstractServerHandler> serverSyncHandlers(a) {
    Collection<AbstractServerHandler> list = new ArrayList<>();
    list.add(getDataHandler());
    list.add(publishDataProcessor());
    list.add(unPublishDataHandler());
    list.add(notifyFetchDatumHandler());
    list.add(notifyOnlineHandler());
    list.add(syncDataHandler());
    list.add(dataSyncServerConnectionHandler());
    return list;
}
Copy the code

Its specific functions are as follows:

  • GetDataHandler: Obtains the registration information of the Data node.
  • PublishDataProcessor: Publishing service registration information;
  • UnPublishDataHandler: Service offline processing;
  • NotifyFetchDatumHandler: Compares the version number and captures the latest service registration data;
  • NotifyOnlineHandler: Checks whether the Data node is online.
  • SyncDataHandler: Data synchronization;
  • DataSyncServerConnectionHandler: connection management;

6.3.3 HttpServer

The HttpServer is an Http communication component that provides a series of REST interfaces for dashboard management and data query.

It communicates based on Jersey.

private void openHttpServer(a) {
    try {
        if (httpServerStarted.compareAndSet(false.true)) {
            bindResourceConfig();
            httpServer = jerseyExchange.open(
                new URL(NetUtil.getLocalAddress().getHostAddress(), dataServerConfig
                    .getHttpServerPort()), newResourceConfig[] { jerseyResourceConfig }); }}}Copy the code

6.3.4 Handler

The specific functions of each Handler are shown in Figure 3:

Figure Handler functions

6.3.5 Raft

Raft is related to:

  • Start Raft client to ensure distributed consistency;
  • Drop MetaServerChangeEvent to EventCenter;
private void startRaftClient(a) {
    metaServerService.startRaftClient();
    eventCenter.post(new MetaServerChangeEvent(metaServerService.getMetaServerMap()));
}
Copy the code

6.3.6 Scheduler

This module assists in a variety of regular tasks, specifically:

  • Start a data synchronization task.
    • Regularly detect the change of data recipient nodes and offline expired nodes;
    • Start data change polling;
  • Put messages to the EventCenter for processing by their corresponding response functions, including:
    • CONNECT_META, concrete by ConnectionRefreshMetaTask processing;
    • CONNECT_DATA, which is handled by ConnectionRefreshTask;
    • VERSION_COMPARE, this is not currently handled;
    • It is important to note that RENEW type messages at system startup time no delivery, but in MetaServerChangeEventHandler. RegisterMetaServer, after the registration, would like to see, to RENEW on a regular basis;
  • Start the dump log task.
private void startScheduler(a) {
    try {
        if (schedulerStarted.compareAndSet(false.true)) {
            syncDataScheduler.startScheduler();
            // start all startTask except correction task
            eventCenter.post(newStartTaskEvent( Arrays.stream(StartTaskTypeEnum.values()).filter(type -> type ! = StartTaskTypeEnum.RENEW) .collect(Collectors.toSet())));//start dump logcacheDigestTask.start(); }}}Copy the code
6.3.6.1 startScheduler

VersionCheckExecutor and Scheduler are started, and functions in LocalAcceptorStore are called for periodic checks.

public class Scheduler {

    public final ExecutorService           versionCheckExecutor;
    private final ScheduledExecutorService scheduler;
    private final ThreadPoolExecutor       expireCheckExecutor;

    @Autowired
    private AcceptorStore                  localAcceptorStore;

    /** * constructor */
    public Scheduler(a) {
        scheduler = new ScheduledThreadPoolExecutor(4.new NamedThreadFactory("SyncDataScheduler"));

        expireCheckExecutor = new ThreadPoolExecutor(1.3.0, TimeUnit.SECONDS,
            new SynchronousQueue<>(), new NamedThreadFactory("SyncDataScheduler-expireChangeCheck"));

        versionCheckExecutor = new ThreadPoolExecutor(2.2.0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(), new NamedThreadFactory(
                "SyncDataScheduler-versionChangeCheck"));

    }

    /** * start scheduler */
    public void startScheduler(a) {

        scheduler.schedule(
                new TimedSupervisorTask("FetchDataLocal", scheduler, expireCheckExecutor, 3,
                        TimeUnit.SECONDS, 10, () -> localAcceptorStore.checkAcceptorsChangAndExpired()),
                30, TimeUnit.SECONDS);


        versionCheckExecutor.execute(() -> localAcceptorStore.changeDataCheck());

    }

    /** * stop scheduler */
    public void stopScheduler(a) {
        if(scheduler ! =null && !scheduler.isShutdown()) {
            scheduler.shutdown();
        }
        if(versionCheckExecutor ! =null&&! versionCheckExecutor.isShutdown()) { versionCheckExecutor.shutdown(); }}}Copy the code
6.3.6.2 StartTaskEventHandler

StartTaskEventHandler has an internal ScheduledExecutorService and Tasks. Once a StartTaskEventHandler receives a StartTaskEvent, Tasks are periodically invoked to execute tasks;

@Bean(name = "tasks")
public List<AbstractTask> tasks(a) {
    List<AbstractTask> list = new ArrayList<>();
    list.add(connectionRefreshTask());
    list.add(connectionRefreshMetaTask());
    list.add(renewNodeTask());
    return list;
}
Copy the code

The specific code is as follows:

public class StartTaskEventHandler extends AbstractEventHandler<StartTaskEvent> {

    @Resource(name = "tasks")
    private List<AbstractTask>       tasks;

    private ScheduledExecutorService executor     = null;

    @Override
    public List<Class<? extends StartTaskEvent>> interest() {
        return Lists.newArrayList(StartTaskEvent.class);
    }

    @Override
    public void doHandle(StartTaskEvent event) {
        if (executor == null || executor.isShutdown()) {
            getExecutor();
        }

        for (AbstractTask task : tasks) {
            if(event.getSuitableTypes().contains(task.getStartTaskTypeEnum())) { executor.scheduleWithFixedDelay(task, task.getInitialDelay(), task.getDelay(), task.getTimeUnit()); }}}private void getExecutor(a) {
        executor = ExecutorFactory.newScheduledThreadPool(tasks.size(), this.getClass() .getSimpleName()); }}Copy the code

6.4 processing Task

StartTaskEventHandler starts tasks declared in the Tasks Bean.

However, the specific task to start depends on the Settings in the event. The loop in the following code is to see how tasks and events match.

        for (AbstractTask task : tasks) {
            if(event.getSuitableTypes().contains(task.getStartTaskTypeEnum())) { executor.scheduleWithFixedDelay(task, task.getInitialDelay(), task.getDelay(),task.getTimeUnit()); }}Copy the code

The specific code is as follows:

public class StartTaskEventHandler extends AbstractEventHandler<StartTaskEvent> {

    @Resource(name = "tasks")
    private List<AbstractTask>       tasks;

    private ScheduledExecutorService executor     = null;

    @Override
    public List<Class<? extends StartTaskEvent>> interest() {
        return Lists.newArrayList(StartTaskEvent.class);
    }

    @Override
    public void doHandle(StartTaskEvent event) {
        if (executor == null || executor.isShutdown()) {
            getExecutor();
        }

        for (AbstractTask task : tasks) {
            if(event.getSuitableTypes().contains(task.getStartTaskTypeEnum())) { executor.scheduleWithFixedDelay(task, task.getInitialDelay(), task.getDelay(), task.getTimeUnit()); }}}private void getExecutor(a) {
        executor = ExecutorFactory.newScheduledThreadPool(tasks.size(), this.getClass() .getSimpleName()); }}Copy the code

6.4.1 beans

Corresponding beans, a total of three tasks.

@Bean(name = "tasks")
public List<AbstractTask> tasks(a) {
    List<AbstractTask> list = new ArrayList<>();
    list.add(connectionRefreshTask());
    list.add(connectionRefreshMetaTask());
    list.add(renewNodeTask());
    return list;
}
Copy the code

Corresponds to an enumeration in StartTaskTypeEnum, where VersionCompareTask is not implemented.

public enum StartTaskTypeEnum {

    /** * ConnectionRefreshMetaTask */
    CONNECT_META,

    /** * ConnectionRefreshDataTask */
    CONNECT_DATA,

    /** * RenewNodeTask */
    RENEW,

    /** * VersionCompareTask */
    VERSION_COMPARE
}
Copy the code

6.4.2 decoupling

Let’s use StartTaskEvent as an example, where Set is used to specify what task processing this message applies to.

public class StartTaskEvent implements Event {
    private final Set<StartTaskTypeEnum> suitableTypes;

    public StartTaskEvent(Set<StartTaskTypeEnum> suitableTypes) {
        this.suitableTypes = suitableTypes;
    }

    public Set<StartTaskTypeEnum> getSuitableTypes(a) {
        returnsuitableTypes; }}Copy the code

Among MetaServerChangeEventHandler, launched the renew task.

if (obj instanceof NodeChangeResult) {
    NodeChangeResult<DataNode> result = (NodeChangeResult<DataNode>) obj;
    Map<String, Long> versionMap = result.getDataCenterListVersions();

    //send renew after first register dataNode
    Set<StartTaskTypeEnum> set = new HashSet<>();
    set.add(StartTaskTypeEnum.RENEW);
    eventCenter.post(new StartTaskEvent(set));

    eventCenter.post(new DataServerChangeEvent(result.getNodes(), versionMap,
            DataServerChangeEvent.FromType.REGISTER_META));
    break;
}
Copy the code

At startup time, an event is posted, but a non-Renew task is specified to start.

private void startScheduler(a) {
    try {
        if (schedulerStarted.compareAndSet(false.true)) {
            syncDataScheduler.startScheduler();
            // start all startTask except correction task
            eventCenter.post(newStartTaskEvent( Arrays.stream(StartTaskTypeEnum.values()).filter(type -> type ! = StartTaskTypeEnum.RENEW) .collect(Collectors.toSet())));//start dump logcacheDigestTask.start(); }}catch (Exception e) {
        schedulerStarted.set(false);
        throw new RuntimeException("Data Scheduler start error!", e); }}Copy the code

0x07 Dynamic Structure

Finally, the dynamic architecture is as follows, we also know roughly, DataServer is a SpringBoot program, has several servers, a number of beans, a number of timing services, specific some other business modules, etc., which is helpful for our understanding of the following.

+---------------------------------------------------------------------------+ | [DataServerBootstrap] | | | | | | +------------------------------------------+ +------------------------+ | | | [Bolt related] | | [http relatged] | | | |  | | | | | | DataServerConfig | | httpServer | | | | | | | | | | boltExchange | | jerseyExchange | | | | | | | | | | server +-----------> serverHandlers | | applicationContext | | | | | | | | | | dataSyncServer+----> serverSyncHandlers |  | jerseyResourceConfig | | | | | | | | | +------------------------------------------+ +------------------------+ | | +---------------------+ +----------------+ +------------------------+ | | |[meta related] | |[JVM related] | |[Timer related] | | | | | | | | | | | | metaServerService | | | | syncDataScheduler | | | | | | EventCenter | | | | | | datumLeaseManager | | | | CacheDigestTask | | | +---------------------+ +----------------+ | | | | +------------------------+ | +---------------------------------------------------------------------------+Copy the code

0x08 Problem list

Since it is more helpful to start with problems, we have compiled a list of problems that we expect to address in the future analysis.

  • Question: What exactly is an Datacenter?
  • Question: What kind of system should DataServer be?
  • Question: What should DataServer do? How to do that?
  • Question: How do you maintain high availability?
  • Question: How to load balance?
  • Question: How do dataservers synchronize? Data consistency?
  • Question: How does SessionServer address DataServer?
  • Question: how does the client know which SessionServer to contact?
  • Question: how does the SessionServer express itself within DataServer? Is there a cache?
  • Question: What does a hash routing table look like?
  • Question: How does DataServer push information to all session Servers?
  • Question: How does DataServer synchronize to other dataservers?
  • Problem: dataSyncServer handles data synchronization services; DataServer is responsible for data-related services; What’s the difference?
  • Q: How many events are there in EventCenter?
  • Question: How do I poll MetaServer?
  • Question: How do I determine the current equipment room node?
  • Question: How does data migrate within a DataServer cluster?
  • Question: Is the communication between SessionServer and DataServer based on push-pull mechanism?
  • Question: why is there a startRaftClient in DataServerBootstrap?
  • Question: how is MetaServerChangeEventHandler start, who will control, used to do?
  • Question: What does the DatumLeaseManager do?
  • Question: What does SessionServer pull from DataServer?
  • Question: How does DataServer renew itself to MetaServer? Is it regular?
  • Question: How does DataServer know to save other dataservers? Is it used elsewhere?
  • Question: What does DataServer need to save to consider?
  • Question: What are version numbers used for?
  • Question: What is DatumCache used for?
  • Question: Why AfterWorkingProcess?
  • Q: How does Bolt maintain connection?

0 XFF reference

How does ant Financial Service registry realize the smooth scaling of DataServer

Ant gold uniform service registry SOFARegistry parsing | service discovery path optimization

The service registry Session storage policy | SOFARegistry parsing

Introduction to the Registry – SOFARegistry architecture for Massive Data

Service registry data fragmentation and synchronization scheme, rounding | SOFARegistry parsing

Ant Financial open source communication framework SOFABolt analysis of connection management analysis

Timeout control mechanism and heartbeat mechanism resolved by SOFABolt, ant Financial’s open source communication framework

SOFABolt protocol framework analysis of Ant Financial open source communication framework

Ant gold uniform service registry data consistency analysis | SOFARegistry parsing

Ant communication framework practice

Sofa – Bolt remote call

Sofa – bolt to study

SOFABolt Design Summary – Elegant and simple design approach

SofaBolt source code analysis – Service startup to message processing

SOFABolt source code analysis

SOFABolt source code analysis 9-userProcessor custom processor design

SOFARegistry introduction

SOFABolt source code analysis of the design of the 13-Connection event processing mechanism