-
Notifications
You must be signed in to change notification settings - Fork 43
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #2 from yangzhenkun/zk
增加注册中心zk
- Loading branch information
Showing
26 changed files
with
766 additions
and
389 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
155 changes: 81 additions & 74 deletions
155
com.krpc.client/src/main/java/com/krpc/client/KRPC.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,87 +1,94 @@ | ||
| package com.krpc.client; | ||
|
|
||
| import java.io.File; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| import com.krpc.client.core.ZkRegisterCenter; | ||
| import com.krpc.client.entity.Address; | ||
| import com.krpc.client.entity.ServiceParams; | ||
| import com.krpc.common.entity.ZookeeperInfo; | ||
| import org.dom4j.Document; | ||
| import org.dom4j.Element; | ||
| import org.dom4j.io.SAXReader; | ||
|
|
||
| import com.krpc.client.entity.Address; | ||
| import com.krpc.client.entity.ServiceParams; | ||
| import java.io.File; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| _ _______ _____ _____ | ||
| | |/ | __ \| __ \ / ____| | ||
| | ' /| |__) | |__) | | | ||
| | < | _ /| ___/| | | ||
| | . \| | \ \| | | |____ | ||
| |_|\_|_| \_|_| \_____| | ||
| @author yangzhenkun | ||
| * _ _______ _____ _____ | ||
| * | |/ | __ \| __ \ / ____| | ||
| * | ' /| |__) | |__) | | | ||
| * | < | _ /| ___/| | | ||
| * | . \| | \ \| | | |____ | ||
| * |_|\_|_| \_|_| \_____| | ||
| * | ||
| * @author yangzhenkun | ||
| */ | ||
|
|
||
|
|
||
|
|
||
| public class KRPC { | ||
|
|
||
| private static Map<String,ServiceParams> serviceCache = new HashMap<String,ServiceParams>(); | ||
|
|
||
| /** | ||
| * 初始化客户端配置文件 | ||
| * | ||
| * @param clientPath | ||
| * @throws Exception | ||
| */ | ||
| public static void init(String clientPath) throws Exception { | ||
|
|
||
| // 读取该服务的配置文件 | ||
| SAXReader reader = new SAXReader(); | ||
| Document document = reader.read(new File(clientPath)); | ||
| Element root = document.getRootElement(); | ||
|
|
||
| List<Element> serviceNodes = root.elements("Service"); | ||
|
|
||
| for(Element serviceNode:serviceNodes){ | ||
| ServiceParams serviceParams = new ServiceParams(); | ||
|
|
||
| serviceParams.setServiceName(serviceNode.attributeValue("name")); | ||
|
|
||
| Element loadBalanceNode = serviceNode.element("Loadbalance"); | ||
| Element serverNode = loadBalanceNode.element("Server"); | ||
| serviceParams.setTimeout(Integer.parseInt(serverNode.attributeValue("timeout"))); | ||
| List<Element> addrNodes = serverNode.elements("addr"); | ||
|
|
||
| for(Element addrNode : addrNodes){ | ||
| Address addr = new Address(); | ||
| addr.setName(addrNode.attributeValue("name")); | ||
| addr.setHost(addrNode.attributeValue("host")); | ||
| addr.setPort(Integer.parseInt(addrNode.attributeValue("port"))); | ||
|
|
||
| serviceParams.addAddress(addr); | ||
| } | ||
|
|
||
| serviceCache.put(serviceParams.getServiceName(), serviceParams); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * 获取服务配置 | ||
| * @param serviceName | ||
| * @return | ||
| */ | ||
| public static ServiceParams getService(String serviceName){ | ||
|
|
||
| return serviceCache.get(serviceName); | ||
| } | ||
|
|
||
|
|
||
|
|
||
|
|
||
| } | ||
| /** | ||
| * 初始化客户端配置文件 | ||
| * | ||
| * @param clientPath | ||
| * @throws Exception | ||
| */ | ||
| public static void init(String clientPath) throws Exception { | ||
|
|
||
| // 读取该服务的配置文件 | ||
| SAXReader reader = new SAXReader(); | ||
| Document document = reader.read(new File(clientPath)); | ||
| Element root = document.getRootElement(); | ||
|
|
||
| List<Element> serviceNodes = root.elements("Service"); | ||
|
|
||
| ZookeeperInfo zookeeperInfo = ZookeeperInfo.createByElement(root); | ||
| if (zookeeperInfo != null) { | ||
| ZkRegisterCenter.init(zookeeperInfo); | ||
| } | ||
|
|
||
| /** | ||
| * 解析所有服务配置信息 | ||
| */ | ||
| for (Element serviceNode : serviceNodes) { | ||
| String serverName = serviceNode.attributeValue("name"); | ||
| ServiceParams serviceParams = new ServiceParams(serverName); | ||
| serviceParams.setServiceName(serverName); | ||
| serviceParams.setTimeout(Integer.valueOf(serviceNode.attributeValue("timeout"))); | ||
|
|
||
| /** | ||
| * 如果配置了注册中心，直接获取 | ||
| */ | ||
| if (zookeeperInfo != null) { | ||
| serviceParams.setAddresses(ZkRegisterCenter.getServerAddr(serverName)); | ||
| } | ||
|
|
||
| /** | ||
| * 解析直连ip，如果使用注册中心会覆盖zk中的配置信息 | ||
| */ | ||
| Element loadBalanceNode = serviceNode.element("Loadbalance"); | ||
| if (loadBalanceNode != null) { | ||
|
|
||
| Element serverNode = loadBalanceNode.element("Server"); | ||
| List<Element> addrNodes = serverNode.elements("addr"); | ||
| List<Address> addresses = serviceParams.getAddresses(); | ||
| if (addresses != null && addresses.size() > 0) { | ||
| addresses.clear(); | ||
| } else { | ||
| addresses = new ArrayList<>(); | ||
| } | ||
| for (Element addrNode : addrNodes) { | ||
| Address addr = new Address(); | ||
| addr.setName(addrNode.attributeValue("name")); | ||
| addr.setHost(addrNode.attributeValue("host")); | ||
| addr.setPort(Integer.parseInt(addrNode.attributeValue("port"))); | ||
| addresses.add(addr); | ||
| } | ||
|
|
||
| serviceParams.setAddresses(addresses); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
79 changes: 39 additions & 40 deletions
79
com.krpc.client/src/main/java/com/krpc/client/core/RequestHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,67 +1,66 @@ | ||
| package com.krpc.client.core; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Map; | ||
| import java.util.Objects; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import com.krpc.client.KRPC; | ||
| import com.krpc.client.entity.Address; | ||
| import com.krpc.client.entity.ServiceParams; | ||
| import com.krpc.client.net.TCPClient; | ||
| import com.krpc.common.entity.Request; | ||
| import com.krpc.common.serializer.HessianUtil; | ||
| import com.krpc.common.util.CompressUtil; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Map; | ||
| import java.util.Objects; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| /** | ||
| * 选择服务，进行tcp请求 | ||
| * | ||
| * @author yangzhenkun | ||
| * | ||
| * @author yangzhenkun | ||
| */ | ||
| public class RequestHandler { | ||
|
|
||
| private static Logger log = LoggerFactory.getLogger(RequestHandler.class); | ||
| private static Logger log = LoggerFactory.getLogger(RequestHandler.class); | ||
|
|
||
| private static Map<Address, TCPClient> tcpClientCache = new ConcurrentHashMap(); | ||
|
|
||
| private static Object lockHelper = new Object(); | ||
|
|
||
| private static Map<Address, TCPClient> tcpClientCache = new ConcurrentHashMap(); | ||
| public static Object request(String serviceName, Request request, Class returnType) throws Exception { | ||
|
|
||
| private static Object lockHelper = new Object(); | ||
| Address addr = LoadBalance.loadbalanceRandom(serviceName); | ||
|
|
||
| public static Object request(String serviceName, Request request, Class returnType) throws Exception { | ||
| byte[] requestBytes = CompressUtil.compress(HessianUtil.serialize(request)); | ||
|
|
||
| Address addr = LoadBalance.loadbalanceRandom(serviceName); | ||
| byte[] requestBytes = CompressUtil.compress(HessianUtil.serialize(request)); | ||
| TCPClient tcpClient = getTCPClient(addr, ServiceParams.getService(serviceName).getTimeout()); | ||
|
|
||
| TCPClient tcpClient = getTCPClient(addr,KRPC.getService(serviceName).getTimeout()); | ||
| log.debug("客户端发送数据:{}", requestBytes.length); | ||
| Integer sessionID = tcpClient.sendMsg(requestBytes); | ||
| if (Objects.isNull(sessionID)) { | ||
| throw new Exception("send data error!"); | ||
| } | ||
|
|
||
| log.debug("客户端发送数据:{}" , requestBytes.length); | ||
| Integer sessionID = tcpClient.sendMsg(requestBytes); | ||
| if(Objects.isNull(sessionID)) { | ||
| throw new Exception("send data error!"); | ||
| } | ||
|
|
||
| byte[] responseBytessrc = tcpClient.getData(sessionID); | ||
| return HessianUtil.deserialize( CompressUtil.uncompress(responseBytessrc), null); | ||
| } | ||
| byte[] responseBytessrc = tcpClient.getData(sessionID); | ||
| return HessianUtil.deserialize(CompressUtil.uncompress(responseBytessrc), null); | ||
| } | ||
|
|
||
| private static TCPClient getTCPClient(Address address,Integer timeout) throws IOException { | ||
| TCPClient tcpClient= tcpClientCache.get(address); | ||
| if (Objects.isNull(tcpClient)) { | ||
| private static TCPClient getTCPClient(Address address, Integer timeout) throws IOException { | ||
| TCPClient tcpClient = tcpClientCache.get(address); | ||
| if (Objects.isNull(tcpClient)) { | ||
|
|
||
| synchronized (lockHelper) { | ||
| tcpClient = tcpClientCache.get(address); | ||
| if (Objects.isNull(tcpClient)) { | ||
| tcpClient = new TCPClient(address.getHost(), address.getPort(),timeout); | ||
| tcpClientCache.put(address, tcpClient); | ||
| } | ||
| synchronized (lockHelper) { | ||
| tcpClient = tcpClientCache.get(address); | ||
| if (Objects.isNull(tcpClient)) { | ||
| tcpClient = new TCPClient(address.getHost(), address.getPort(), timeout); | ||
| tcpClientCache.put(address, tcpClient); | ||
| } | ||
|
|
||
| } | ||
| } | ||
|
|
||
| } | ||
| } | ||
|
|
||
| return tcpClient; | ||
| } | ||
| return tcpClient; | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.