@Override
    public void subscribe(String cluster, ConsulListener listener) throws Exception {
        //1.add listener to subscribe list
        listenerMap.computeIfAbsent(cluster, key -> new HashSet<>())
                .add(listener);
        //2.get healthy services
        Response<List<HealthService>> response = getHealthyServices(cluster, -1, DEFAULT_WATCH_TIMEOUT);
        //3.get current consul index.
        Long index = response.getConsulIndex();
        ConsulNotifier notifier = notifiers.computeIfAbsent(cluster, key -> new ConsulNotifier(cluster, index));
        //4.run notifier
        notifierExecutor.submit(notifier);
    }
@Test
    public void testSubscribe() throws Exception {
        RegistryService registryService = mock(ConsulRegistryServiceImpl.class);
        ConsulListener consulListener = mock(ConsulListener.class);
        registryService.subscribe("test", consulListener);
        verify(registryService).subscribe("test", consulListener);
    }