Dubbo源码学习--Multicast 注册中心（四）

- 日理万妓 2022-05-29 11:26 216阅读 0赞

Multicast 注册中心不需要启动任何中心节点，只要广播地址一样，就可以互相发现。

![/user-guide/images/multicast.jpg][user-guide_images_multicast.jpg]

1.  提供方启动时广播自己的地址
2.  消费方启动时广播订阅请求
3.  提供方收到订阅请求时，单播自己的地址给订阅者，如果设置了 `unicast=false`，则广播给订阅者
4.  消费方收到提供方地址时，连接该地址进行 RPC 调用。

组播受网络结构限制，只适合小规模应用或开发阶段使用。组播地址段: 224.0.0.0 - 239.255.255.255

## 配置 ##

<dubbo:registry address="multicast://224.5.6.7:1234" />

或

<dubbo:registry protocol="multicast" address="224.5.6.7:1234" />

为了减少广播量，Dubbo 缺省使用单播发送提供者地址信息给消费者，如果一个机器上同时启了多个消费者进程，消费者需声明 `unicast=false`，否则只会有一个消费者能收到消息：

<dubbo:registry address="multicast://224.5.6.7:1234?unicast=false" />

或

<dubbo:registry protocol="multicast" address="224.5.6.7:1234">
        <dubbo:parameter key="unicast" value="false" />

`</dubbo:registry>`

MulticastRegistryFactory是一个工厂类，用来创建MulticastRegistry实例。

public class MulticastRegistryFactory extends AbstractRegistryFactory {
    
        public Registry createRegistry(URL url) {
            return new MulticastRegistry(url);
        }
    
    }

MulticastRegistry中主要实现了如下接口：  
（1）doRegister(URL url);//向注册中心注册服务  
（2）doUnregister(URL url);//取消服务注册  
（3）doSubscribe(URL url, NotifyListener listener);//向注册中心监听服务

（4）doUnsubscribe(URL url, NotifyListener listener);//取消监听注册中心服务

1、在MulticastRegistry构造函数中会根据广播IP建立一个线程用来接收通知数据，在接收到通知数据时会进行广播并通知本地数据

public class MulticastRegistry extends FailbackRegistry {
    
        // logging output
        private static final Logger logger = LoggerFactory.getLogger(MulticastRegistry.class);
    
        private static final int DEFAULT_MULTICAST_PORT = 1234;
    
        private final InetAddress mutilcastAddress;
    
        private final MulticastSocket mutilcastSocket;
    
        private final int mutilcastPort;
    
        private final ConcurrentMap<URL, Set<URL>> received = new ConcurrentHashMap<URL, Set<URL>>();
    
