共计 6423 个字符,预计需要花费 17 分钟才能阅读完成。
序
本文主要研究一下 nacos 的 DistroFilter
CanDistro
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/web/CanDistro.java
@Retention(RetentionPolicy.RUNTIME)
public @interface CanDistro {}
- CanDistro 用于标识一个方法需要判断是否应该根据 distro 被重定向
DistroFilter
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/web/DistroFilter.java
public class DistroFilter implements Filter {
private static final int PROXY_CONNECT_TIMEOUT = 2000;
private static final int PROXY_READ_TIMEOUT = 2000;
@Autowired
private DistroMapper distroMapper;
@Autowired
private SwitchDomain switchDomain;
@Autowired
private FilterBase filterBase;
@Override
public void init(FilterConfig filterConfig) throws ServletException { }
@Override
public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {HttpServletRequest req = (HttpServletRequest) servletRequest;
HttpServletResponse resp = (HttpServletResponse) servletResponse;
String urlString = req.getRequestURI();
if (StringUtils.isNotBlank(req.getQueryString())) {urlString += "?" + req.getQueryString();
}
try {String path = new URI(req.getRequestURI()).getPath();
String serviceName = req.getParameter(CommonParams.SERVICE_NAME);
// For client under 0.8.0:
if (StringUtils.isBlank(serviceName)) {serviceName = req.getParameter("dom");
}
Method method = filterBase.getMethod(req.getMethod(), path);
if (method == null) {throw new NoSuchMethodException(req.getMethod() + " " + path);
}
String groupName = req.getParameter(CommonParams.GROUP_NAME);
if (StringUtils.isBlank(groupName)) {groupName = Constants.DEFAULT_GROUP;}
// use groupName@@serviceName as new service name:
String groupedServiceName = serviceName;
if (StringUtils.isNotBlank(serviceName) && !serviceName.contains(Constants.SERVICE_INFO_SPLITER)) {groupedServiceName = groupName + Constants.SERVICE_INFO_SPLITER + serviceName;}
// proxy request to other server if necessary:
if (method.isAnnotationPresent(CanDistro.class) && !distroMapper.responsible(groupedServiceName)) {String userAgent = req.getHeader("User-Agent");
if (StringUtils.isNotBlank(userAgent) && userAgent.contains(UtilsAndCommons.NACOS_SERVER_HEADER)) {
// This request is sent from peer server, should not be redirected again:
Loggers.SRV_LOG.error("receive invalid redirect request from peer {}", req.getRemoteAddr());
resp.sendError(HttpServletResponse.SC_BAD_REQUEST,
"receive invalid redirect request from peer" + req.getRemoteAddr());
return;
}
List<String> headerList = new ArrayList<>(16);
Enumeration<String> headers = req.getHeaderNames();
while (headers.hasMoreElements()) {String headerName = headers.nextElement();
headerList.add(headerName);
headerList.add(req.getHeader(headerName));
}
HttpClient.HttpResult result =
HttpClient.request("http://" + distroMapper.mapSrv(groupedServiceName) + urlString, headerList,
StringUtils.isBlank(req.getQueryString()) ? HttpClient.translateParameterMap(req.getParameterMap()) : new HashMap<>(2)
, PROXY_CONNECT_TIMEOUT, PROXY_READ_TIMEOUT, "UTF-8", req.getMethod());
try {resp.setCharacterEncoding("UTF-8");
resp.getWriter().write(result.content);
resp.setStatus(result.code);
} catch (Exception ignore) {Loggers.SRV_LOG.warn("[DISTRO-FILTER] request failed:" + distroMapper.mapSrv(groupedServiceName) + urlString);
}
return;
}
OverrideParameterRequestWrapper requestWrapper = OverrideParameterRequestWrapper.buildRequest(req);
requestWrapper.addParameter(CommonParams.SERVICE_NAME, groupedServiceName);
filterChain.doFilter(requestWrapper, resp);
} catch (AccessControlException e) {resp.sendError(HttpServletResponse.SC_FORBIDDEN, "access denied:" + UtilsAndCommons.getAllExceptionMsg(e));
return;
} catch (NoSuchMethodException e) {resp.sendError(HttpServletResponse.SC_NOT_IMPLEMENTED, "no such api:" + e.getMessage());
return;
} catch (Exception e) {
resp.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
"Server failed," + UtilsAndCommons.getAllExceptionMsg(e));
return;
}
}
@Override
public void destroy() {}
}
- DistroFilter 实现了 servlet 的 Filter 接口;其 doFilter 方法会从 servletRequest 中读取 serviceName、method、groupName 等,然后判断 method 是否标注 CanDistro,如果是而且 distroMapper 不负责该 service 则构建 http 请求然后将结果写回 Filter;如果不需要重定向则继续 filterChain.doFilter
HttpClient.request
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/misc/HttpClient.java
public class HttpClient {
private static final int TIME_OUT_MILLIS = 10000;
private static final int CON_TIME_OUT_MILLIS = 5000;
private static AsyncHttpClient asyncHttpClient;
private static CloseableHttpClient postClient;
//......
public static HttpResult request(String url, List<String> headers, Map<String, String> paramValues, int connectTimeout, int readTimeout, String encoding, String method) {
HttpURLConnection conn = null;
try {String encodedContent = encodingParams(paramValues, encoding);
url += (null == encodedContent) ? "": ("?" + encodedContent);
conn = (HttpURLConnection) new URL(url).openConnection();
conn.setConnectTimeout(connectTimeout);
conn.setReadTimeout(readTimeout);
conn.setRequestMethod(method);
conn.addRequestProperty("Client-Version", UtilsAndCommons.SERVER_VERSION);
conn.addRequestProperty("User-Agent", UtilsAndCommons.SERVER_VERSION);
setHeaders(conn, headers, encoding);
conn.connect();
return getResult(conn);
} catch (Exception e) {Loggers.SRV_LOG.warn("Exception while request: {}, caused: {}", url, e);
return new HttpResult(500, e.toString(), Collections.<String, String>emptyMap());
} finally {if (conn != null) {conn.disconnect();
}
}
}
private static HttpResult getResult(HttpURLConnection conn) throws IOException {int respCode = conn.getResponseCode();
InputStream inputStream;
if (HttpURLConnection.HTTP_OK == respCode) {inputStream = conn.getInputStream();
} else {inputStream = conn.getErrorStream();
}
Map<String, String> respHeaders = new HashMap<String, String>(conn.getHeaderFields().size());
for (Map.Entry<String, List<String>> entry : conn.getHeaderFields().entrySet()) {respHeaders.put(entry.getKey(), entry.getValue().get(0));
}
String gzipEncoding = "gzip";
if (gzipEncoding.equals(respHeaders.get(HttpHeaders.CONTENT_ENCODING))) {inputStream = new GZIPInputStream(inputStream);
}
HttpResult result = new HttpResult(respCode, IOUtils.toString(inputStream, getCharset(conn)), respHeaders);
inputStream.close();
return result;
}
public static class HttpResult {
final public int code;
final public String content;
final private Map<String, String> respHeaders;
public HttpResult(int code, String content, Map<String, String> respHeaders) {
this.code = code;
this.content = content;
this.respHeaders = respHeaders;
}
public String getHeader(String name) {return respHeaders.get(name);
}
}
//......
}
- HttpClient 的 request 方法直接使用 jdk 的 HttpURLConnection 进行请求,返回结果封装为 HttpResult,其 content 即为响应的 body
小结
DistroFilter 实现了 servlet 的 Filter 接口;其 doFilter 方法会从 servletRequest 中读取 serviceName、method、groupName 等,然后判断 method 是否标注 CanDistro,如果是而且 distroMapper 不负责该 service 则构建 http 请求然后将结果写回 Filter;如果不需要重定向则继续 filterChain.doFilter
doc
- DistroFilter
正文完