Talk about NacOS’s configWatchers

sequence

This paper mainly studies configWatchers of NACOS

CommunicationController

Nacos – 1.1.3 / config/SRC/main/Java/com/alibaba/nacos/config/server/controller/CommunicationController. Java

@Controller
@RequestMapping(Constants.COMMUNICATION_CONTROLLER_PATH)
public class CommunicationController {

    private final DumpService dumpService;

    private final LongPollingService longPollingService;

    private String trueStr = "true"; @Autowired public CommunicationController(DumpService dumpService, LongPollingService longPollingService) { this.dumpService = dumpService; this.longPollingService = longPollingService; } / /... /** * @requestMapping (value ="/configWatchers", method = RequestMethod.GET)
    @ResponseBody
    public SampleResult getSubClientConfig(HttpServletRequest request,
                                           HttpServletResponse response,
                                           @RequestParam("dataId") String dataId,
                                           @RequestParam("group") String group,
                                           @RequestParam(value = "tenant", required = false) String tenant,
                                           ModelMap modelMap) {
        group = StringUtils.isBlank(group) ? Constants.DEFAULT_GROUP : group;
        returnlongPollingService.getCollectSubscribleInfo(dataId, group, tenant); } / /... }Copy the code

• CommunicationController provides/configWatchersInterface, which through longPollingService. SampleResult getCollectSubscribleInfo returned

LongPollingService

Nacos – 1.1.3 / config/SRC/main/Java/com/alibaba/nacos/config/server/service/LongPollingService Java

@Service
public class LongPollingService extends AbstractEventListener {

    private static final int FIXED_POLLING_INTERVAL_MS = 10000;

    private static final int SAMPLE_PERIOD = 100;

    private static final int SAMPLE_TIMES = 3;

    private static final String TRUE_STR = "true"; private Map<String, Long> retainIps = new ConcurrentHashMap<String, Long>(); / /... public SampleResult getCollectSubscribleInfo(String dataId, String group, String tenant) { List<SampleResult> sampleResultLst = new ArrayList<SampleResult>(50);for (int i = 0; i < SAMPLE_TIMES; i++) {
            SampleResult sampleTmp = getSubscribleInfo(dataId, group, tenant);
            if(sampleTmp ! = null) { sampleResultLst.add(sampleTmp); }if (i < SAMPLE_TIMES - 1) {
                try {
                    Thread.sleep(SAMPLE_PERIOD);
                } catch (InterruptedException e) {
                    LogUtil.clientLog.error("sleep wrong", e);
                }
            }
        }

        SampleResult sampleResult = mergeSampleResult(sampleResultLst);
        return sampleResult;
    }

    public SampleResult getSubscribleInfo(String dataId, String group, String tenant) {
        String groupKey = GroupKey.getKeyTenant(dataId, group, tenant);
        SampleResult sampleResult = new SampleResult();
        Map<String, String> lisentersGroupkeyStatus = new HashMap<String, String>(50);

        for (ClientLongPolling clientLongPolling : allSubs) {
            if (clientLongPolling.clientMd5Map.containsKey(groupKey)) {
                lisentersGroupkeyStatus.put(clientLongPolling.ip, clientLongPolling.clientMd5Map.get(groupKey));
            }
        }
        sampleResult.setLisentersGroupkeyStatus(lisentersGroupkeyStatus);
        returnsampleResult; } /** * Aggregate the sampling IP and listening configuration information in the sampling result; Param sampleResults sampleResults * @param sampleResults sampleResults * @param sampleResults sampleResults * @return Results
     */
    public SampleResult mergeSampleResult(List<SampleResult> sampleResults) {
        SampleResult mergeResult = new SampleResult();
        Map<String, String> lisentersGroupkeyStatus = new HashMap<String, String>(50);
        for (SampleResult sampleResult : sampleResults) {
            Map<String, String> lisentersGroupkeyStatusTmp = sampleResult.getLisentersGroupkeyStatus();
            for (Map.Entry<String, String> entry : lisentersGroupkeyStatusTmp.entrySet()) {
                lisentersGroupkeyStatus.put(entry.getKey(), entry.getValue());
            }
        }
        mergeResult.setLisentersGroupkeyStatus(lisentersGroupkeyStatus);
        returnmergeResult; } / /... }Copy the code

• The LongPollingService’s getCollectSubscribleInfo method loops SAMPLE_TIMES(The default value is 3), each time SampleResult is fetched and added to sampleResultLst through the getSubscribleInfo method, and the cycle time interval SAMPLE_PERIOD(The default value is 100) ms; Finally, the mergeSampleResult method is used to merge, and SampleResult is returned
• The getSubscribleInfo method iterates through each clientLongPolling in allSubs, Collect clientLongPolling. ClientMd5Map. Either containsKey (groupKey) clientLongPolling IP and md5 to lisentersGroupkeyStatus, Finally, assign to the sampleResult
• The mergeSampleResult method merges the lisentersGroupkeyStatus of each SampleResult and creates a new SampleResult return

summary

CommunicationController provides/configWatchers interface, it through longPollingService. SampleResult getCollectSubscribleInfo returned

doc

• CommunicationController