Spring cloud中的服務自定義路由

阿新 • • 發佈：2019-01-21

很多情況下我們需要服務自定義路由，比如需要灰度釋出時線上驗證環境、生產環境的服務例項路由是需要區分的，還有在SAAS化應用中，經常會把租戶分成一個個組，每組分配幾個服務例項，就是說組內服務例項共享，組間是隔離的。

本文在Spring Cloud的基礎上，給出了一個服務分組和自定義路由的方案，並提供了範例程式碼，程式碼開源地址為：

https://github.com/tangaiyun/custom-routing-for-Spring-Cloud-service

方案的基本思路為：

服務釋出時註冊到Eureka上，服務釋出時必須指定appGroupName，常用的指定方法為：
1. 在java程式啟動引數中新增
--eureka.instance.app-group-name=group_1
2. docker 啟動的話，在docker執行環境引數中新增：
-e "eureka.instance.app-group-name=group_1"
以上都是要指定服務例項的appGroupName為group_1。

服務訪問時必須通過ZUUL閘道器按服務名字訪問
比如http://localhost:8060/microservice-provider-user/1
8086為ZUUL的埠，microservice-provider-user為服務的名字
服務訪問時必須在HTTP header 或cookie中提供一個路由碼，本例中它的名字為“ROUTECODE”
自定義ZUUL，實現一個父類為AbstractLoadBalancerRule的類，本案中名字為“MyCustomRule”

重點定義MyCustomRule的 public Server choose(ILoadBalancer lb, Object key)方法
1. 讀取ZK中的配置，初始化物件,並監控ZK中配置的變化
2. 獲取request中cookie或header中名為"ROUTECODE"屬性值
3. 根據ROUTECODE值對映到一個服務分組或者對映失敗則使用預設分組
4. 獲取所有服務例項，按按照appGroupName過濾並排序

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
 
import javax.servlet.http.Cookie;
import javax.servlet.http.HttpServletRequest;
 
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
 
import com.netflix.client.config.IClientConfig;
import com.netflix.loadbalancer.AbstractLoadBalancerRule;
import com.netflix.loadbalancer.ILoadBalancer;
import com.netflix.loadbalancer.Server;
import com.netflix.niws.loadbalancer.DiscoveryEnabledServer;
import com.netflix.zuul.context.RequestContext;
 
public class MyCustomRule extends AbstractLoadBalancerRule {
 
	private ConcurrentHashMap<String, AtomicInteger> nextServerCyclicCounterMap;
 
	private static Logger log = LoggerFactory.getLogger(MyCustomRule.class);
 
	private static final String ROUTE_HEAD_COOKIE_NAME = "ROUTECODE";
	private static final String ZK_CONN_STR = "192.168.0.106:2181,192.168.0.135:2181,192.168.0.143:2181";
	private static final String ROUTE_CONFIG_ZK_PATH = "/config/zuul/route";
	private static final String GROUP_PREFIX = "group_";
	private static final String ALL_AS_ONE_GROUP = "onegroup";
	private static final String DEFAULT_ROUTE_KEY = "default_route";
	private static final String ENABLE_KEY = "enable";
	private volatile boolean isAcmInited = false;
	private Properties routeConfigProperties = new Properties();
	private boolean isCustomRouteEnable = false;
	private ConcurrentHashMap<String, ArrayList<String>> groupMemberMap;
	private String defaultRoute;
 
	public MyCustomRule() {
		nextServerCyclicCounterMap = new ConcurrentHashMap<String, AtomicInteger>();
		nextServerCyclicCounterMap.put(ALL_AS_ONE_GROUP, new AtomicInteger(0));
		groupMemberMap = new ConcurrentHashMap<String, ArrayList<String>>();
	}
 
	public MyCustomRule(ILoadBalancer lb) {
		this();
		setLoadBalancer(lb);
	}
 
	private synchronized void configZKInit() {
 
		System.out.println(" loading route config from ZK*****************************************************");
		CuratorFramework client = CuratorFrameworkFactory.newClient(ZK_CONN_STR, new ExponentialBackoffRetry(1000, 3));
		client.start();
		try {
			final CountDownLatch cdl = new CountDownLatch(1);
			@SuppressWarnings("resource")
			final NodeCache cache = new NodeCache(client, ROUTE_CONFIG_ZK_PATH);
			NodeCacheListener listener = () -> {
				System.out.println("enter listener...");
				ChildData data = cache.getCurrentData();
				if (null != data) {
					String strRouteConfig = new String(cache.getCurrentData().getData());
					System.out.println("節點資料：" + strRouteConfig);
					loadConfig(strRouteConfig);
 
				} else {
					System.out.println("節點被刪除!");
				}
				cdl.countDown();
				System.out.println("countdownlatch count!");
			};
			cache.getListenable().addListener(listener);
 
			cache.start();
			isAcmInited = true;
			cdl.await();
			System.out.println("countdownlatch await finish!");
		} catch (Exception e) {
			log.error("路由配置節點監控器啟動錯誤", e);
		}
	}
 
