分布式注册中心xxl-registry源码分析

野性酷女 2024-04-17 23:38 138阅读 0赞

官网概述：XXL-REGISTRY 是一个轻量级分布式服务注册中心，拥有”轻量级、秒级注册上线、多环境、跨语言、跨机房”等特性。现已开放源代码，开箱即用。

基本思路：注册中心的基本的功能一般包括服务注册(registry)、服务发现(discovery)、服务摘除(remove)等；然后就涉及到服务宕机后能否及时摘除、如何高效的服务发现、注册中心数据备份等；这些任务由注册中心的服务端完成(server)，另外注册中心还会提供一个客户端(client)，服务提供者通过client端进行服务注册、摘除、心跳检测等，服务消费者通过client端进行服务发现；

项目结构：

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xkdzIwMTUxMDgwMzAwNg_size_16_color_FFFFFF_t_70

设计的三张表:

20190903075456260.png

a、xxlRegistry实体：

  1. public class XxlRegistry {
  3.     private int id;
  4.     private String biz;         // 业务标识
  5.     private String env;         // 环境标识
  6.     private String key;         // 注册Key
  7.     private String data;        // 注册Value有效数据
  8.     private int status;         // 状态：0-正常、1-锁定、2-禁用

其中data为服务的注册地址集合以String类型在数据库中存储如：[“address001”, “address002”]

b、xxlRegistryData实体：

  1. public class XxlRegistryData {
  3.     private int id;
  4.     private String biz;         // 业务标识
  5.     private String env;         // 环境标识
  6.     private String key;         // 注册Key
  7.     private String value;       // 注册Value
  8.     private Date updateTime;    // 更新时间

xxlRegistryData与xxlRegistry相当于是多对一的关系, 存数据的时候将biz+env+key对应的value取出来存到xxlRegistry的data中。

c、xxlRegistryMessage实体：

  1. public class XxlRegistryMessage {
  3.     private int id;
  4.     private int type;         // 消息类型：0-注册更新
  5.     private String data;      // 消息内容
  6.     private Date addTime;

这里data存的是xxlRegistry信息

一、注册中心Server端

1、API：包括Admin界面管理api、Client端访问api

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xkdzIwMTUxMDgwMzAwNg_size_16_color_FFFFFF_t_70 1

其中 Api的service层实现以及服务定时处理的各类线程都在XxlRegistryServiceImpl中完成：

  1. @Service
  2. public class XxlRegistryServiceImpl implements IXxlRegistryService, InitializingBean, DisposableBean {
  3.     private static Logger logger = LoggerFactory.getLogger(XxlRegistryServiceImpl.class);
  6.     @Resource
  7.     private IXxlRegistryDao xxlRegistryDao;
  8.     @Resource
  9.     private IXxlRegistryDataDao xxlRegistryDataDao;
  10.     @Resource
  11.     private IXxlRegistryMessageDao xxlRegistryMessageDao;
  13.     @Value("${xxl.registry.data.filepath}")
  14.     private String registryDataFilePath;
  15.     @Value("${xxl.registry.accessToken}")
  16.     private String accessToken;
  18.     private int registryBeatTime = 10;

1.1、RegistryController:

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xkdzIwMTUxMDgwMzAwNg_size_16_color_FFFFFF_t_70 2

对注册中心的基本信息进行人工维护, 可以修改状态信息, 禁止自动服务注册与发现等；这里是直接针对数据库层面的修改。

1.2、ApiController:

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xkdzIwMTUxMDgwMzAwNg_size_16_color_FFFFFF_t_70 3

提供服务注册、发现、摘除、心跳维护等接口

a、registry实现：

  1. @Override
  2.     public ReturnT<String> registry(String accessToken, String biz, String env, List<XxlRegistryData> registryDataList) {
  4.         // valid
  5.         if (this.accessToken!=null && this.accessToken.trim().length()>0 && !this.accessToken.equals(accessToken)) {
  6.             return new ReturnT<String>(ReturnT.FAIL_CODE, "AccessToken Invalid");
  7.         }
  8.         if (biz==null || biz.trim().length()<4 || biz.trim().length()>255) {
  9.             return new ReturnT<String>(ReturnT.FAIL_CODE, "Biz Invalid[4~255]");
  10.         }
  11.         if (env==null || env.trim().length()<2 || env.trim().length()>255) {
  12.             return new ReturnT<String>(ReturnT.FAIL_CODE, "Env Invalid[2~255]");
  13.         }
  14.         if (registryDataList==null || registryDataList.size()==0) {
  15.             return new ReturnT<String>(ReturnT.FAIL_CODE, "Registry DataList Invalid");
  16.         }
  17.         for (XxlRegistryData registryData: registryDataList) {
  18.             if (registryData.getKey()==null || registryData.getKey().trim().length()<4 || registryData.getKey().trim().length()>255) {
  19.                 return new ReturnT<String>(ReturnT.FAIL_CODE, "Registry Key Invalid[4~255]");
  20.             }
  21.             if (registryData.getValue()==null || registryData.getValue().trim().length()<4 || registryData.getValue().trim().length()>255) {
  22.                 return new ReturnT<String>(ReturnT.FAIL_CODE, "Registry Value Invalid[4~255]");
  23.             }
  24.         }
  26.         // fill + add queue
  27.         for (XxlRegistryData registryData: registryDataList) {
  28.             registryData.setBiz(biz);
  29.             registryData.setEnv(env);
  30.         }
  31.         registryQueue.addAll(registryDataList);
  33.         return ReturnT.SUCCESS;
  34.     }

可以看到前面都在对参赛进行校验，后面是直接将数据存如registryQueue队列中，持久到数据库是通过开启线程同步到数据库中：

  1. /**
  2.          * registry registry data         (client-num/10 s)
  3.          */
  4.         for (int i = 0; i < 10; i++) {
  5.             executorService.execute(new Runnable() {
  6.                 @Override
  7.                 public void run() {
  8.                     while (!executorStoped) {
  9.                         try {
  10.                             XxlRegistryData xxlRegistryData = registryQueue.take();
  11.                             if (xxlRegistryData !=null) {
  13.                                 // refresh or add
  14.                                 int ret = xxlRegistryDataDao.refresh(xxlRegistryData);
  15.                                 if (ret == 0) {
  16.                                     xxlRegistryDataDao.add(xxlRegistryData);
  17.                                 }
  19.                                 // valid file status
  20.                                 XxlRegistry fileXxlRegistry = getFileRegistryData(xxlRegistryData);
  21.                                 if (fileXxlRegistry == null) {
  22.                                     // go on
  23.                                 } else if (fileXxlRegistry.getStatus() != 0) {
  24.                                     continue;     // "Status limited."
  25.                                 } else {
  26.                                     if (fileXxlRegistry.getDataList().contains(xxlRegistryData.getValue())) {
  27.                                         continue;     // "Repeated limited."
  28.                                     }
  29.                                 }
  31.                                 // checkRegistryDataAndSendMessage
  32.                                 checkRegistryDataAndSendMessage(xxlRegistryData);
  33.                             }
  34.                         } catch (Exception e) {
  35.                             if (!executorStoped) {
  36.                                 logger.error(e.getMessage(), e);
  37.                             }
  38.                         }
  39.                     }
  40.                 }
  41.             });
  42.         }

这里就是从注册服务的队列取出数据并同步更新到xxl-registry-data表中，并将信息同步到xxl-registry和xxl-registry-message表中；

服务摘除与注册方式类似，通过接口存到队列，再另跑线程同步到表中(xxl-registry-data)；服务摘除时会同步更新xxl-regsitry和xxl-registry-message表，另跑线程将xxl-registry-message的最新信息同步到数据文件中，以保证服务的最新可用性。

b、服务发现：其高效性就体现在从文件中直接读取数据

  1. @Override
  2.     public ReturnT<Map<String, List<String>>> discovery(String accessToken, String biz, String env, List<String> keys) {
  4.         // valid
  5.         if (this.accessToken!=null && this.accessToken.trim().length()>0 && !this.accessToken.equals(accessToken)) {
  6.             return new ReturnT<>(ReturnT.FAIL_CODE, "AccessToken Invalid");
  7.         }
  8.         if (biz==null || biz.trim().length()<2 || biz.trim().length()>255) {
  9.             return new ReturnT<>(ReturnT.FAIL_CODE, "Biz Invalid[2~255]");
  10.         }
  11.         if (env==null || env.trim().length()<2 || env.trim().length()>255) {
  12.             return new ReturnT<>(ReturnT.FAIL_CODE, "Env Invalid[2~255]");
  13.         }
  14.         if (keys==null || keys.size()==0) {
  15.             return new ReturnT<>(ReturnT.FAIL_CODE, "keys Invalid.");
  16.         }
  17.         for (String key: keys) {
  18.             if (key==null || key.trim().length()<4 || key.trim().length()>255) {
  19.                 return new ReturnT<>(ReturnT.FAIL_CODE, "Key Invalid[4~255]");
  20.             }
  21.         }
  23.         Map<String, List<String>> result = new HashMap<String, List<String>>();
  24.         for (String key: keys) {
  25.             XxlRegistryData xxlRegistryData = new XxlRegistryData();
  26.             xxlRegistryData.setBiz(biz);
  27.             xxlRegistryData.setEnv(env);
  28.             xxlRegistryData.setKey(key);
  30.             List<String> dataList = new ArrayList<String>();
  31.             XxlRegistry fileXxlRegistry = getFileRegistryData(xxlRegistryData);
  32.             if (fileXxlRegistry!=null) {
  33.                 dataList = fileXxlRegistry.getDataList();
  34.             }
  36.             result.put(key, dataList);
  37.         }
  39.         return new ReturnT<Map<String, List<String>>>(result);
  40.     }

文件中的注册信息通过xxl-registry-message表中同步而来，并定时清除表中

  1. /**
  2.          * broadcase new one registry-data-file     (1/1s)
  3.          *
  4.          * clean old message   (1/10s)
  5.          */
  6.         executorService.execute(new Runnable() {
  7.             @Override
  8.             public void run() {
  9.                 while (!executorStoped) {
  10.                     try {
  11.                         // new message, filter readed
  12.                         List<XxlRegistryMessage> messageList = xxlRegistryMessageDao.findMessage(readedMessageIds);
  13.                         if (messageList!=null && messageList.size()>0) {
  14.                             for (XxlRegistryMessage message: messageList) {
  15.                                 readedMessageIds.add(message.getId());
  17.                                 if (message.getType() == 0) {   // from registry、add、update、deelete，ne need sync from db, only write
  19.                                     XxlRegistry xxlRegistry = JacksonUtil.readValue(message.getData(), XxlRegistry.class);
  21.                                     // process data by status
  22.                                     if (xxlRegistry.getStatus() == 1) {
  23.                                         // locked, not updated
  24.                                     } else if (xxlRegistry.getStatus() == 2) {
  25.                                         // disabled, write empty
  26.                                         xxlRegistry.setData(JacksonUtil.writeValueAsString(new ArrayList<String>()));
  27.                                     } else {
  28.                                         // default, sync from db （aready sync before message, only write）
  29.                                     }
  31.                                     // sync fileI(将数据同步到文件中)
  32.                                     setFileRegistryData(xxlRegistry);
  33.                                 }
  34.                             }
  35.                         }
  37.                         // clean old message;
  38.                         if ( (System.currentTimeMillis()/1000) % registryBeatTime ==0) {
  39.                             xxlRegistryMessageDao.cleanMessage(registryBeatTime);
  40.                             readedMessageIds.clear();
  41.                         }
  42.                     } catch (Exception e) {
  43.                         if (!executorStoped) {
  44.                             logger.error(e.getMessage(), e);
  45.                         }
  46.                     }
  47.                     try {
  48.                         TimeUnit.SECONDS.sleep(1);
  49.                     } catch (Exception e) {
  50.                         if (!executorStoped) {
  51.                             logger.error(e.getMessage(), e);
  52.                         }
  53.                     }
  54.                 }
  55.             }
  56.         });

c、定时清理xxl-registry-data数据表、同步xxl-regsitry-data到xxl-regsitry以及数据文件、清理无效数据文件

  1. /**
  2.          *  clean old registry-data     (1/10s)
  3.          *
  4.          *  sync total registry-data db + file      (1+N/10s)
  5.          *
  6.          *  clean old registry-data file
  7.          */
  8.         executorService.execute(new Runnable() {
  9.             @Override
  10.             public void run() {
  11.                 while (!executorStoped) {
  13.                     // align to beattime
  14.                     try {
  15.                         long sleepSecond = registryBeatTime - (System.currentTimeMillis()/1000)%registryBeatTime;
  16.                         if (sleepSecond>0 && sleepSecond<registryBeatTime) {
  17.                             TimeUnit.SECONDS.sleep(sleepSecond);
  18.                         }
  19.                     } catch (Exception e) {
  20.                         if (!executorStoped) {
  21.                             logger.error(e.getMessage(), e);
  22.                         }
  23.                     }
  25.                     try {
  26.                         // clean old registry-data in db
  27.                         xxlRegistryDataDao.cleanData(registryBeatTime * 3);
  29.                         // sync registry-data, db + file
  30.                         int offset = 0;
  31.                         int pagesize = 1000;
  32.                         List<String> registryDataFileList = new ArrayList<>();
  34.                         List<XxlRegistry> registryList = xxlRegistryDao.pageList(offset, pagesize, null, null, null);
  35.                         while (registryList!=null && registryList.size()>0) {
  37.                             for (XxlRegistry registryItem: registryList) {
  39.                                 // process data by status
  40.                                 if (registryItem.getStatus() == 1) {
  41.                                     // locked, not updated
  42.                                 } else if (registryItem.getStatus() == 2) {
  43.                                     // disabled, write empty
  44.                                     String dataJson = JacksonUtil.writeValueAsString(new ArrayList<String>());
  45.                                     registryItem.setData(dataJson);
  46.                                 } else {
  47.                                     // default, sync from db
  48.                                     List<XxlRegistryData> xxlRegistryDataList = xxlRegistryDataDao.findData(registryItem.getBiz(), registryItem.getEnv(), registryItem.getKey());
  49.                                     List<String> valueList = new ArrayList<String>();
  50.                                     if (xxlRegistryDataList!=null && xxlRegistryDataList.size()>0) {
  51.                                         for (XxlRegistryData dataItem: xxlRegistryDataList) {
  52.                                             valueList.add(dataItem.getValue());
  53.                                         }
  54.                                     }
  55.                                     String dataJson = JacksonUtil.writeValueAsString(valueList);
  57.                                     // check update, sync db
  58.                                     if (!registryItem.getData().equals(dataJson)) {
  59.                                         registryItem.setData(dataJson);
  60.                                         xxlRegistryDao.update(registryItem);
  61.                                     }
  62.                                 }
  64.                                 // sync file(将最新数据同步到数据文件)
  65.                                 String registryDataFile = setFileRegistryData(registryItem);
  67.                                 // collect registryDataFile
  68.                                 registryDataFileList.add(registryDataFile);
  69.                             }
  72.                             offset += 1000;
  73.                             registryList = xxlRegistryDao.pageList(offset, pagesize, null, null, null);
  74.                         }
  76.                         // clean old registry-data file(清除不在registryDataFileList中的无效数据文件)
  77.                         cleanFileRegistryData(registryDataFileList);
  79.                     } catch (Exception e) {
  80.                         if (!executorStoped) {
  81.                             logger.error(e.getMessage(), e);
  82.                         }
  83.                     }
  84.                     try {
  85.                         TimeUnit.SECONDS.sleep(registryBeatTime);
  86.                     } catch (Exception e) {
  87.                         if (!executorStoped) {
  88.                             logger.error(e.getMessage(), e);
  89.                         }
  90.                     }
  91.                 }
  92.             }
  93.         });

d、心跳检测(monitor)：

  1. @Override
  2.     public DeferredResult<ReturnT<String>> monitor(String accessToken, String biz, String env, List<String> keys) {
  4.         // init
  5.         DeferredResult deferredResult = new DeferredResult(30 * 1000L, new ReturnT<>(ReturnT.SUCCESS_CODE, "Monitor timeout, no key updated."));
  7.         // valid
  8.         if (this.accessToken!=null && this.accessToken.trim().length()>0 && !this.accessToken.equals(accessToken)) {
  9.             deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "AccessToken Invalid"));
  10.             return deferredResult;
  11.         }
  12.         if (biz==null || biz.trim().length()<4 || biz.trim().length()>255) {
  13.             deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "Biz Invalid[4~255]"));
  14.             return deferredResult;
  15.         }
  16.         if (env==null || env.trim().length()<2 || env.trim().length()>255) {
  17.             deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "Env Invalid[2~255]"));
  18.             return deferredResult;
  19.         }
  20.         if (keys==null || keys.size()==0) {
  21.             deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "keys Invalid."));
  22.             return deferredResult;
  23.         }
  24.         for (String key: keys) {
  25.             if (key==null || key.trim().length()<4 || key.trim().length()>255) {
  26.                 deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "Key Invalid[4~255]"));
  27.                 return deferredResult;
  28.             }
  29.         }
  31.         // monitor by client
  32.         for (String key: keys) {
  33.             String fileName = parseRegistryDataFileName(biz, env, key);
  35.             List<DeferredResult> deferredResultList = registryDeferredResultMap.get(fileName);
  36.             if (deferredResultList == null) {
  37.                 deferredResultList = new ArrayList<>();
  38.                 registryDeferredResultMap.put(fileName, deferredResultList);
  39.             }
  41.             deferredResultList.add(deferredResult);
  42.         }
  44.         return deferredResult;
  45.     }

