Skip to content

B
byit-myth-job
Commit 5e13875d by liyuan
Options

修改

parent 74f3a090
Showing with 154 additions and 74 deletions
byit-myth-register/myth-register-client/src/main/java/com/byit/registry/client/RegistryBaseClient.java
......@@ -181,7 +181,7 @@ public class RegistryBaseClient {
registryParamVO.setAccessToken(this.accessToken);
registryParamVO.setBiz(this.biz);
registryParamVO.setEnv(this.env);
registryParamVO.setKeys(new ArrayList<String>(keys));
registryParamVO.setKeys(new ArrayList<>(keys));
String paramsJson = BasicJson.toJson(registryParamVO);
......
byit-myth-register/myth-register-server/src/main/java/com/byit/registry/admin/service/impl/RegistryServiceImpl.java
......@@ -259,14 +259,14 @@ public class RegistryServiceImpl implements IRegistryService, InitializingBean,
}
}
Map<String, List<String>> result = new HashMap<String, List<String>>();
Map<String, List<String>> result = new HashMap<>(16);
for (String key: keys) {
RegistryData registryData = new RegistryData();
registryData.setBiz(biz);
registryData.setEnv(env);
registryData.setKey(key);
List<String> dataList = new ArrayList<String>();
List<String> dataList = new ArrayList<>();
Registry fileRegistry = getFileRegistryData(registryData);
if (fileRegistry!=null) {
dataList = fileRegistry.getDataList();
......@@ -482,7 +482,7 @@ public class RegistryServiceImpl implements IRegistryService, InitializingBean,
public void run() {
while (!executorStoped) {
try {
// new message, filter readed
// new message, filters readed
List<RegistryMessage> messageList = registryMessageDao.findMessage(readedMessageIds);
if (messageList!=null && messageList.size()>0) {
for (RegistryMessage message: messageList) {
......
byit-myth-register/myth-register-server/src/main/resources/application.properties
### web
server.port=8080
server.context-path=/myth-register
logging.config=classpath:logback.xml
### resources
spring.mvc.static-path-pattern=/static/**
......
byit-myth-rpc/src/main/java/com/byit/rpc/registry/ServiceRegistry.java
......@@ -30,11 +30,11 @@ public abstract class ServiceRegistry {
/**
* registry service, for mult
*
* @param keys service key
* @param value service value/ip:port
* @return
* @param keyServerAddress Map's key is service key Map'svalue is service value/ip:port
*/
public abstract boolean registry(Set<String> keys, String value);
public abstract boolean registry(Map<String,String> keyServerAddress);
/**
......
byit-myth-rpc/src/main/java/com/byit/rpc/registry/impl/LocalServiceRegistry.java
package com.byit.rpc.registry.impl;
import com.byit.registry.client.model.RegistryDataParamVO;
import com.byit.rpc.registry.ServiceRegistry;
import org.springframework.util.CollectionUtils;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.*;
/**
* service registry for "local"
......@@ -33,20 +32,22 @@ public class LocalServiceRegistry extends ServiceRegistry {
registryData.clear();
}
@Override
public boolean registry(Set<String> keys, String value) {
if (keys==null || keys.size()==0 || value==null || value.trim().length()==0) {
public boolean registry(Map<String,String> keyServerAddress){
if (CollectionUtils.isEmpty(keyServerAddress)) {
return false;
}
for (String key : keys) {
TreeSet<String> values = registryData.get(key);
// init
for (Map.Entry<String,String> entry : keyServerAddress.entrySet()) {
TreeSet<String> values = registryData.get(entry.getKey());
if (values == null) {
values = new TreeSet<>();
registryData.put(key, values);
registryData.put(entry.getKey(), values);
}
values.add(value);
values.add(entry.getValue());
}
return true;
}
......
byit-myth-rpc/src/main/java/com/byit/rpc/registry/impl/RegistryServiceRegistry.java
......@@ -3,6 +3,7 @@ package com.byit.rpc.registry.impl;
import com.byit.registry.client.RegistryClient;
import com.byit.registry.client.model.RegistryDataParamVO;
import com.byit.rpc.registry.ServiceRegistry;
import org.springframework.util.CollectionUtils;
import java.util.*;
......@@ -43,16 +44,17 @@ public class RegistryServiceRegistry extends ServiceRegistry {
}
}
@Override
public boolean registry(Set<String> keys, String value) {
if (keys==null || keys.size() == 0 || value == null) {
public boolean registry(Map<String,String> keyServerAddress){
if (CollectionUtils.isEmpty(keyServerAddress)) {
return false;
}
// init
List<RegistryDataParamVO> registryDataList = new ArrayList<>();
for (String key:keys) {
registryDataList.add(new RegistryDataParamVO(key, value));
for (Map.Entry<String,String> entry : keyServerAddress.entrySet()) {
registryDataList.add(new RegistryDataParamVO(entry.getKey(), entry.getValue()));
}
return registryClient.registry(registryDataList);
......
byit-myth-rpc/src/main/java/com/byit/rpc/registry/impl/ZkServiceRegistry.java
package com.byit.rpc.registry.impl;
import com.byit.registry.client.model.RegistryDataParamVO;
import com.byit.rpc.registry.ServiceRegistry;
import com.byit.rpc.util.RpcException;
import com.byit.rpc.util.ZkClient;
......@@ -7,6 +8,7 @@ import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
......@@ -232,25 +234,37 @@ public class ZkServiceRegistry extends ServiceRegistry {
}
}
@Override
public boolean registry(Set<String> keys, String value) {
for (String key : keys) {
public boolean registry(Map<String,String> keyServerAddress){
if (CollectionUtils.isEmpty(keyServerAddress)) {
return false;
}
StringBuilder keys = new StringBuilder();
StringBuilder addressValues = new StringBuilder();
// init
for (Map.Entry<String,String> entry : keyServerAddress.entrySet()) {
// local cache
TreeSet<String> values = registryData.get(key);
TreeSet<String> values = registryData.get(entry.getKey());
if (values == null) {
values = new TreeSet<>();
registryData.put(key, values);
registryData.put(entry.getKey(), values);
}
values.add(value);
values.add(entry.getValue());
// make path, child path
String path = keyToPath(key);
zkClient.setChildPathData(path, value, "");
String path = keyToPath(entry.getKey());
keys.append(entry.getKey() + ",");
addressValues.append(entry.getValue() + ",");
zkClient.setChildPathData(path, entry.getValue(), "");
}
logger.info(">>>>>>>>>>> myth-rpc, registry success, keys = {}, value = {}", keys, value);
logger.info(">>>>>>>>>>> myth-rpc, registry success, keys = {}, value = {}", keys, addressValues);
return true;
}
@Override
public boolean remove(Set<String> keys, String value) {
for (String key : keys) {
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/reference/RpcReferenceBean.java
......@@ -147,7 +147,7 @@ public class RpcReferenceBean {
Class<?>[] parameterTypes = method.getParameterTypes();
Object[] parameters = args;
// filter for generic
// filters for generic
if (className.equals(RpcGenericService.class.getName()) && methodName.equals("invoke")) {
Class<?>[] paramTypes = null;
......@@ -168,7 +168,7 @@ public class RpcReferenceBean {
parameters = (Object[]) args[4];
}
// filter method like "Object.toString()"
// filters method like "Object.toString()"
if (className.equals(Object.class.getName())) {
logger.info(">>>>>>>>>>> myth-rpc proxy class-method not support [{}#{}]", className, methodName);
throw new RpcException("myth-rpc proxy class-method not support");
......@@ -187,7 +187,7 @@ public class RpcReferenceBean {
} else if (addressSet.size()==1) {
finalAddress = addressSet.first();
} else {
finalAddress = loadBalance.xxlRpcInvokerRouter.route(serviceKey, addressSet);
finalAddress = loadBalance.rpcInvokerRouter.route(serviceKey, addressSet);
}
}
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/route/LoadBalance.java
......@@ -11,10 +11,10 @@ public enum LoadBalance {
CONSISTENT_HASH(new RpcLoadBalanceConsistentHashStrategy());
public final RpcLoadBalance xxlRpcInvokerRouter;
public final RpcLoadBalance rpcInvokerRouter;
private LoadBalance(RpcLoadBalance xxlRpcInvokerRouter) {
this.xxlRpcInvokerRouter = xxlRpcInvokerRouter;
private LoadBalance(RpcLoadBalance rpcInvokerRouter) {
this.rpcInvokerRouter = rpcInvokerRouter;
}
......@@ -26,6 +26,4 @@ public enum LoadBalance {
}
return defaultRouter;
}
}
\ No newline at end of file
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/net/impl/mina/client/MinaClientHandler.java
......@@ -26,7 +26,7 @@ public class MinaClientHandler extends IoHandlerAdapter {
public void messageReceived(IoSession session, Object message) throws Exception {
RpcResponse xxlRpcResponse = (RpcResponse) message;
// filter beat
// filters beat
if (Beat.BEAT_ID.equalsIgnoreCase(xxlRpcResponse.getRequestId())){
return;
}
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/net/impl/mina/server/MinaServerHandler.java
......@@ -37,7 +37,7 @@ public class MinaServerHandler extends IoHandlerAdapter {
// request
final RpcRequest rpcRequest = (RpcRequest) message;
// filter beat
// filters beat
if (Beat.BEAT_ID.equalsIgnoreCase(rpcRequest.getRequestId())){
return;
}
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/net/impl/netty/server/NettyServerHandler.java
......@@ -33,7 +33,7 @@ public class NettyServerHandler extends SimpleChannelInboundHandler<RpcRequest>
@Override
public void channelRead0(final ChannelHandlerContext ctx, final RpcRequest rpcRequest) throws Exception {
// filter beat
// filters beat
if (Beat.BEAT_ID.equalsIgnoreCase(rpcRequest.getRequestId())){
logger.debug(">>>>>>>>>>> myth-rpc provider netty server read beat-ping.");
return;
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/net/impl/netty_http/server/NettyHttpServerHandler.java
......@@ -80,7 +80,7 @@ public class NettyHttpServerHandler extends SimpleChannelInboundHandler<FullHttp
RpcRequest rpcRequest = (RpcRequest) rpcProviderFactory.getSerializer().deserialize(requestBytes, RpcRequest.class);
requestId = rpcRequest.getRequestId();
// filter beat
// filters beat
if (Beat.BEAT_ID.equalsIgnoreCase(rpcRequest.getRequestId())){
logger.debug(">>>>>>>>>>> myth-rpc provider netty_http server read beat-ping.");
return;
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/provider/RpcProviderFactory.java
......@@ -20,7 +20,6 @@ import java.util.Map;
/**
* provider
*
*/
public class RpcProviderFactory {
private static final Logger logger = LoggerFactory.getLogger(RpcProviderFactory.class);
......@@ -36,6 +35,12 @@ public class RpcProviderFactory {
private String ip; // for registry
private int port; // default port
private String accessToken;
private int serverPort;
private String serverName;
/**
* 该服务是否提供http服务
*/
private Boolean isHttp;
private Class<? extends ServiceRegistry> serviceRegistryClass;
private Map<String, String> serviceRegistryParam;
......@@ -43,12 +48,16 @@ public class RpcProviderFactory {
public RpcProviderFactory() {
}
public void initConfig(NetEnum netType,
Serializer serializer,
int corePoolSize,
int maxPoolSize,
String ip,
int port,
boolean isHttp,
int serverPort,
String serverName,
String accessToken,
Class<? extends ServiceRegistry> serviceRegistryClass,
Map<String, String> serviceRegistryParam) {
......@@ -60,18 +69,21 @@ public class RpcProviderFactory {
this.maxPoolSize = maxPoolSize;
this.ip = ip;
this.port = port;
this.isHttp = isHttp;
this.serverPort = serverPort;
this.serverName = serverName;
this.accessToken = accessToken;
this.serviceRegistryClass = serviceRegistryClass;
this.serviceRegistryParam = serviceRegistryParam;
// valid
if (this.netType==null) {
if (this.netType == null) {
throw new RpcException("myth-rpc provider netType missing.");
}
if (this.serializer==null) {
if (this.serializer == null) {
throw new RpcException("myth-rpc provider serializer missing.");
}
if (!(this.corePoolSize>=0 && this.maxPoolSize>0 && this.maxPoolSize>=this.corePoolSize)) {
if (!(this.corePoolSize >= 0 && this.maxPoolSize > 0 && this.maxPoolSize >= this.corePoolSize)) {
this.corePoolSize = 60;
this.maxPoolSize = 300;
}
......@@ -82,7 +94,7 @@ public class RpcProviderFactory {
this.port = 7080;
}
if (NetUtil.isPortUsed(this.port)) {
throw new RpcException("myth-rpc provider port["+ this.port +"] is used.");
throw new RpcException("myth-rpc provider port[" + this.port + "] is used.");
}
if (this.serviceRegistryClass != null) {
if (this.serviceRegistryParam == null) {
......@@ -113,10 +125,15 @@ public class RpcProviderFactory {
private Server server;
private ServiceRegistry serviceRegistry;
private String serviceAddress;
private String httpServiceAddress;
public void start() throws Exception {
// start server
if (isHttp) {
httpServiceAddress = IpUtil.getIpPort(this.ip,this.serverPort);
}
serviceAddress = IpUtil.getIpPort(this.ip, port);
server = netType.serverClass.newInstance();
server.setStartedCallback(new BaseCallback() { // serviceRegistry started
@Override
......@@ -126,7 +143,17 @@ public class RpcProviderFactory {
serviceRegistry = serviceRegistryClass.newInstance();
serviceRegistry.start(serviceRegistryParam);
if (serviceData.size() > 0) {
serviceRegistry.registry(serviceData.keySet(), serviceAddress);
Map<String, String> keyServerAddress = new HashMap<>(16);
for (Map.Entry<String, Object> entry : serviceData.entrySet()) {
//提供http请求的是服务的id，不包含.
if(entry.getKey().contains(".")){
keyServerAddress.put(entry.getKey(), serviceAddress);
}else {
keyServerAddress.put(entry.getKey(),httpServiceAddress);
}
}
serviceRegistry.registry(keyServerAddress);
}
}
}
......@@ -158,7 +185,8 @@ public class RpcProviderFactory {
/**
* init local rpc service map
*/
private Map<String, Object> serviceData = new HashMap<String, Object>();
private Map<String, Object> serviceData = new HashMap<>();
public Map<String, Object> getServiceData() {
return serviceData;
}
......@@ -170,9 +198,9 @@ public class RpcProviderFactory {
* @param version
* @return
*/
public static String makeServiceKey(String iface, String version){
public static String makeServiceKey(String iface, String version) {
String serviceKey = iface;
if (version!=null && version.trim().length()>0) {
if (version != null && version.trim().length() > 0) {
serviceKey += "#".concat(version);
}
return serviceKey;
......@@ -185,7 +213,7 @@ public class RpcProviderFactory {
* @param version
* @param serviceBean
*/
public void addService(String iface, String version, Object serviceBean){
public void addService(String iface, String version, Object serviceBean) {
String serviceKey = makeServiceKey(iface, version);
serviceData.put(serviceKey, serviceBean);
......@@ -210,15 +238,15 @@ public class RpcProviderFactory {
// valid
if (serviceBean == null) {
rpcResponse.setErrorMsg("The serviceKey["+ serviceKey +"] not found.");
rpcResponse.setErrorMsg("The serviceKey[" + serviceKey + "] not found.");
return rpcResponse;
}
if (System.currentTimeMillis() - rpcRequest.getCreateMillisTime() > 3*60*1000) {
if (System.currentTimeMillis() - rpcRequest.getCreateMillisTime() > 3 * 60 * 1000) {
rpcResponse.setErrorMsg("The timestamp difference between admin and executor exceeds the limit.");
return rpcResponse;
}
if (accessToken!=null && accessToken.trim().length()>0 && !accessToken.trim().equals(rpcRequest.getAccessToken())) {
if (accessToken != null && accessToken.trim().length() > 0 && !accessToken.trim().equals(rpcRequest.getAccessToken())) {
rpcResponse.setErrorMsg("The access token[" + rpcRequest.getAccessToken() + "] is wrong.");
return rpcResponse;
}
......@@ -234,9 +262,6 @@ public class RpcProviderFactory {
method.setAccessible(true);
Object result = method.invoke(serviceBean, parameters);
/*FastClass serviceFastClass = FastClass.create(serviceClass);
FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);
Object result = serviceFastMethod.invoke(serviceBean, parameters);*/
rpcResponse.setResult(result);
} catch (Throwable t) {
......
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/provider/annotation/RpcService.java
......@@ -11,9 +11,14 @@ import java.lang.annotation.*;
@Inherited
public @interface RpcService {
String version() default "";
/**
* @return
* 提供http服务
* 向注册中心注册 服务的id 同一个业务中必须唯一
* 端口号从配置文件中获取springboot的端口号
*/
String version() default "";
boolean http_type() default false;
}
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/provider/impl/RpcSpringProviderFactory.java
......@@ -9,6 +9,7 @@ import com.byit.rpc.util.RpcException;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
......@@ -16,9 +17,8 @@ import java.util.Map;
/**
* myth-rpc provider (for spring)
*
*/
public class RpcSpringProviderFactory extends RpcProviderFactory implements ApplicationContextAware, InitializingBean,DisposableBean {
public class RpcSpringProviderFactory extends RpcProviderFactory implements ApplicationContextAware, InitializingBean, DisposableBean {
// ---------------------- config ----------------------
......@@ -35,6 +35,15 @@ public class RpcSpringProviderFactory extends RpcProviderFactory implements Appl
private Class<? extends ServiceRegistry> serviceRegistryClass; // class.forname
private Map<String, String> serviceRegistryParam;
@Value("${server.port}")
private int serverPort;
@Value("${spring.application.name}")
private String serverName;
/**
* 该服务是否提供http服务
*/
private Boolean isHttp = false;
// set
public void setNetType(String netType) {
......@@ -85,15 +94,16 @@ public class RpcSpringProviderFactory extends RpcProviderFactory implements Appl
// util
private void prepareConfig(){
private void prepareConfig() {
// prepare config
NetEnum netTypeEnum = NetEnum.autoMatch(netType, null);
Serializer.SerializeEnum serializeEnum = Serializer.SerializeEnum.match(serialize, null);
Serializer serializer = serializeEnum!=null?serializeEnum.getSerializer():null;
Serializer serializer = serializeEnum != null ? serializeEnum.getSerializer() : null;
// init config
super.initConfig(netTypeEnum, serializer, corePoolSize, maxPoolSize, ip, port, accessToken, serviceRegistryClass, serviceRegistryParam);
super.initConfig(netTypeEnum, serializer, corePoolSize, maxPoolSize, ip, port, isHttp, serverPort, serverName
, accessToken, serviceRegistryClass, serviceRegistryParam);
}
......@@ -103,23 +113,27 @@ public class RpcSpringProviderFactory extends RpcProviderFactory implements Appl
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(RpcService.class);
if (serviceBeanMap!=null && serviceBeanMap.size()>0) {
if (serviceBeanMap != null && serviceBeanMap.size() > 0) {
for (Object serviceBean : serviceBeanMap.values()) {
// valid
if (serviceBean.getClass().getInterfaces().length ==0) {
if (serviceBean.getClass().getInterfaces().length == 0) {
throw new RpcException("myth-rpc, service(RpcService) must inherit interface.");
}
// add service
RpcService xxlRpcService = serviceBean.getClass().getAnnotation(RpcService.class);
RpcService rpcService = serviceBean.getClass().getAnnotation(RpcService.class);
String iface = serviceBean.getClass().getInterfaces()[0].getName();
String version = xxlRpcService.version();
String version = rpcService.version();
//如果提供的是htp服务
if (rpcService.http_type()) {
this.isHttp = true;
super.addService(serverName, version, serviceBean);
} else {
super.addService(iface, version, serviceBean);
}
}
// TODO，addServices by api + prop
}
}
}
......
demo-client/byit-demo-service/src/main/resources/application-dev.properties 0 → 100644
# myth-rpc
myth-rpc.remoting.port=9998
myth-rpc.registry.address=http://localhost:8080/myth-register
myth-rpc.registry.env=liyuan
myth-rpc.registry.biz=byit-myth-job
logging.config=classpath:logback.xml
\ No newline at end of file
demo-client/byit-demo-service/src/main/resources/application-pre.properties 0 → 100644
# myth-rpc
myth-rpc.remoting.port=9998
myth-rpc.registry.address=http://localhost:8080/myth-register
myth-rpc.registry.env=pre
myth-rpc.registry.biz=byit-myth-job
logging.config=classpath:logback.xml
\ No newline at end of file
demo-client/byit-demo-service/src/main/resources/application-test.properties 0 → 100644
# myth-rpc
myth-rpc.remoting.port=9998
myth-rpc.registry.address=http://localhost:8080/myth-register
myth-rpc.registry.env=test
myth-rpc.registry.biz=byit-myth-job
logging.config=classpath:logback.xml
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment