最近路径负载均衡，多线程任务ping各个注册中心，获取速度

NearestLoadBalance.java

package com.jd.registry.failover;

import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.*;

import com.jd.registry.Registry;
import com.jd.registry.util.URL;

/**
 * 最近路径负载均衡
 *
 * @author hexiaofeng
 * @version 1.0.0
 * @since 13-1-8 下午2:42
 */
public class NearestLoadBalance implements LoadBalance {

    @Override
    public List<Registry> sort(List<Registry> registrys) {
        if (registrys == null || registrys.size() <= 1) {
            return registrys;
        }

        // 创建多线程任务
        ExecutorService service = Executors.newFixedThreadPool(registrys.size());
        List<PingTask> tasks = new ArrayList<PingTask>();
        for (Registry registry : registrys) {
            tasks.add(new PingTask(registry));
        }
        try {
            // ping各个注册中心，获取速度
            List<Future<Speed>> futures = service.invokeAll(tasks);
            List<Speed> speeds = new ArrayList<Speed>();

            // 获取速度
            for (Future<Speed> future : futures) {
                try {
                    speeds.add(future.get());
                }
                catch (ExecutionException e) {}
            }

            // 按响应时间升序排序
            Collections.sort(speeds, new Comparator<Speed>() {
                @Override
                public int compare(Speed o1, Speed o2) {
                    if (o1.getHost().equals(o2.getHost())) {
                        return 0;
                    }
                    long time = o1.getTime() - o2.getTime();
                    return (time == 0) ? 1 : ((time > 0) ? 1 : -1);
                }
            });

            // 返回结果
            List<Registry> result = new ArrayList<Registry>();
            for (Speed speed : speeds) {
                result.add(speed.getRegistry());
            }
            return result;
        }
        catch (InterruptedException e) {
            return registrys;
        }
        finally {
            service.shutdownNow();
        }
    }

    @Override
    public String getType() {
        return "nearest";
    }

    @Override
    public void setUrl(URL url) {}

    /**
     * 速度
     */
    protected class Speed {
        private Registry registry;
        private String host;
        private long time;

        public Speed(Registry registry, String host, long time) {
            this.registry = registry;
            this.host = host;
            this.time = time;
        }

        public Registry getRegistry() {
            return registry;
        }

        public long getTime() {
            return time;
        }

        public String getHost() {
            return host;
        }
    }

    /**
     * 获取注册中心访问速度
     */
    protected class PingTask implements Callable<Speed> {
        private Registry registry;

        public PingTask(Registry registry) {
            this.registry = registry;
        }

        @Override
        public Speed call() throws Exception {
            URL url = registry.getUrl();
            if (url == null || url.getHost() == null || url.getHost().isEmpty()) {
                return new Speed(registry, "", Long.MAX_VALUE);
            }

            // zookeeper包含多个地址，用逗号隔开，取第一个地址测速
            String[] hosts = url.getHost().split(",");
            String host = hosts[0];
            // 去除端口
            int pos = host.indexOf(':');
            if (pos > 0) {
                host = host.substring(0, pos);
            }
            // 超时时间
            int timeout = url.getPositiveParameter("timeout", 5000);
            try {
                InetAddress address = InetAddress.getByName(host);
                long time = System.currentTimeMillis();
                if (address.isReachable(timeout)) {
                    time = System.currentTimeMillis() - time;
                } else {
                    return new Speed(registry, host, Long.MAX_VALUE);
                }
                return new Speed(registry, host, time);
            }
            catch (Exception e) {
                return new Speed(registry, host, Long.MAX_VALUE);
            }
        }
    }

}