通过Spring提供的DeferredResult类实现自动延迟返回信息给客户端，在更新文件数据时修改deferredResultList返回信息，因此文件数据是否最新都会反应在这个结果集中。

以上就是注册中心服务端的基本实现

一般问题：如果服务提供者意外宕机如何处理：一般情况下对应用程序会设置服务停止时的一些正常的清理工作，比如调用remove接口摘除宕机的服务等；另一方便xxl-regsitry-data数据三倍于心跳时间清除，因此也可保证异常服务的正常清除。

二、注册中心Client端

Client端维护着两个重要的线程，服务注册、服务发现

1、服务注册：

  1. // registry thread
  2.         registryThread = new Thread(new Runnable() {
  3.             @Override
  4.             public void run() {
  5.                 while (!registryThreadStop) {
  6.                     try {
  7.                         if (registryData.size() > 0) {
  9.                             boolean ret = registryBaseClient.registry(new ArrayList<XxlRegistryDataParamVO>(registryData));
  10.                             logger.debug(">>>>>>>>>>> xxl-registry, refresh registry data {}, registryData = {}", ret?"success":"fail",registryData);
  11.                         }
  12.                     } catch (Exception e) {
  13.                         if (!registryThreadStop) {
  14.                             logger.error(">>>>>>>>>>> xxl-registry, registryThread error.", e);
  15.                         }
  16.                     }
  17.                     try {
  18.                         TimeUnit.SECONDS.sleep(10);
  19.                     } catch (Exception e) {
  20.                         if (!registryThreadStop) {
  21.                             logger.error(">>>>>>>>>>> xxl-registry, registryThread error.", e);
  22.                         }
  23.                     }
  24.                 }
  25.                 logger.info(">>>>>>>>>>> xxl-registry, registryThread stoped.");
  26.             }
  27.         });
  28.         registryThread.setName("xxl-registry, XxlRegistryClient registryThread.");
  29.         registryThread.setDaemon(true);
  30.         registryThread.start();

