MulticastRegistry extends Registry { @Override public void init() { if (multicastSocket != null) { return; } String addressInput = registryConfig.getAddress(); if (StringUtils.isEmpty(addressInput)) { throw new SofaRpcRuntimeException(LogCodes.getLog(LogCodes.ERROR_EMPTY_ADDRESS, EXT_NAME)); } try { if (!addressInput.startsWith(EXT_NAME)) { addressInput = EXT_NAME + ": } URI url = new URI(addressInput); multicastPort = url.getPort(); if (multicastPort <= 0) { multicastPort = DEFAULT_MULTICAST_PORT; } multicastAddress = InetAddress.getByName(url.getHost()); MulticastRegistryHelper.checkMulticastAddress(multicastAddress); multicastSocket = new MulticastSocket(multicastPort); NetUtils.joinMulticastGroup(multicastSocket, multicastAddress); Thread thread = new Thread(new Runnable() { @Override public void run() { byte[] buf = new byte[2048]; DatagramPacket recv = new DatagramPacket(buf, buf.length); while (!multicastSocket.isClosed()) { try { multicastSocket.receive(recv); String msg = new String(recv.getData()).trim(); int i = msg.indexOf('\n'); if (i > 0) { msg = msg.substring(0, i).trim(); } MulticastRegistry.this.receive(msg, (InetSocketAddress) recv.getSocketAddress()); Arrays.fill(buf, (byte) 0); } catch (Throwable e) { if (!multicastSocket.isClosed()) { LOGGER.error(e.getMessage(), e); } } } } }, "SofaMulticastRegistryReceiver"); thread.setDaemon(true); thread.start(); } catch (Exception e) { throw new SofaRpcRuntimeException(LogCodes.getLog(LogCodes.ERROR_REGISTRY_INIT, EXT_NAME), e); } String cleanPeriodStr = registryConfig.getParameter(CLEAN_PERIOD); if (StringUtils.isNotBlank(cleanPeriodStr)) { this.cleanPeriod = Integer.parseInt(cleanPeriodStr); } if (!Boolean.FALSE.toString().equalsIgnoreCase(registryConfig.getParameter(CLEAN))) { this.cleanFuture = cleanExecutor.scheduleWithFixedDelay(new Runnable() { @Override public void run() { try { clean(); } catch (Throwable t) { LOGGER.error("Unexpected exception occur at clean expired provider, cause: " + t.getMessage(), t); } } }, cleanPeriod, cleanPeriod, TimeUnit.MILLISECONDS); } else { this.cleanFuture = null; } } protected  MulticastRegistry(RegistryConfig registryConfig); @Override boolean start(); @Override void register(ProviderConfig config); @Override void unRegister(ProviderConfig config); @Override void batchUnRegister(List<ProviderConfig> configs); @Override List<ProviderGroup> subscribe(ConsumerConfig config); @Override void unSubscribe(ConsumerConfig config); @Override void batchUnSubscribe(List<ConsumerConfig> configs); @Override void destroy(); @Override void init(); Map<String, ProviderGroup> getAllProviderCache();  }
@Test public void testInit() throws NoSuchFieldException, IllegalAccessException { RegistryConfig registryConfig = new RegistryConfig() .setProtocol("multicast") .setAddress("224.5.6.7:6666"); MulticastRegistry multicastRegistry = new MulticastRegistry(registryConfig); multicastRegistry.init(); Assert.assertEquals("224.5.6.7", getAddress(multicastRegistry)); Assert.assertEquals(6666, getPort(multicastRegistry)); }