关于一个简单的负载均衡实现的问题（多线程相关）

0 0

关于一个简单的负载均衡实现的问题（多线程相关）5

最近做一个Demo，里面涉及到一个客户端负载均衡的问题。基本问题如下：
1. A 服务依赖 B，C等服务（均为Rest风格服务，JSON作为数据协议）。
2. B，C等服务分别有N个节点，忽略权重（所有节点权重一样）。
3. A可以通过Zookeeper拿到每个依赖服务的节点列表。比如，A - [10.10.10.10, 10.10.10.11, 10.10.10.12], B - [10.10.10.20, 10.10.10.21, 10.10.10.22]
4. 现在我要做的是通过简单的统一包装类，封装请求的执行过程。负载均衡使用简单Polling模式。

问题：
Endpoints的next 方法是否需要加锁？为什么
目前的实现有什么问题？

下面是我的实现代码（基于JDK1.8）：

import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author: nealmi
 * Date: 4/9/14
 * Time: 10:24 PM
 */
public class PollingEndpointManagerImpl implements EndpointManager {
    private static final Log logger = LogFactory.getLog(PollingEndpointManagerImpl.class);
    private final Map<String, Endpoints> endpointsMapping = new ConcurrentHashMap<>(3, .75f, 16);
    private final RestTemplate restTemplate;
    @Autowired(required = false)
    private int maxRetries = 3;

    public PollingEndpointManagerImpl() {
        this.restTemplate = new RestTemplate();
    }

    public PollingEndpointManagerImpl(RestTemplate restTemplate) {
        this.restTemplate = restTemplate;
    }

    public PollingEndpointManagerImpl(RestTemplate restTemplate, int maxRetries) {
        this.restTemplate = restTemplate;
        this.maxRetries = maxRetries;
    }

    public int getMaxRetries() {
        return maxRetries;
    }

    public void setMaxRetries(int maxRetries) {
        this.maxRetries = maxRetries;
    }

    @Override
    public void refresh(String key, List<String> endpoints) {
        endpointsMapping.put(key, Endpoints.create(key, endpoints));
    }

    private String next(Endpoints endpoints) {
        return endpoints.next();
    }

    @Override
    public String next(String key) {
        if (!endpointsMapping.containsKey(key)) {
            throw new RuntimeException("No endpoints for key:" + key);
        }
        return next(endpointsMapping.get(key));
    }

    @Override
    public <T> T get(String key, String uri, Class<T> t) {
        return doIt(() -> {
            String url = getFullUrl(key, uri);
            if (logger.isInfoEnabled()) {
                logger.info("Retrieving data from " + url);
            }
            ResponseEntity<T> responseEntity = restTemplate.getForEntity(url, t);
            return responseEntity.getBody();
        });
    }

    @Override
    public <T> T get(String key, String uri, ParameterizedTypeReference<T> t) {
        return doIt(() -> {
            String url = getFullUrl(key, uri);
            if (logger.isInfoEnabled()) {
                logger.info("Retrieving data from " + url);
            }
            ResponseEntity<T> responseEntity = restTemplate.exchange(url, HttpMethod.GET, null, t);
            return responseEntity.getBody();
        });
    }

    private <T> T doIt(RequestExecutor<T> requestExecutor) {
        int counter = 0;
        while (true) {
            try {
                return requestExecutor.execute();
            } catch (Exception e) {
                if (counter < maxRetries) {
                    counter++;
                    if (logger.isInfoEnabled()) {
                        logger.info("Request failure, retrying... counter= " + counter + ", error:" + e.getMessage());
                    }
                } else {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private static interface RequestExecutor<T> {
        T execute();
    }

    private String getFullUrl(String key, String uri) {
        return "http://" + next(key) + uri;
    }

    private static class Endpoints {
        private String key;
        private int counter = 0;
        private List<String> endpoints = Lists.newArrayList();
        private ReentrantLock lock = new ReentrantLock();

        private Endpoints() {
        }

        private Endpoints(String key, List<String> endpoints) {
            this.key = key;
            this.endpoints = endpoints;
        }

        private static Endpoints create(String key, List<String> endpoints) {
            return new Endpoints(key, endpoints);
        }

        public String getKey() {
            return key;
        }

        public List<String> getEndpoints() {
            return endpoints;
        }

        public String next() {
            try {
                lock.tryLock(10, TimeUnit.SECONDS);
                if (counter >= endpoints.size()) {
                    counter = (0);
                }
                String endpoint = endpoints.get(counter);
                counter++;
                return endpoint;
            } catch (Exception e) {
                logger.error("Error when getting next endpoint. counter=" + counter + "  endpoints.size=" + endpoints.size(), e);
            } finally {
                lock.unlock();
            }

            return null;
        }
    }
}

多线程负载均衡

2014年4月21日 16:45

iMzw
25
0 0 1

1个答案按时间排序按投票排序

0 0

Endpoints的next需要加锁，它负责轮训，如果多个线程同一时间执行到
if (counter >= endpoints.size()) {
counter = (0);
}
有可能连续二次都是从endpoints.get(0)的服务执行，也就没法保证轮询的正确性，或者count++也是多线程同时读的0，有个可能线程1， 0++，现场 0++。

你用到了最新的语法，还没具体看新的语法，不会。不过看程序大致是一个前端的一个
负载均衡,

2014年4月22日 08:02