定期的将注册信息同步到注册中心，通过服务摘除自动从注册中心下线等操作，供服务提供方使用

2、服务发现：

  1. // discovery thread
  2.         discoveryThread = new Thread(new Runnable() {
  3.             @Override
  4.             public void run() {
  5.                 while (!registryThreadStop) {
  7.                     if (discoveryData.size() == 0) {
  8.                         try {
  9.                             TimeUnit.SECONDS.sleep(3);
  10.                         } catch (Exception e) {
  11.                             if (!registryThreadStop) {
  12.                                 logger.error(">>>>>>>>>>> xxl-registry, discoveryThread error.", e);
  13.                             }
  14.                         }
  15.                     } else {
  16.                         try {
  17.                             // monitor
  18.                             boolean monitorRet = registryBaseClient.monitor(discoveryData.keySet());
  20.                             // avoid fail-retry request too quick
  21.                             if (!monitorRet){
  22.                                 TimeUnit.SECONDS.sleep(10);
  23.                             }
  25.                             // refreshDiscoveryData, all
  26.                             refreshDiscoveryData(discoveryData.keySet());
  27.                         } catch (Exception e) {
  28.                             if (!registryThreadStop) {
  29.                                 logger.error(">>>>>>>>>>> xxl-registry, discoveryThread error.", e);
  30.                             }
  31.                         }
  32.                     }
  34.                 }
  35.                 logger.info(">>>>>>>>>>> xxl-registry, discoveryThread stoped.");
  36.             }
  37.         });
  38.         discoveryThread.setName("xxl-registry, XxlRegistryClient discoveryThread.");
  39.         discoveryThread.setDaemon(true);
  40.         discoveryThread.start();

通过心跳检测发现服务是否可用，避免请求过快造成的问题

一般问题：注册中心意外宕机对服务提供者和服务消费者的影响，因为客户端存在缓存，注册中心宕机后一般可以正常请求服务；但此过程如出现服务宕机，就会出现一些问题，所以一般需要保证尽快恢复注册中心或者注册中心的高可用

发表评论

表情：
评论列表 （有 0 条评论，138人围观）

还没有评论，来说两句吧...

相关阅读