        private final ScheduledExecutorService cleanExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("DubboMulticastRegistryCleanTimer", true));
    
        private final ScheduledFuture<?> cleanFuture;
    
        private final int cleanPeriod;
    
        private volatile boolean admin = false;
    
        public MulticastRegistry(URL url) {
            super(url);
            if (url.isAnyHost()) {
                throw new IllegalStateException("registry address == null");
            }
            if (!isMulticastAddress(url.getHost())) {
                throw new IllegalArgumentException("Invalid multicast address " + url.getHost() + ", scope: 224.0.0.0 - 239.255.255.255");
            }
            try {
                mutilcastAddress = InetAddress.getByName(url.getHost());
                mutilcastPort = url.getPort() <= 0 ? DEFAULT_MULTICAST_PORT : url.getPort();
                mutilcastSocket = new MulticastSocket(mutilcastPort);
                mutilcastSocket.setLoopbackMode(false);
                mutilcastSocket.joinGroup(mutilcastAddress);
                Thread thread = new Thread(new Runnable() {
                    public void run() {
                        byte[] buf = new byte[2048];
                        DatagramPacket recv = new DatagramPacket(buf, buf.length);
                        while (!mutilcastSocket.isClosed()) {
                            try {
                                mutilcastSocket.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 (!mutilcastSocket.isClosed()) {
                                    logger.error(e.getMessage(), e);
                                }
                            }
                        }
                    }
                }, "DubboMulticastRegistryReceiver");
                thread.setDaemon(true);
                thread.start();
            } catch (IOException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            this.cleanPeriod = url.getParameter(Constants.SESSION_TIMEOUT_KEY, Constants.DEFAULT_SESSION_TIMEOUT);
            if (url.getParameter("clean", true)) {
                this.cleanFuture = cleanExecutor.scheduleWithFixedDelay(new Runnable() {
                    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;
            }
        }
    
        private static boolean isMulticastAddress(String ip) {
            int i = ip.indexOf('.');
            if (i > 0) {
                String prefix = ip.substring(0, i);
                if (StringUtils.isInteger(prefix)) {
                    int p = Integer.parseInt(prefix);
                    return p >= 224 && p <= 239;
                }
            }
            return false;
        }
    
        private void clean() {
            if (admin) {
                for (Set<URL> providers : new HashSet<Set<URL>>(received.values())) {
                    for (URL url : new HashSet<URL>(providers)) {
                        if (isExpired(url)) {
                            if (logger.isWarnEnabled()) {
                                logger.warn("Clean expired provider " + url);
                            }
                            doUnregister(url);
                        }
                    }
                }
            }
        }
    
        private boolean isExpired(URL url) {
            if (!url.getParameter(Constants.DYNAMIC_KEY, true)
                    || url.getPort() <= 0
                    || Constants.CONSUMER_PROTOCOL.equals(url.getProtocol())
                    || Constants.ROUTE_PROTOCOL.equals(url.getProtocol())
                    || Constants.OVERRIDE_PROTOCOL.equals(url.getProtocol())) {
                return false;
            }
            Socket socket = null;
            try {
                socket = new Socket(url.getHost(), url.getPort());
            } catch (Throwable e) {
                try {
                    Thread.sleep(100);
                } catch (Throwable e2) {
                }
                Socket socket2 = null;
                try {
                    socket2 = new Socket(url.getHost(), url.getPort());
                } catch (Throwable e2) {
                    return true;
                } finally {
                    if (socket2 != null) {
                        try {
                            socket2.close();
                        } catch (Throwable e2) {
                        }
                    }
                }
            } finally {
                if (socket != null) {
                    try {
                        socket.close();
                    } catch (Throwable e) {
                    }
                }
            }
            return false;
        }
    	
    	//在接收到数据时会今天广播数据
        private void receive(String msg, InetSocketAddress remoteAddress) {
            if (logger.isInfoEnabled()) {
                logger.info("Receive multicast message: " + msg + " from " + remoteAddress);
            }
            if (msg.startsWith(Constants.REGISTER)) {
                URL url = URL.valueOf(msg.substring(Constants.REGISTER.length()).trim());
    			//如果注册时会通知数据
                registered(url);
            } else if (msg.startsWith(Constants.UNREGISTER)) {
                URL url = URL.valueOf(msg.substring(Constants.UNREGISTER.length()).trim());
    			//解除注册时会通知数据
                unregistered(url);
            } else if (msg.startsWith(Constants.SUBSCRIBE)) {
                URL url = URL.valueOf(msg.substring(Constants.SUBSCRIBE.length()).trim());
                Set<URL> urls = getRegistered();
                if (urls != null && !urls.isEmpty()) {
                    for (URL u : urls) {
                        if (UrlUtils.isMatch(url, u)) {
                            String host = remoteAddress != null && remoteAddress.getAddress() != null
                                    ? remoteAddress.getAddress().getHostAddress() : url.getIp();
                            if (url.getParameter("unicast", true) // Whether the consumer's machine has only one process
                                    && !NetUtils.getLocalHost().equals(host)) { // Multiple processes in the same machine cannot be unicast with unicast or there will be only one process receiving information
    							//单播数据
    						   unicast(Constants.REGISTER + " " + u.toFullString(), host);
                            } else {
    							//广播数据
                                broadcast(Constants.REGISTER + " " + u.toFullString());
                            }
                        }
                    }
                }
            }/* else if (msg.startsWith(UNSUBSCRIBE)) {
            }*/
        }
    	//广播数据
        private void broadcast(String msg) {
            if (logger.isInfoEnabled()) {
                logger.info("Send broadcast message: " + msg + " to " + mutilcastAddress + ":" + mutilcastPort);
            }
            try {
                byte[] data = (msg + "\n").getBytes();
                DatagramPacket hi = new DatagramPacket(data, data.length, mutilcastAddress, mutilcastPort);
                mutilcastSocket.send(hi);
            } catch (Exception e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
        }
    	//单播数据
        private void unicast(String msg, String host) {
            if (logger.isInfoEnabled()) {
                logger.info("Send unicast message: " + msg + " to " + host + ":" + mutilcastPort);
            }
            try {
                byte[] data = (msg + "\n").getBytes();
                DatagramPacket hi = new DatagramPacket(data, data.length, InetAddress.getByName(host), mutilcastPort);
                mutilcastSocket.send(hi);
            } catch (Exception e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
        }
    	
    	//注册时会广播数据
        protected void doRegister(URL url) {
            broadcast(Constants.REGISTER + " " + url.toFullString());
        }
    	//消除注册时会广播数据
        protected void doUnregister(URL url) {
            broadcast(Constants.UNREGISTER + " " + url.toFullString());
        }
    
    	//订阅数据的时候会广播数据
        protected void doSubscribe(URL url, NotifyListener listener) {
            if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
                admin = true;
            }
            broadcast(Constants.SUBSCRIBE + " " + url.toFullString());
            synchronized (listener) {
                try {
                    listener.wait(url.getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
                } catch (InterruptedException e) {
                }
            }
        }
    
    	//取消订阅数据的时候会广播数据
        protected void doUnsubscribe(URL url, NotifyListener listener) {
            if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
                    && url.getParameter(Constants.REGISTER_KEY, true)) {
                unregister(url);
            }
            broadcast(Constants.UNSUBSCRIBE + " " + url.toFullString());
        }
    
        public boolean isAvailable() {
            try {
                return mutilcastSocket != null;
            } catch (Throwable t) {
                return false;
            }
        }
    
        public void destroy() {
            super.destroy();
            try {
                if (cleanFuture != null) {
                    cleanFuture.cancel(true);
                }
            } catch (Throwable t) {
                logger.warn(t.getMessage(), t);
            }
            try {
                mutilcastSocket.leaveGroup(mutilcastAddress);
                mutilcastSocket.close();
            } catch (Throwable t) {
                logger.warn(t.getMessage(), t);
            }
        }
    	//注册时通知监听者
        protected void registered(URL url) {
            for (Map.Entry<URL, Set<NotifyListener>> entry : getSubscribed().entrySet()) {
                URL key = entry.getKey();
                if (UrlUtils.isMatch(key, url)) {
                    Set<URL> urls = received.get(key);
                    if (urls == null) {
                        received.putIfAbsent(key, new ConcurrentHashSet<URL>());
                        urls = received.get(key);
                    }
                    urls.add(url);
                    List<URL> list = toList(urls);
                    for (NotifyListener listener : entry.getValue()) {
                        notify(key, listener, list);
                        synchronized (listener) {
                            listener.notify();
                        }
                    }
                }
            }
        }
    	//消除注册时会通知监听者
        protected void unregistered(URL url) {
            for (Map.Entry<URL, Set<NotifyListener>> entry : getSubscribed().entrySet()) {
                URL key = entry.getKey();
                if (UrlUtils.isMatch(key, url)) {
                    Set<URL> urls = received.get(key);
                    if (urls != null) {
                        urls.remove(url);
                    }
                    if (urls == null || urls.isEmpty()){
                        if (urls == null){
                            urls = new ConcurrentHashSet<URL>();
                        }
                        URL empty = url.setProtocol(Constants.EMPTY_PROTOCOL);
                        urls.add(empty);
                    }
                    List<URL> list = toList(urls);
                    for (NotifyListener listener : entry.getValue()) {
                        notify(key, listener, list);
                    }
                }
            }
        }
    }

[user-guide_images_multicast.jpg]: http://dubbo.io/books/dubbo-user-book/sources/images/multicast.jpg