	private void loadConfig(String content) {
		try {
			routeConfigProperties.load(new ByteArrayInputStream(content.getBytes()));
			isCustomRouteEnable = Boolean.parseBoolean(routeConfigProperties.getProperty(ENABLE_KEY).trim());
			Set<Object> keySet = routeConfigProperties.keySet();
			for(Object key : keySet) {
				String strKey = (String)key;
				if(strKey.startsWith(GROUP_PREFIX)) {
					ArrayList<String> groupMemberList = new ArrayList<String>();
					groupMemberList.addAll(Arrays.asList(routeConfigProperties.getProperty(strKey).split(",")));
					groupMemberMap.put(strKey, groupMemberList);
					nextServerCyclicCounterMap.put(strKey, new AtomicInteger(0));
				}
			}
			defaultRoute = routeConfigProperties.getProperty(DEFAULT_ROUTE_KEY);
		} catch (IOException e) {
			log.error("配置讀入到Properties物件失敗", e);
		}
	}
 
	public Server choose(ILoadBalancer lb, Object key) {
		if (lb == null) {
			log.warn("no load balancer");
			return null;
		}
 
		if (!isAcmInited) {
			configZKInit();
		}
 
		Server server = null;
		if (!isCustomRouteEnable) {
			int count = 0;
			while (server == null && count++ < 10) {
				List<Server> reachableServers = lb.getReachableServers();
				List<Server> allServers = lb.getAllServers();
				int upCount = reachableServers.size();
				int serverCount = allServers.size();
 
				if ((upCount == 0) || (serverCount == 0)) {
					log.warn("No up servers available from load balancer: " + lb);
					return null;
				}
 
				int nextServerIndex = incrementAndGetModulo(ALL_AS_ONE_GROUP, serverCount);
				List<Server> sortedAllServers = allServers.stream().sorted((s1, s2) -> s1.getId().compareTo(s2.getId()))
						.collect(Collectors.toList());
				System.out.println("all servers info:");
				printServersInfo(sortedAllServers);
				server = sortedAllServers.get(nextServerIndex);
 
				if (server == null) {
					/* Transient. */
					Thread.yield();
					continue;
				}
 
				if (server.isAlive() && (server.isReadyToServe())) {
					System.out.println("this request will be served by the following server:");
					printServerInfo(server);
					return (server);
				}
				// Next.
				server = null;
			}
 
			if (count >= 10) {
				log.warn("No available alive servers after 10 tries from load balancer: " + lb);
			}
		} else {
			int count = 0;
			while (server == null && count++ < 10) {
				List<Server> reachableServers = lb.getReachableServers();
				List<Server> allServers = lb.getAllServers();
				int upCount = reachableServers.size();
				int serverCount = allServers.size();
 
				if ((upCount == 0) || (serverCount == 0)) {
					log.warn("No up servers available from load balancer: " + lb);
					return null;
				}
				String routeKey = getRouteKey();
				String routeGroup = getGroupbyRouteKey(routeKey);
				if(routeGroup == null) {
					routeGroup = defaultRoute;
				}
				final String appFilter = routeGroup;
				List<Server> serverCandidates = allServers.stream()
						.filter(s -> appFilter
								.equalsIgnoreCase(((DiscoveryEnabledServer) s).getInstanceInfo().getAppGroupName()))
						.sorted((s1, s2) -> s1.getId().compareTo(s2.getId())).collect(Collectors.toList());
				
				int nextServerIndex = incrementAndGetModulo(routeGroup,serverCandidates.size());
				server = serverCandidates.get(nextServerIndex);
			
				
				if (server == null) {
					/* Transient. */
					Thread.yield();
					continue;
				}
				if (server.isAlive() && (server.isReadyToServe())) {
					System.out.println("this request will be served by the following server:");
					printServerInfo(server);
					return (server);
				}
				// Next.
				server = null;
			}
			if (count >= 10) {
				log.warn("No available alive servers after 10 tries from load balancer: " + lb);
			}
		}
		return server;
	}
 
	private void printServersInfo(Collection<Server> servers) {
		for (Server s : servers) {
			printServerInfo(s);
		}
	}
 
	private void printServerInfo(Server server) {
		System.out.print("appName: " + ((DiscoveryEnabledServer) server).getInstanceInfo().getAppName() + " ");
		System.out.print("appGroup: " + ((DiscoveryEnabledServer) server).getInstanceInfo().getAppGroupName() + " ");
		System.out.print("id: " + server.getId() + " isAlive: " + server.isAlive() + " ");
		System.out.print("id: " + server.getId() + " isReadyToServe: " + server.isReadyToServe() + " ");
		System.out.println();
	}
 
