natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add rpc config

This commit is contained in:
CalvinKirs 2021-01-25 21:40:12 +08:00
parent a2d245ddc1
commit 9f58eb7ea5
33 changed files with 301 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/config/NettyServerConfig.java
View File

@ -56,7 +56,7 @@ public class NettyServerConfig {
/**
* listen port
*/
private int listenPort = 12636;
private int listenPort = 12346;
public int getListenPort() {
return listenPort;

1
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/decoder/NettyEncoder.java
View File

@ -4,7 +4,6 @@ import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.remote.serialize.ProtoStuffUtils;
/**

12
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/Invoker.java
View File

@ -1,12 +0,0 @@
package org.apache.dolphinscheduler.remote.rpc;
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest;
/**
* Invoker
*/
public interface Invoker {
RpcResponse invoke(RpcRequest req) throws Throwable;
}

15
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/MainServerTest.java
View File

@ -1,15 +0,0 @@
package org.apache.dolphinscheduler.remote.rpc;
import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
import org.apache.dolphinscheduler.remote.rpc.remote.NettyServer;
/**
* @author jiangli
* @date 2021-01-20 14:54
*/
public class MainServerTest {
public static void main(String[] args) {
NettyServer nettyServer = new NettyServer(new NettyServerConfig());
}
}

39
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/MainTest.java
View File

@ -1,39 +0,0 @@
package org.apache.dolphinscheduler.remote.rpc;
import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
import org.apache.dolphinscheduler.remote.rpc.client.IRpcClient;
import org.apache.dolphinscheduler.remote.rpc.client.RpcClient;
import org.apache.dolphinscheduler.remote.rpc.remote.NettyServer;
import org.apache.dolphinscheduler.remote.utils.Host;
/**
* @author jiangli
* @date 2021-01-11 21:06
*/
public class MainTest {
public static void main(String[] args) throws Exception {
// NettyServer nettyServer = new NettyServer(new NettyServerConfig());
// NettyClient nettyClient=new NettyClient(new NettyClientConfig());
Host host = new Host("127.0.0.1", 12636);
IRpcClient rpcClient = new RpcClient();
IUserService userService = rpcClient.create(IUserService.class, host);
boolean result = userService.say("calvin");
System.out.println("异步回掉成功" + result);
System.out.println(userService.hi(10));
System.out.println(userService.hi(188888888));
IUserService user = rpcClient.create(IUserService.class, host);
System.out.println(user.hi(99999));
System.out.println(user.hi(998888888));
System.out.println(IUserService.class.getSimpleName());
System.out.println(UserService.class.getSimpleName());
}
}

15
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/UserCallback.java
View File

@ -1,15 +0,0 @@
package org.apache.dolphinscheduler.remote.rpc;
import org.apache.dolphinscheduler.remote.rpc.common.AbstractRpcCallBack;
/**
* @author jiangli
* @date 2021-01-15 07:32
*/
public class UserCallback extends AbstractRpcCallBack {
@Override
public void run(Object object) {
Boolean msg= (Boolean) object;
System.out.println("我是异步回调handle Kris"+msg);
}
}

23
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/UserService.java
View File

@ -1,23 +0,0 @@
package org.apache.dolphinscheduler.remote.rpc;
import org.apache.dolphinscheduler.remote.rpc.base.Rpc;
import org.apache.dolphinscheduler.remote.rpc.base.RpcService;
/**
* @author jiangli
* @date 2021-01-11 21:05
*/
@RpcService("IUserService")
public class UserService implements IUserService{
@Rpc(async = true, serviceCallback = UserCallback.class, retries = 9999)
@Override
public Boolean say(String s) {
return true;
}
@Override
public String hi(int num) {
return "this world has " + num + "sun";
}
}

4
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/base/Rpc.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/base/Rpc.java
View File

@ -15,9 +15,9 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.base;
package org.apache.dolphinscheduler.rpc.base;
import org.apache.dolphinscheduler.remote.rpc.common.AbstractRpcCallBack;
import org.apache.dolphinscheduler.rpc.common.AbstractRpcCallBack;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/base/RpcService.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/base/RpcService.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.base;
package org.apache.dolphinscheduler.rpc.base;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;

10
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/client/ConsumerConfig.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/client/ConsumerConfig.java
View File

@ -15,10 +15,10 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.client;
package org.apache.dolphinscheduler.rpc.client;
import org.apache.dolphinscheduler.remote.rpc.common.AbstractRpcCallBack;
import org.apache.dolphinscheduler.remote.rpc.common.ConsumerConfigConstants;
import org.apache.dolphinscheduler.rpc.common.AbstractRpcCallBack;
import org.apache.dolphinscheduler.rpc.common.ConsumerConfigConstants;
/**
* We will cache the consumer configuration, when the rpc call is generated, the consumer configuration will be first obtained from here
@ -39,7 +39,7 @@ public class ConsumerConfig {
return serviceCallBackClass;
}
public void setServiceCallBackClass(Class<? extends AbstractRpcCallBack> serviceCallBackClass) {
void setServiceCallBackClass(Class<? extends AbstractRpcCallBack> serviceCallBackClass) {
this.serviceCallBackClass = serviceCallBackClass;
}
@ -47,7 +47,7 @@ public class ConsumerConfig {
return ackCallBackClass;
}
public void setAckCallBackClass(Class<? extends AbstractRpcCallBack> ackCallBackClass) {
void setAckCallBackClass(Class<? extends AbstractRpcCallBack> ackCallBackClass) {
this.ackCallBackClass = ackCallBackClass;
}

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/client/ConsumerConfigCache.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/client/ConsumerConfigCache.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.client;
package org.apache.dolphinscheduler.rpc.client;
import java.util.concurrent.ConcurrentHashMap;

10
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/client/ConsumerInterceptor.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/client/ConsumerInterceptor.java
View File

@ -15,14 +15,14 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.client;
package org.apache.dolphinscheduler.rpc.client;
import org.apache.dolphinscheduler.remote.exceptions.RemotingException;
import org.apache.dolphinscheduler.remote.rpc.base.Rpc;
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.remote.rpc.remote.NettyClient;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.rpc.base.Rpc;
import org.apache.dolphinscheduler.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.rpc.remote.NettyClient;
import java.lang.reflect.Method;
import java.util.UUID;

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/client/IRpcClient.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/client/IRpcClient.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.client;
package org.apache.dolphinscheduler.rpc.client;
import org.apache.dolphinscheduler.remote.utils.Host;

14
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/client/RpcClient.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/client/RpcClient.java
View File

@ -15,14 +15,12 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.client;
package org.apache.dolphinscheduler.rpc.client;
import static net.bytebuddy.matcher.ElementMatchers.isDeclaredBy;
import org.apache.dolphinscheduler.remote.utils.Host;
import java.util.concurrent.ConcurrentHashMap;
import net.bytebuddy.ByteBuddy;
import net.bytebuddy.implementation.MethodDelegation;
@ -31,22 +29,14 @@ import net.bytebuddy.implementation.MethodDelegation;
*/
public class RpcClient implements IRpcClient {
private ConcurrentHashMap<String, Object> classMap = new ConcurrentHashMap<>();
@Override
public <T> T create(Class<T> clazz, Host host) throws Exception {
// if(!classMap.containsKey(clazz.getName())){
T proxy = new ByteBuddy()
return new ByteBuddy()
.subclass(clazz)
.method(isDeclaredBy(clazz)).intercept(MethodDelegation.to(new ConsumerInterceptor(host)))
.make()
.load(getClass().getClassLoader())
.getLoaded()
.getDeclaredConstructor().newInstance();
// classMap.putIfAbsent(clazz.getName(),proxy);
return proxy;
// }
// return (T) classMap.get(clazz.getName());
}
}

4
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/client/RpcRequestCache.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/client/RpcRequestCache.java
View File

@ -15,9 +15,9 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.client;
package org.apache.dolphinscheduler.rpc.client;
import org.apache.dolphinscheduler.remote.rpc.future.RpcFuture;
import org.apache.dolphinscheduler.rpc.future.RpcFuture;
/**
* RpcRequestCache

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/client/RpcRequestTable.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/client/RpcRequestTable.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.client;
package org.apache.dolphinscheduler.rpc.client;
import java.util.concurrent.ConcurrentHashMap;

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/common/AbstractRpcCallBack.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/common/AbstractRpcCallBack.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.common;
package org.apache.dolphinscheduler.rpc.common;
/**
* AbstractRpcCallBack

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/common/ConsumerConfigConstants.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/common/ConsumerConfigConstants.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.common;
package org.apache.dolphinscheduler.rpc.common;
/**
* ConsumerConfigConstants

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/common/RequestEventType.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/common/RequestEventType.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.common;
package org.apache.dolphinscheduler.rpc.common;
public enum RequestEventType {

3
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/common/ResponseEventType.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/common/ResponseEventType.java
View File

@ -1,4 +1,4 @@
package org.apache.dolphinscheduler.remote.rpc.common;/*
package org.apache.dolphinscheduler.rpc.common;/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
@ -29,7 +29,6 @@ public enum ResponseEventType {
this.description = description;
}
public Byte getType() {
return type;
}

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/common/RpcRequest.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/common/RpcRequest.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.common;
package org.apache.dolphinscheduler.rpc.common;
/**
* RpcRequest

2
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/common/RpcResponse.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/common/RpcResponse.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.common;
package org.apache.dolphinscheduler.rpc.common;
/**
* RpcResponse

45
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/common/ThreadPoolManager.java Normal file
View File

@ -0,0 +1,45 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.rpc.common;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.DiscardPolicy;
import java.util.concurrent.TimeUnit;
public enum ThreadPoolManager {
INSTANCE;
ExecutorService executorService;
ThreadPoolManager() {
int SIZE_WORK_QUEUE = 200;
long KEEP_ALIVE_TIME = 60;
int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
int MAXI_MUM_POOL_SIZE = CORE_POOL_SIZE * 4;
executorService = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXI_MUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(SIZE_WORK_QUEUE),
new DiscardPolicy());
}
public void addExecuteTask(Runnable task) {
executorService.submit(task);
}
}

19
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/config/ServiceBean.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/config/ServiceBean.java
View File

@ -15,16 +15,14 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.config;
package org.apache.dolphinscheduler.rpc.config;
import org.apache.dolphinscheduler.remote.rpc.IUserService;
import org.apache.dolphinscheduler.remote.rpc.base.RpcService;
import org.apache.dolphinscheduler.rpc.base.RpcService;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.concurrent.atomic.AtomicBoolean;
@ -32,7 +30,9 @@ import org.reflections.Reflections;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* ServiceBean find all rpcService
*/
public class ServiceBean {
private static final Logger logger = LoggerFactory.getLogger(ServiceBean.class);
@ -42,9 +42,8 @@ public class ServiceBean {
private static AtomicBoolean initialized = new AtomicBoolean(false);
private static synchronized void init() {
Reflections f = new Reflections("org/apache/dolphinscheduler/remote/rpc");
// todo config
Reflections f = new Reflections("org/apache/dolphinscheduler/rpc");
List<Class<?>> list = new ArrayList<>(f.getTypesAnnotatedWith(RpcService.class));
list.forEach(rpcClass -> {
RpcService rpcService = rpcClass.getAnnotation(RpcService.class);
@ -52,10 +51,6 @@ public class ServiceBean {
});
}
public static void main(String[] args) {
init();
}
public static Class getServiceClass(String className) {
if (initialized.get()) {
return (Class) serviceMap.get(className);

4
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/future/RpcFuture.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/future/RpcFuture.java
View File

@ -15,9 +15,9 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.future;
package org.apache.dolphinscheduler.rpc.future;
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.rpc.common.RpcResponse;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;

63
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/remote/NettyClient.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/remote/NettyClient.java
View File

@ -15,16 +15,16 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.remote;
package org.apache.dolphinscheduler.rpc.remote;
import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
import org.apache.dolphinscheduler.remote.decoder.NettyDecoder;
import org.apache.dolphinscheduler.remote.decoder.NettyEncoder;
import org.apache.dolphinscheduler.remote.rpc.client.RpcRequestCache;
import org.apache.dolphinscheduler.remote.rpc.client.RpcRequestTable;
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.remote.rpc.future.RpcFuture;
import org.apache.dolphinscheduler.rpc.client.RpcRequestCache;
import org.apache.dolphinscheduler.rpc.client.RpcRequestTable;
import org.apache.dolphinscheduler.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.rpc.future.RpcFuture;
import org.apache.dolphinscheduler.remote.utils.Constants;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.remote.utils.NettyUtils;
@ -98,7 +98,7 @@ public class NettyClient {
/**
* get channel
*/
public Channel getChannel(Host host) {
private Channel getChannel(Host host) {
Channel channel = channels.get(host);
if (channel != null && channel.isActive()) {
return channel;
@ -138,7 +138,7 @@ public class NettyClient {
*
* @param clientConfig client config
*/
public NettyClient(final NettyClientConfig clientConfig) {
private NettyClient(final NettyClientConfig clientConfig) {
this.clientConfig = clientConfig;
if (NettyUtils.useEpoll()) {
this.workerGroup = new EpollEventLoopGroup(clientConfig.getWorkerThreads(), new ThreadFactory() {
@ -169,24 +169,24 @@ public class NettyClient {
private void start() {
this.bootstrap
.group(this.workerGroup)
.channel(NettyUtils.getSocketChannelClass())
.option(ChannelOption.SO_KEEPALIVE, clientConfig.isSoKeepalive())
.option(ChannelOption.TCP_NODELAY, clientConfig.isTcpNoDelay())
.option(ChannelOption.SO_SNDBUF, clientConfig.getSendBufferSize())
.option(ChannelOption.SO_RCVBUF, clientConfig.getReceiveBufferSize())
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfig.getConnectTimeoutMillis())
.handler(new LoggingHandler(LogLevel.DEBUG))
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline()
.addLast(new NettyEncoder(RpcRequest.class))
.addLast(new NettyDecoder(RpcResponse.class))
.addLast("client-idle-handler", new IdleStateHandler(Constants.NETTY_CLIENT_HEART_BEAT_TIME, 0, 0, TimeUnit.MILLISECONDS))
.addLast(new NettyClientHandler());
}
});
.group(this.workerGroup)
.channel(NettyUtils.getSocketChannelClass())
.option(ChannelOption.SO_KEEPALIVE, clientConfig.isSoKeepalive())
.option(ChannelOption.TCP_NODELAY, clientConfig.isTcpNoDelay())
.option(ChannelOption.SO_SNDBUF, clientConfig.getSendBufferSize())
.option(ChannelOption.SO_RCVBUF, clientConfig.getReceiveBufferSize())
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfig.getConnectTimeoutMillis())
.handler(new LoggingHandler(LogLevel.DEBUG))
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline()
.addLast(new NettyEncoder(RpcRequest.class))
.addLast(new NettyDecoder(RpcResponse.class))
.addLast("client-idle-handler", new IdleStateHandler(Constants.NETTY_CLIENT_HEART_BEAT_TIME, 0, 0, TimeUnit.MILLISECONDS))
.addLast(new NettyClientHandler());
}
});
isStarted.compareAndSet(false, true);
}
@ -196,9 +196,8 @@ public class NettyClient {
Channel channel = getChannel(host);
assert channel != null;
RpcRequestCache rpcRequestCache = new RpcRequestCache();
rpcRequestCache.setServiceName(request.getClassName() + request.getMethodName());
String serviceName = request.getClassName() + request.getMethodName();
rpcRequestCache.setServiceName(serviceName);
RpcFuture future = null;
if (!async) {
future = new RpcFuture();
@ -209,15 +208,15 @@ public class NettyClient {
RpcResponse result = null;
if (async) {
result=new RpcResponse();
result.setStatus((byte)0);
result = new RpcResponse();
result.setStatus((byte) 0);
result.setResult(true);
return result;
}
try {
result = future.get();
} catch (InterruptedException | ExecutionException e) {
logger.error("send msg error",e);
logger.error("send msg error，service name is {}", serviceName, e);
}
return result;
}

51
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/remote/NettyClientHandler.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/remote/NettyClientHandler.java
View File

@ -15,21 +15,16 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.remote;
package org.apache.dolphinscheduler.rpc.remote;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.FastThreadLocalThread;
import org.apache.dolphinscheduler.remote.rpc.client.ConsumerConfig;
import org.apache.dolphinscheduler.remote.rpc.client.ConsumerConfigCache;
import org.apache.dolphinscheduler.remote.rpc.client.RpcRequestCache;
import org.apache.dolphinscheduler.remote.rpc.client.RpcRequestTable;
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.remote.rpc.future.RpcFuture;
import org.apache.dolphinscheduler.rpc.client.ConsumerConfig;
import org.apache.dolphinscheduler.rpc.client.ConsumerConfigCache;
import org.apache.dolphinscheduler.rpc.client.RpcRequestCache;
import org.apache.dolphinscheduler.rpc.client.RpcRequestTable;
import org.apache.dolphinscheduler.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.rpc.common.ThreadPoolManager;
import org.apache.dolphinscheduler.rpc.future.RpcFuture;
import java.lang.reflect.InvocationTargetException;
import java.net.InetSocketAddress;
@ -37,6 +32,12 @@ import java.net.InetSocketAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.FastThreadLocalThread;
/**
* NettyClientHandler
*/
@ -46,6 +47,8 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class);
private final ThreadPoolManager threadPoolManager = ThreadPoolManager.INSTANCE;
@Override
public void channelInactive(ChannelHandlerContext ctx) {
InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
@ -58,10 +61,13 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
RpcRequestCache rpcRequest = RpcRequestTable.get(rsp.getRequestId());
if (null == rpcRequest) {
logger.warn("未知响应");
logger.warn("rpc read error,this request does not exist");
return;
}
threadPoolManager.addExecuteTask(() -> readHandler(rsp, rpcRequest));
}
private void readHandler(RpcResponse rsp, RpcRequestCache rpcRequest) {
String serviceName = rpcRequest.getServiceName();
ConsumerConfig consumerConfig = ConsumerConfigCache.getConfigByServersName(serviceName);
if (!consumerConfig.getAsync()) {
@ -71,7 +77,6 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
return;
}
//async
new FastThreadLocalThread(() -> {
try {
@ -79,18 +84,16 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
try {
consumerConfig.getServiceCallBackClass().getDeclaredConstructor().newInstance().run(rsp.getResult());
} catch (InvocationTargetException | NoSuchMethodException e) {
logger.error("rpc call back error",e);
logger.error("rpc call back error, serviceName {} ", serviceName, e);
}
} else {
logger.error("xxxx fail");
logger.error("rpc response error ,serviceName {}", serviceName);
}
} catch (InstantiationException | IllegalAccessException e) {
logger.error("execute async error", e);
logger.error("execute async error,serviceName {}", serviceName, e);
}
}).start();
}
@Override
@ -99,7 +102,7 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
RpcRequest request = new RpcRequest();
request.setEventType((byte)0);
request.setEventType((byte) 0);
ctx.channel().writeAndFlush(request);
logger.debug("send heart beat msg...");
@ -110,12 +113,8 @@ public class NettyClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
System.out.println("exceptionCaught");
logger.error("exceptionCaught : {}", cause.getMessage(), cause);
ctx.channel().close();
}
private void executeAsyncHandler(){
}
}

80
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/remote/NettyServer.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/remote/NettyServer.java
View File

@ -15,7 +15,23 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.remote;
package org.apache.dolphinscheduler.rpc.remote;
import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
import org.apache.dolphinscheduler.remote.decoder.NettyDecoder;
import org.apache.dolphinscheduler.remote.decoder.NettyEncoder;
import org.apache.dolphinscheduler.remote.utils.Constants;
import org.apache.dolphinscheduler.remote.utils.NettyUtils;
import org.apache.dolphinscheduler.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.rpc.common.RpcResponse;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
@ -29,30 +45,12 @@ import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.dolphinscheduler.remote.decoder.NettyDecoder;
import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
import org.apache.dolphinscheduler.remote.decoder.NettyEncoder;
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.remote.utils.Constants;
import org.apache.dolphinscheduler.remote.utils.NettyUtils;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* NettyServer
*/
public class NettyServer {
private static final Logger logger =LoggerFactory.getLogger(NettyServer.class);
private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
/**
* boss group
@ -126,29 +124,28 @@ public class NettyServer {
this.start();
}
/**
* server start
*/
public void start() {
if (isStarted.compareAndSet(false, true)) {
this.serverBootstrap
.group(this.bossGroup, this.workGroup)
.channel(NettyUtils.getServerSocketChannelClass())
.option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.SO_BACKLOG, serverConfig.getSoBacklog())
.childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isSoKeepalive())
.childOption(ChannelOption.TCP_NODELAY, serverConfig.isTcpNoDelay())
.childOption(ChannelOption.SO_SNDBUF, serverConfig.getSendBufferSize())
.childOption(ChannelOption.SO_RCVBUF, serverConfig.getReceiveBufferSize())
.handler(new LoggingHandler(LogLevel.DEBUG))
.childHandler(new ChannelInitializer<SocketChannel>() {
.group(this.bossGroup, this.workGroup)
.channel(NettyUtils.getServerSocketChannelClass())
.option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.SO_BACKLOG, serverConfig.getSoBacklog())
.childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isSoKeepalive())
.childOption(ChannelOption.TCP_NODELAY, serverConfig.isTcpNoDelay())
.childOption(ChannelOption.SO_SNDBUF, serverConfig.getSendBufferSize())
.childOption(ChannelOption.SO_RCVBUF, serverConfig.getReceiveBufferSize())
.handler(new LoggingHandler(LogLevel.DEBUG))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch){
initNettyChannel(ch);
}
});
@Override
protected void initChannel(SocketChannel ch) {
initNettyChannel(ch);
}
});
ChannelFuture future;
try {
@ -165,7 +162,6 @@ public class NettyServer {
throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()));
}
}
System.out.println("netty ser ver start");
}
/**
@ -175,13 +171,12 @@ public class NettyServer {
*/
private void initNettyChannel(SocketChannel ch) {
ch.pipeline()
.addLast(new NettyDecoder(RpcRequest.class))
.addLast(new NettyEncoder(RpcResponse.class))
.addLast("server-idle-handle", new IdleStateHandler(0, 0, Constants.NETTY_SERVER_HEART_BEAT_TIME, TimeUnit.MILLISECONDS))
.addLast("handler", new NettyServerHandler());
.addLast(new NettyDecoder(RpcRequest.class))
.addLast(new NettyEncoder(RpcResponse.class))
.addLast("server-idle-handle", new IdleStateHandler(0, 0, Constants.NETTY_SERVER_HEART_BEAT_TIME, TimeUnit.MILLISECONDS))
.addLast("handler", new NettyServerHandler());
}
public void close() {
if (isStarted.compareAndSet(true, false)) {
try {
@ -199,5 +194,4 @@ public class NettyServer {
}
}
}

29
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/rpc/remote/NettyServerHandler.java → dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/rpc/remote/NettyServerHandler.java
View File

@ -15,18 +15,14 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.rpc.remote;
package org.apache.dolphinscheduler.rpc.remote;
import org.apache.dolphinscheduler.remote.rpc.IUserService;
import org.apache.dolphinscheduler.remote.rpc.base.RpcService;
import org.apache.dolphinscheduler.remote.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.remote.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.remote.rpc.config.ServiceBean;
import org.apache.dolphinscheduler.rpc.common.RpcRequest;
import org.apache.dolphinscheduler.rpc.common.RpcResponse;
import org.apache.dolphinscheduler.rpc.common.ThreadPoolManager;
import org.apache.dolphinscheduler.rpc.config.ServiceBean;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.ServiceLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -42,19 +38,19 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);
private final ThreadPoolManager threadPoolManager = ThreadPoolManager.INSTANCE;
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
super.channelRegistered(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
logger.info("channel close");
ctx.channel().close();
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
logger.info("client connect success !" + ctx.channel().remoteAddress());
@ -65,19 +61,20 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
RpcRequest req = (RpcRequest) msg;
RpcResponse response = new RpcResponse();
if (req.getEventType() == 0) {
logger.info("accept heartbeat msg");
return;
}
//todo 使用业务线程池去处理 不要占用netty的资源
threadPoolManager.addExecuteTask(() -> readHandler(ctx, req));
}
private void readHandler(ChannelHandlerContext ctx, RpcRequest req) {
RpcResponse response = new RpcResponse();
response.setRequestId(req.getRequestId());
response.setStatus((byte) 0);
String classname = req.getClassName();
String methodName = req.getMethodName();
@ -95,7 +92,7 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
result = method.invoke(object, arguments);
} catch (Exception e) {
logger.error("netty server execute error", e);
logger.error("netty server execute error,service name {}", classname + methodName, e);
response.setStatus((byte) -1);
}

63
dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/rpc/RpcTest.java Normal file
View File

@ -0,0 +1,63 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.rpc;
import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.rpc.client.IRpcClient;
import org.apache.dolphinscheduler.rpc.client.RpcClient;
import org.apache.dolphinscheduler.rpc.remote.NettyClient;
import org.apache.dolphinscheduler.rpc.remote.NettyServer;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class RpcTest {
private NettyServer nettyServer;
private IUserService userService;
private Host host;
@Before
public void before() throws Exception {
nettyServer = new NettyServer(new NettyServerConfig());
IRpcClient rpcClient = new RpcClient();
host = new Host("127.0.0.1", 12346);
userService = rpcClient.create(IUserService.class, host);
}
@Test
public void sendTest() {
Integer result = userService.hi(3);
Assert.assertSame(4, result);
result = userService.hi(4);
Assert.assertSame(5, result);
userService.say("sync");
}
@After
public void after() {
NettyClient.getInstance().close();
nettyServer.close();
}
}

30
dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/rpc/UserCallback.java Normal file
View File

@ -0,0 +1,30 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.rpc;
import org.apache.dolphinscheduler.rpc.common.AbstractRpcCallBack;
/**
* UserCallback
*/
public class UserCallback extends AbstractRpcCallBack {
@Override
public void run(Object object) {
}
}

20
dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/rpc/UserService.java Normal file
View File

@ -0,0 +1,20 @@
package org.apache.dolphinscheduler.rpc;
import org.apache.dolphinscheduler.rpc.base.RpcService;
/**
* UserService
*/
@RpcService("IUserService")
public class UserService implements IUserService{
@Override
public Boolean say(String s) {
return true;
}
@Override
public Integer hi(int num) {
return ++num;
}
}

1
pom.xml
View File

@ -868,6 +868,7 @@
<include>**/dao/entity/UdfFuncTest.java</include>
<include>**/remote/JsonSerializerTest.java</include>
<include>**/remote/RemoveTaskLogResponseCommandTest.java</include>
<include>**/rpc/RpcTest.java</include>
<include>**/remote/RemoveTaskLogRequestCommandTest.java</include>
<include>**/remote/NettyRemotingClientTest.java</include>
<include>**/remote/NettyUtilTest.java</include>