class 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 + "://" + addressInput;
            }
            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(); // Remove the expired
                    } catch (Throwable t) { // Defensive fault tolerance
                        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);

}