	/**
	 * first find
	 * 
	 * @return
	 */
	private String getRouteKey() {
		RequestContext ctx = RequestContext.getCurrentContext();
		HttpServletRequest request = ctx.getRequest();
		String orgCode = "";
		Cookie[] cookies = request.getCookies();
		if (cookies != null) {
			for (Cookie cookie : cookies) {
				if (ROUTE_HEAD_COOKIE_NAME.equals(cookie.getName())) {
					orgCode = cookie.getValue();
					break;
				}
			}
		}
		if ("".equals(orgCode)) {
			orgCode = request.getHeader(ROUTE_HEAD_COOKIE_NAME);
		}
 
		return orgCode;
	}
	
	private String getGroupbyRouteKey(String routeKey) {
		String group = null;
		Set<Entry<String, ArrayList<String>>> entrySet = groupMemberMap.entrySet();
		for(Entry<String, ArrayList<String>> entry : entrySet) {
			List<String> memberList = entry.getValue();
			if(memberList.contains(routeKey)) {
				group = entry.getKey();
				break;
			}
		}
		return group;
	}
 
	/**
	 * Inspired by the implementation of {@link AtomicInteger#incrementAndGet()}.
	 *
	 * @param modulo
	 *            The modulo to bound the value of the counter.
	 * @return The next value.
	 */
	private int incrementAndGetModulo(String group, int modulo) {
		AtomicInteger groupCurrent = nextServerCyclicCounterMap.get(group);
		for (;;) {
			int current = groupCurrent.get();
			int next = (current + 1) % modulo;
			if (groupCurrent.compareAndSet(current, next))
				return next;
		}
	}
 
	@Override
	public Server choose(Object key) {
		return choose(getLoadBalancer(), key);
	}
 
	@Override
	public void initWithNiwsConfig(IClientConfig clientConfig) {
	}
}

ZUUL中的配置

server:
  port: 8060
spring:
  application:
    name: microservice-gateway-zuul
eureka:
  client:
    service-url:
      defaultZone: http://peer1:8761/eureka/,http://peer2:8762/eureka/
  instance:
    prefer-ip-address: true
            
microservice-provider-user:
  ribbon:
    NFLoadBalancerRuleClassName: com.tay.customroute.MyCustomRule  
microservice-provider-user2:
  ribbon:
    NFLoadBalancerRuleClassName: com.tay.customroute.MyCustomRule

特別注意下面的

    microservice-provider-user:
      ribbon:
        NFLoadBalancerRuleClassName: com.tay.customroute.MyCustomRule

這個配置的意思就是名字為
```
microservice-provider-user
```
的服務將使用com.tay.customroute.MyCustomRule 作為負載均衡器的路由規則，即我們自義定的路由規則。
ZK中配置範例
```
    enable=true
    group_1=org1,org2,org3,org4,org5,org6,org7,org8,org9,org10
    group_2=org11,org12,org13,org15
    group_3=org20,org21,org22,org23
    default_route=group_1
```
這個配置的含義為：
自定義路由配置當前生效。
當前有三個組group_1,group_2,group_3,每個組包含一組租戶機構碼（tenaut code），有多少個組完全由使用者自己定義，但每個組組名必須以“group_”開頭。
如果一個tenaut code沒有包含在以上3個組中，則預設它歸屬為group_1。
元件關係圖

Spring cloud中的服務自定義路由

spring-cloud中zuul自定義service級別，api級別的路由白名單

spring cloud中Ribbon自定義負載均衡策略

Spring cloud中的服務自定義路由

Spring Boot中的自定義start pom

在spring boot中使用自定義的properties

在spring boot中如何自定義Exception異常

spring專案中通過自定義applicationContext工具類獲取到applicationContext上下文物件

Spring Boot中的自定義起步依賴

在spring Controller中返回自定義的Http code

spring cloud的Eureka自定義介面，啟動介面自定義

Spring Cloud中服務的發現與消費（3）

掃描自定義註解並在spring容器中注入自定義bean

閘道器服務自定義路由規則（springcloud+nacos）

Spring Boot下如何自定義Repository中的DAO方法

spring中增加自定義配置支持

微服務實施Spring Cloud中踩過的坑（轉）

如何在SAP Cloud for Customer自定義BO中創建訪問控制

spring cloud中微服務之間的調用以及eureka的自我保護機制

Spring Cloud中的服務間通訊(RestTemplate和Feign)

Spring Cloud 中Eureka服務註冊中心的搭建

Spring cloud中的服務自定義路由

相關推薦