ehlxr
/
did
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Optimized code

master
ehlxr 2021-01-20 15:25:58 +08:00
parent eaeeb56e93
commit 13e80884d7
23 changed files with 341 additions and 282 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
did-common/src/main/java/cn/ceres/did/common/Constants.java
View File

@ -12,34 +12,42 @@ public class Constants {
return SYS_ENV.get(key) == null ? "" : SYS_ENV.get(key);
}
public static String DEFAULT_HOST = "localhost";
public static String SERVER_HOST = "localhost";
/**
* HTTP协议和SDK协议服务器的端口
* HTTP 协议和 SDK 协议服务器默认端口
*/
public static int HTTP_PORT = 16830;
public static int SDKS_PORT = 16831;
public static int SDK_PORT = 16831;
/**
* HTTP协议和SDK协议的请求路径
*/
public static String HTTP_REQUEST = "did";
public static String SDKS_REQUEST = "did";
/**
* 数据中心的标识ID，取值范围：0~31
* 机器或进程的标识ID，取值范围：0~31
* 两个标识ID组合在分布式环境中必须唯一
* 数据中心默认标识 ID，取值范围：0~31
* 机器或进程默认标识 ID，取值范围：0~31
* <p>
* 两个标识 ID 组合在分布式环境中必须唯一
*/
public static long DATACENTER_ID = 1;
public static long MACHINES_ID = 1;
/**
* 流量控制，表示每秒处理的并发数
* Server 流量控制，表示每秒处理的并发数
*/
public static int HANDLE_HTTP_TPS = 10000;
public static int HANDLE_SDKS_TPS = 50000;
public static int HANDLE_SDK_TPS = 50000;
/**
* sdk client 流量控制，表示每秒处理的并发数
*/
public static int SDK_CLIENT_ASYNC_TPS = 100000;
public static int SDK_CLIENT_ONEWAY_TPS = 100000;
public static int ACQUIRE_TIMEOUTMILLIS = 5000;
/**
* sdk client 默认超时时间
*/
public static int SDK_CLIENT_TIMEOUTMILLIS = 2000;
private Constants() {
}
}

10
did-common/src/main/java/cn/ceres/did/common/NettyUtil.java
View File

@ -1,7 +1,6 @@
package cn.ceres.did.common;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -36,11 +35,8 @@ public class NettyUtil {
public static void closeChannel(Channel channel) {
final String addrRemote = parseRemoteAddr(channel);
channel.close().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
logger.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote, future.isSuccess());
}
});
channel.close().addListener((ChannelFutureListener) future ->
logger.info("closeChannel: close the connection to remote address[{}] result: {}",
addrRemote, future.isSuccess()));
}
}

2
did-sdk/src/main/java/cn/ceres/did/sdk/SdkProto.java → did-common/src/main/java/cn/ceres/did/common/SdkProto.java
View File

@ -1,4 +1,4 @@
package cn.ceres.did.sdk;
package cn.ceres.did.common;
import java.util.concurrent.atomic.AtomicInteger;

130
did-sdk/src/main/java/cn/ceres/did/client/AbstractClient.java
View File

@ -1,7 +1,8 @@
package cn.ceres.did.client;
import cn.ceres.did.common.Constants;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.sdk.SdkProto;
import cn.ceres.did.common.SdkProto;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
@ -10,36 +11,43 @@ import io.netty.channel.nio.NioEventLoopGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author ehlxr
*/
@SuppressWarnings({"unused", "UnusedReturnValue"})
public abstract class AbstractClient implements Client {
private final Logger logger = LoggerFactory.getLogger(AbstractClient.class);
private final Semaphore asyncSemaphore = new Semaphore(100000);
private final Semaphore onewaySemaphore = new Semaphore(100000);
private final Semaphore asyncSemaphore = new Semaphore(Constants.SDK_CLIENT_ASYNC_TPS);
private final Semaphore onewaySemaphore = new Semaphore(Constants.SDK_CLIENT_ONEWAY_TPS);
ConcurrentMap<Integer, ResponseFuture> asyncResponse;
NioEventLoopGroup workGroup;
ChannelFuture channelFuture;
Bootstrap bootstrap;
int timeoutMillis = 2000;
public int getTimeoutMillis() {
return timeoutMillis;
}
int timeoutMillis;
String host;
int port;
public void setTimeoutMillis(int timeoutMillis) {
this.timeoutMillis = timeoutMillis;
}
public void init() {
asyncResponse = new ConcurrentHashMap<>(16);
workGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 10, new ThreadFactory() {
public void setHost(String host) {
this.host = host;
}
public void setPort(int port) {
this.port = port;
}
void init() {
workGroup = new NioEventLoopGroup(new ThreadFactory() {
private final AtomicInteger index = new AtomicInteger(0);
@Override
@ -53,47 +61,49 @@ public abstract class AbstractClient implements Client {
@Override
public void shutdown() {
if (workGroup != null) {
workGroup.shutdownGracefully();
logger.info("SDK Client shutdowning......");
try {
if (workGroup != null) {
workGroup.shutdownGracefully().sync();
}
} catch (Exception e) {
logger.error("Client EventLoopGroup shutdown error.", e);
}
logger.info("SDK Client shutdown finish!");
}
@Override
public SdkProto invokeSync(SdkProto sdkProto, long timeoutMillis) throws Exception {
public SdkProto invokeSync(long timeoutMillis) throws Exception {
final Channel channel = channelFuture.channel();
if (channel.isActive()) {
final SdkProto sdkProto = new SdkProto();
final int rqid = sdkProto.getRqid();
try {
final ResponseFuture responseFuture = new ResponseFuture(rqid, timeoutMillis, null, null);
asyncResponse.put(rqid, responseFuture);
final ResponseFuture responseFuture = new ResponseFuture(timeoutMillis, null, null);
REPONSE_MAP.put(rqid, responseFuture);
channel.writeAndFlush(sdkProto).addListener((ChannelFutureListener) channelFuture -> {
if (channelFuture.isSuccess()) {
//发送成功后立即跳出
responseFuture.setIsSendStateOk(true);
return;
}
// 代码执行到此说明发送失败，需要释放资源
asyncResponse.remove(rqid);
REPONSE_MAP.remove(rqid);
responseFuture.putResponse(null);
responseFuture.setCause(channelFuture.cause());
logger.warn("send a request command to channel <" + NettyUtil.parseRemoteAddr(channel) + "> failed.");
logger.warn("send a request command to channel <{}> failed.", NettyUtil.parseRemoteAddr(channel));
});
// 阻塞等待响应
SdkProto resultProto = responseFuture.waitResponse(timeoutMillis);
if (null == resultProto) {
if (responseFuture.isSendStateOk()) {
throw new Exception(NettyUtil.parseRemoteAddr(channel) + timeoutMillis + responseFuture.getCause());
} else {
throw new Exception(NettyUtil.parseRemoteAddr(channel), responseFuture.getCause());
}
throw new Exception(NettyUtil.parseRemoteAddr(channel), responseFuture.getCause());
}
return resultProto;
} catch (Exception e) {
logger.error("invokeSync fail, addr is " + NettyUtil.parseRemoteAddr(channel), e);
throw new Exception(NettyUtil.parseRemoteAddr(channel), e);
} finally {
asyncResponse.remove(rqid);
REPONSE_MAP.remove(rqid);
}
} else {
NettyUtil.closeChannel(channel);
@ -102,22 +112,21 @@ public abstract class AbstractClient implements Client {
}
@Override
public void invokeAsync(SdkProto sdkProto, long timeoutMillis, final InvokeCallback invokeCallback) throws Exception {
public void invokeAsync(long timeoutMillis, InvokeCallback invokeCallback) throws Exception {
final Channel channel = channelFuture.channel();
if (channel.isOpen() && channel.isActive()) {
final SdkProto sdkProto = new SdkProto();
final int rqid = sdkProto.getRqid();
boolean acquired = asyncSemaphore.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
if (acquired) {
final ResponseFuture responseFuture = new ResponseFuture(rqid, timeoutMillis, invokeCallback, asyncSemaphore);
asyncResponse.put(rqid, responseFuture);
if (asyncSemaphore.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS)) {
final ResponseFuture responseFuture = new ResponseFuture(timeoutMillis, invokeCallback, asyncSemaphore);
REPONSE_MAP.put(rqid, responseFuture);
try {
channelFuture.channel().writeAndFlush(sdkProto).addListener((ChannelFutureListener) channelFuture -> {
channelFuture.channel().writeAndFlush(sdkProto).addListener(channelFuture -> {
if (channelFuture.isSuccess()) {
responseFuture.setIsSendStateOk(true);
return;
}
// 代码执行到些说明发送失败，需要释放资源
asyncResponse.remove(rqid);
REPONSE_MAP.remove(rqid);
responseFuture.setCause(channelFuture.cause());
responseFuture.putResponse(null);
@ -128,47 +137,18 @@ public abstract class AbstractClient implements Client {
} finally {
responseFuture.release();
}
logger.warn("send a request command to channel <" + NettyUtil.parseRemoteAddr(channel) + "> failed.", channelFuture.cause());
logger.warn("send a request command to channel <{}> failed.",
NettyUtil.parseRemoteAddr(channel), channelFuture.cause());
});
} catch (Exception e) {
responseFuture.release();
logger.warn("send a request to channel <" + NettyUtil.parseRemoteAddr(channel) + "> Exception", e);
logger.warn("send a request to channel <{}> Exception",
NettyUtil.parseRemoteAddr(channel), e);
throw new Exception(NettyUtil.parseRemoteAddr(channel), e);
}
} else {
String info = String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread " + "nums: %d semaphoreAsyncValue: %d",
timeoutMillis, this.asyncSemaphore.getQueueLength(), this.asyncSemaphore.availablePermits());
logger.warn(info);
throw new Exception(info);
}
} else {
NettyUtil.closeChannel(channel);
throw new Exception(NettyUtil.parseRemoteAddr(channel));
}
}
@Override
public void invokeOneWay(SdkProto sdkProto, long timeoutMillis) throws Exception {
final Channel channel = channelFuture.channel();
if (channel.isActive()) {
final int rqid = sdkProto.getRqid();
boolean acquired = onewaySemaphore.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
if (acquired) {
try {
channelFuture.channel().writeAndFlush(sdkProto).addListener((ChannelFutureListener) channelFuture -> {
onewaySemaphore.release();
if (!channelFuture.isSuccess()) {
logger.warn("send a request command to channel <" + NettyUtil.parseRemoteAddr(channel) + "> failed.");
}
});
} catch (Exception e) {
logger.warn("send a request to channel <" + NettyUtil.parseRemoteAddr(channel) + "> Exception");
throw new Exception(NettyUtil.parseRemoteAddr(channel), e);
} finally {
asyncResponse.remove(rqid);
}
} else {
String info = String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread " + "nums: %d semaphoreAsyncValue: %d",
String info = String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread " +
"nums: %d semaphoreAsyncValue: %d",
timeoutMillis, this.asyncSemaphore.getQueueLength(), this.asyncSemaphore.availablePermits());
logger.warn(info);
throw new Exception(info);
@ -182,4 +162,12 @@ public abstract class AbstractClient implements Client {
public long invoke() throws Exception {
return invoke(timeoutMillis);
}
public SdkProto invokeSync() throws Exception {
return invokeSync(timeoutMillis);
}
public void invokeAsync(InvokeCallback invokeCallback) throws Exception {
invokeAsync(timeoutMillis, invokeCallback);
}
}

43
did-sdk/src/main/java/cn/ceres/did/client/Client.java
View File

@ -1,22 +1,53 @@
package cn.ceres.did.client;
import cn.ceres.did.sdk.SdkProto;
import cn.ceres.did.common.SdkProto;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* @author ehlxr
*/
public interface Client {
ConcurrentMap<Integer, ResponseFuture> REPONSE_MAP = new ConcurrentHashMap<>();
/**
* 启动 sdk client
*/
void start();
/**
* 停止 sdk client
*/
void shutdown();
SdkProto invokeSync(SdkProto proto, long timeoutMillis) throws Exception;
/**
* 同步调用
*
* @param timeoutMillis 超时时间
* @return {@link SdkProto}
* @throws Exception 调用异常
*/
SdkProto invokeSync(long timeoutMillis) throws Exception;
void invokeAsync(SdkProto proto, long timeoutMillis, InvokeCallback invokeCallback) throws Exception;
void invokeOneWay(SdkProto proto, long timeoutMillis) throws Exception;
/**
* 异步调用
*
* @param timeoutMillis 超时时间
* @param invokeCallback 回调接口
* @throws Exception 调用异常
*/
void invokeAsync(long timeoutMillis, InvokeCallback invokeCallback) throws Exception;
/**
* 获取 id
*
* @param timeoutMillis 超时时间
* @return id
* @throws Exception 调用异常
*/
default long invoke(long timeoutMillis) throws Exception {
return invokeSync(new SdkProto(), timeoutMillis).getDid();
return invokeSync(timeoutMillis).getDid();
}
}

5
did-sdk/src/main/java/cn/ceres/did/client/InvokeCallback.java
View File

@ -5,5 +5,10 @@ package cn.ceres.did.client;
*/
public interface InvokeCallback {
/**
* 异步回调处理方法
*
* @param responseFuture {@link ResponseFuture}
*/
void operationComplete(ResponseFuture responseFuture);
}

18
did-sdk/src/main/java/cn/ceres/did/client/ResponseFuture.java
View File

@ -1,6 +1,6 @@
package cn.ceres.did.client;
import cn.ceres.did.sdk.SdkProto;
import cn.ceres.did.common.SdkProto;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
@ -11,7 +11,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
* @author ehlxr
*/
public class ResponseFuture {
private final int rqid;
private final long timeoutMillis;
private final Semaphore semaphore;
private final InvokeCallback invokeCallback;
@ -21,14 +20,11 @@ public class ResponseFuture {
private volatile Throwable cause;
private volatile SdkProto sdkProto;
private volatile boolean isSendStateOk;
public ResponseFuture(int rqid, long timeoutMillis, InvokeCallback invokeCallback, Semaphore semaphore) {
this.rqid = rqid;
public ResponseFuture(long timeoutMillis, InvokeCallback invokeCallback, Semaphore semaphore) {
this.timeoutMillis = timeoutMillis;
this.invokeCallback = invokeCallback;
this.semaphore = semaphore;
this.isSendStateOk = false;
}
/**
@ -75,14 +71,6 @@ public class ResponseFuture {
this.sdkProto = sdkProto;
}
public boolean isSendStateOk() {
return isSendStateOk;
}
public void setIsSendStateOk(boolean isSendStateOk) {
this.isSendStateOk = isSendStateOk;
}
public Throwable getCause() {
return cause;
}
@ -94,7 +82,6 @@ public class ResponseFuture {
@Override
public String toString() {
return "ResponseFuture{" +
"rqid=" + rqid +
", timeoutMillis=" + timeoutMillis +
", semaphore=" + semaphore +
", invokeCallback=" + invokeCallback +
@ -103,7 +90,6 @@ public class ResponseFuture {
", countDownLatch=" + countDownLatch +
", cause=" + cause +
", sdkProto=" + sdkProto +
", isSendStateOk=" + isSendStateOk +
'}';
}
}

63
did-sdk/src/main/java/cn/ceres/did/client/SdkClient.java
View File

@ -1,11 +1,12 @@
package cn.ceres.did.client;
import cn.ceres.did.client.handler.SdkClientDecoder;
import cn.ceres.did.client.handler.SdkClientEncoder;
import cn.ceres.did.client.handler.SdkClientHandler;
import cn.ceres.did.common.Constants;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.sdk.SdkClientDecoder;
import cn.ceres.did.sdk.SdkClientEncoder;
import cn.ceres.did.sdk.SdkProto;
import io.netty.channel.*;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
@ -18,19 +19,26 @@ import java.net.InetSocketAddress;
*/
public class SdkClient extends AbstractClient {
private final Logger logger = LoggerFactory.getLogger(SdkClient.class);
private String host;
private int port;
public SdkClient(String host, int port) {
this(host, port, Constants.SDK_CLIENT_TIMEOUTMILLIS);
}
public SdkClient(String host, int port, int timeoutMillis) {
this.host = host;
this.port = port;
this.timeoutMillis = timeoutMillis;
}
public SdkClient() {
this("".equals(Constants.getEnv("SDK_HOST")) ? Constants.SERVER_HOST : Constants.getEnv("HTTP_PORT"),
"".equals(Constants.getEnv("SDK_PORT")) ? Constants.SDK_PORT : Integer.parseInt(Constants.getEnv("SDK_PORT")));
}
@Override
public void start() {
init();
bootstrap.group(workGroup)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
// .option(ChannelOption.TCP_NODELAY, true)
@ -46,8 +54,7 @@ public class SdkClient extends AbstractClient {
});
try {
channelFuture = bootstrap.connect((host == null || "".equals(host)) ? Constants.DEFAULT_HOST : host,
port == 0 ? Constants.SDKS_PORT : port).sync();
channelFuture = bootstrap.connect(host, port).sync();
channelFuture.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> {
logger.warn("client channel close.", channelFuture.cause());
@ -62,42 +69,4 @@ public class SdkClient extends AbstractClient {
}
}
class SdkClientHandler extends SimpleChannelInboundHandler<SdkProto> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, SdkProto sdkProto) {
final int rqid = sdkProto.getRqid();
final ResponseFuture responseFuture = asyncResponse.get(rqid);
if (responseFuture != null) {
responseFuture.setSdkProto(sdkProto);
responseFuture.release();
asyncResponse.remove(rqid);
// 异步请求，执行回调函数
if (responseFuture.getInvokeCallback() != null) {
responseFuture.executeInvokeCallback();
} else {
// 同步请求，返回数据并释放CountDown
responseFuture.putResponse(sdkProto);
}
} else {
logger.warn("receive response, but not matched any request, address is {}", NettyUtil.parseRemoteAddr(ctx.channel()));
logger.warn("response data is {}", sdkProto.toString());
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
logger.error("SdkHandler error", cause);
NettyUtil.closeChannel(ctx.channel());
}
}
public void setHost(String host) {
this.host = host;
}
public void setPort(int port) {
this.port = port;
}
}

30
did-sdk/src/main/java/cn/ceres/did/sdk/SdkClientDecoder.java → did-sdk/src/main/java/cn/ceres/did/client/handler/SdkClientDecoder.java
View File

@ -1,6 +1,31 @@
package cn.ceres.did.sdk;
/*
* The MIT License (MIT)
*
* Copyright © 2020 xrv <xrg@live.com>
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package cn.ceres.did.client.handler;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.common.SdkProto;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
@ -10,9 +35,10 @@ import org.slf4j.LoggerFactory;
/**
* @author ehlxr
* @since 2021-01-20 14:42.
*/
public class SdkClientDecoder extends FixedLengthFrameDecoder {
private static final Logger logger = LoggerFactory.getLogger(SdkClientDecoder.class);
private final Logger logger = LoggerFactory.getLogger(SdkClientDecoder.class);
public SdkClientDecoder(int frameLength) {
super(frameLength);

59
did-sdk/src/main/java/cn/ceres/did/client/handler/SdkClientEncoder.java Normal file
View File

@ -0,0 +1,59 @@
/*
* The MIT License (MIT)
*
* Copyright © 2020 xrv <xrg@live.com>
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package cn.ceres.did.client.handler;
import cn.ceres.did.client.SdkClient;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.common.SdkProto;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author ehlxr
* @since 2021-01-20 14:43.
*/
public class SdkClientEncoder extends MessageToByteEncoder<SdkProto> {
private final Logger logger = LoggerFactory.getLogger(SdkClient.class);
@Override
protected void encode(ChannelHandlerContext ctx, SdkProto sdkProto, ByteBuf out) {
out.writeInt(sdkProto.getRqid());
out.writeLong(sdkProto.getDid());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
Channel channel = ctx.channel();
logger.error(String.format("SdkServerEncoder channel [%s] error and will be closed",
NettyUtil.parseRemoteAddr(channel)), cause);
NettyUtil.closeChannel(channel);
}
}

70
did-sdk/src/main/java/cn/ceres/did/client/handler/SdkClientHandler.java Normal file
View File

@ -0,0 +1,70 @@
/*
* The MIT License (MIT)
*
* Copyright © 2020 xrv <xrg@live.com>
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package cn.ceres.did.client.handler;
import cn.ceres.did.client.Client;
import cn.ceres.did.client.ResponseFuture;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.common.SdkProto;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author ehlxr
* @since 2021-01-20 14:43.
*/
public class SdkClientHandler extends SimpleChannelInboundHandler<SdkProto> {
private final Logger logger = LoggerFactory.getLogger(SdkClientHandler.class);
@Override
protected void channelRead0(ChannelHandlerContext ctx, SdkProto sdkProto) {
final int rqid = sdkProto.getRqid();
final ResponseFuture responseFuture = Client.REPONSE_MAP.get(rqid);
if (responseFuture != null) {
responseFuture.setSdkProto(sdkProto);
responseFuture.release();
Client.REPONSE_MAP.remove(rqid);
// 异步请求，执行回调函数
if (responseFuture.getInvokeCallback() != null) {
responseFuture.executeInvokeCallback();
} else {
// 同步请求，返回数据并释放 CountDown
responseFuture.putResponse(sdkProto);
}
} else {
logger.warn("receive response {}, but not matched any request, address is {}",
sdkProto, NettyUtil.parseRemoteAddr(ctx.channel()));
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
logger.error("SdkHandler error", cause);
NettyUtil.closeChannel(ctx.channel());
}
}

29
did-sdk/src/main/java/cn/ceres/did/sdk/SdkClientEncoder.java
View File

@ -1,29 +0,0 @@
package cn.ceres.did.sdk;
import cn.ceres.did.common.NettyUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author ehlxr
*/
public class SdkClientEncoder extends MessageToByteEncoder<SdkProto> {
private static final Logger logger = LoggerFactory.getLogger(SdkClientEncoder.class);
@Override
protected void encode(ChannelHandlerContext ctx, SdkProto sdkProto, ByteBuf out) {
out.writeInt(sdkProto.getRqid());
out.writeLong(sdkProto.getDid());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
Channel channel = ctx.channel();
logger.error(String.format("SdkServerEncoder channel [%s] error and will be closed", NettyUtil.parseRemoteAddr(channel)), cause);
NettyUtil.closeChannel(channel);
}
}

14
did-sdk/src/test/java/cn/ceres/did/DidSdkPressTest.java
View File

@ -1,7 +1,6 @@
package cn.ceres.did;
import cn.ceres.did.client.SdkClient;
import cn.ceres.did.sdk.SdkProto;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@ -21,9 +20,7 @@ public class DidSdkPressTest {
@Before
public void init() {
client = new SdkClient("127.0.0.1", 16831);
// client = new SdkClient();
client.init();
client = new SdkClient();
client.start();
}
@ -32,7 +29,6 @@ public class DidSdkPressTest {
client.shutdown();
}
@Test
public void asyncTest() throws Exception {
long start;
@ -47,8 +43,7 @@ public class DidSdkPressTest {
final CountDownLatch countDownLatch = new CountDownLatch(NUM);
start = System.currentTimeMillis();
for (int i = 0; i < NUM; i++) {
final SdkProto sdkProto = new SdkProto();
client.invokeAsync(sdkProto, 5000, responseFuture -> countDownLatch.countDown());
client.invokeAsync(responseFuture -> countDownLatch.countDown());
}
// countDownLatch.await(10, TimeUnit.SECONDS);
@ -74,12 +69,11 @@ public class DidSdkPressTest {
long amount = 0;
long allcast = 0;
for (int k = 0; k < 20; k++) {
for (int k = 0; k < 10; k++) {
start = System.currentTimeMillis();
int NUM = 60000;
for (int i = 0; i < NUM; i++) {
final SdkProto sdkProto = new SdkProto();
client.invokeSync(sdkProto, 5000);
client.invokeSync();
}
end = System.currentTimeMillis();

25
did-sdk/src/test/java/cn/ceres/did/DidSdkTest.java
View File

@ -1,7 +1,8 @@
package cn.ceres.did;
import cn.ceres.did.client.SdkClient;
import cn.ceres.did.sdk.SdkProto;
import cn.ceres.did.common.SdkProto;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@ -17,37 +18,35 @@ public class DidSdkTest {
@Before
public void init() {
client = new SdkClient("127.0.0.1", 16831);
// SdkClient client = new SdkClient();
client.init();
client = new SdkClient("127.0.0.1", 16831, 5000);
client.start();
}
@After
public void destroy() {
client.shutdown();
}
@Test
public void didSdkTest() throws Exception {
// 测试同步请求，关注rqid是否对应
for (int i = 0; i < NUM; i++) {
SdkProto sdkProto = new SdkProto();
System.out.println(i + " sendProto: " + sdkProto.toString());
SdkProto resultProto = client.invokeSync(sdkProto, 2000);
System.out.println(i + " resultProto: " + resultProto.toString());
SdkProto resultProto = client.invokeSync();
System.out.println(i + " resultProto: " + resultProto);
}
System.out.println("invokeync test finish");
// 测试异步请求，关注rqid是否对应
final CountDownLatch countDownLatch = new CountDownLatch(NUM);
for (int i = 0; i < NUM; i++) {
final SdkProto sdkProto = new SdkProto();
final int finalI = i;
client.invokeAsync(sdkProto, 2000, responseFuture -> {
System.out.println(finalI + " sendProto: " + sdkProto.toString());
client.invokeAsync(responseFuture -> {
countDownLatch.countDown();
System.out.println(finalI + " resultProto: " + responseFuture.getSdkProto().toString());
System.out.println(finalI + " resultProto: " + responseFuture.getSdkProto());
});
}
countDownLatch.await(10, TimeUnit.SECONDS);
System.out.println("invokeAsync test finish");
}
@Test

2
did-server/src/main/java/cn/ceres/did/core/SnowFlake.java
View File

@ -8,7 +8,7 @@ package cn.ceres.did.core;
* 协议格式：0 - 0000000000 0000000000 0000000000 0000000000 0 - 00000 - 00000 - 000000000000
* 协议解释：0 - 41 位时间戳 - 5 位数据中心标识 - 5 位机器标识 - 12 位序列号
* <p>
* 1 位标识，由于 long 基本类型在 Java 中是带符号的，最高位是符号位，正数是 0，负数是 1，所以 id 一般是正数，最高位是 0
* 1 位标识，由于 Long 基本类型在 Java 中是带符号的，最高位是符号位，正数是 0，负数是 1，所以 id 一般是正数，最高位是 0
* <p>
* 41 位时间截（毫秒级），注意，41 位时间截不是存储当前时间的时间截，而是存储时间截的差值（当前时间截 - 开始时间截)得到的值，这里的的开始时间截。
* 一般是我们的 id 生成器开始使用的时间，由我们程序来指定的（如下下面程序 START_STMP 属性）。41 位的时间截，可以使用 69 年，(1L << 41) / (1000L * 60 * 60 * 24 * 365) = 69

2
did-server/src/main/java/cn/ceres/did/server/BaseServer.java
View File

@ -1,5 +1,6 @@
package cn.ceres.did.server;
import cn.ceres.did.core.SnowFlake;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.DefaultEventLoopGroup;
@ -22,6 +23,7 @@ public abstract class BaseServer implements Server {
protected ChannelFuture channelFuture;
protected ServerBootstrap serverBootstrap;
protected int port;
protected SnowFlake snowFlake;
public void init() {
defLoopGroup = new DefaultEventLoopGroup(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {

17
did-server/src/main/java/cn/ceres/did/server/http/HttpServer.java
View File

@ -14,20 +14,23 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Http服务器，使用Netty中的Http协议栈，
* 实现中支持多条请求路径，对于不存在的请求路径返回404状态码
* 如：http://localhost:8099/getTime
* Http 服务器，使用 Netty 中的 Http 协议栈，
*
* @author ehlxr
*/
public class HttpServer extends BaseServer {
protected Logger logger = LoggerFactory.getLogger(HttpServer.class);
private final SnowFlake snowFlake;
public HttpServer(SnowFlake snowFlake, int port) {
this.snowFlake = snowFlake;
this.port = port;
}
public HttpServer(SnowFlake snowFlake) {
this.snowFlake = snowFlake;
this.port = "".equals(Constants.getEnv("HTTP_PORT")) ? Constants.HTTP_PORT : Integer.parseInt(Constants.getEnv("HTTP_PORT"));
this.port = "".equals(Constants.getEnv("HTTP_PORT")) ?
Constants.HTTP_PORT :
Integer.parseInt(Constants.getEnv("HTTP_PORT"));
}
@Override
@ -57,10 +60,6 @@ public class HttpServer extends BaseServer {
try {
channelFuture = serverBootstrap.bind(port).sync();
logger.info("HttpServer start success, port is:{}", port);
// channelFuture = serverBootstrap.bind().sync();
// InetSocketAddress addr = (InetSocketAddress) channelFuture.channel().localAddress();
// logger.info("HttpServer start success, port is:{}", addr.getPort());
} catch (InterruptedException e) {
logger.error("HttpServer start fail,", e);
}

5
did-server/src/main/java/cn/ceres/did/server/http/HttpServerHandler.java
View File

@ -15,11 +15,6 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
* 自定义的处理器，目前支持三种请求：
* getTime: 获取服务器当前时间；
* clientInfo: 获取请求客户端的User-Agent信息
* 其它： 返回404状态，并且提示404信息
*
* @author ehlxr
*/
public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

44
did-server/src/main/java/cn/ceres/did/server/sdk/SdkProto.java
View File

@ -1,44 +0,0 @@
package cn.ceres.did.server.sdk;
/**
* @author ehlxr
*/
public class SdkProto {
/**
* 请求的ID
*/
private int rqid;
/**
* 全局的 ID
*/
private long did;
SdkProto(int rqid, long did) {
this.rqid = rqid;
this.did = did;
}
public int getRqid() {
return rqid;
}
public void setRqid(int rqid) {
this.rqid = rqid;
}
public long getDid() {
return did;
}
public void setDid(long did) {
this.did = did;
}
@Override
public String toString() {
return "SdkProto{" +
"rqid=" + rqid +
", did=" + did +
'}';
}
}

14
did-server/src/main/java/cn/ceres/did/server/sdk/SdkServer.java
View File

@ -15,11 +15,17 @@ import org.slf4j.LoggerFactory;
*/
public class SdkServer extends BaseServer {
protected Logger logger = LoggerFactory.getLogger(SdkServer.class);
private final SnowFlake snowFlake;
public SdkServer(SnowFlake snowFlake) {
this.snowFlake = snowFlake;
this.port = "".equals(Constants.getEnv("SDKS_PORT")) ? Constants.SDKS_PORT : Integer.parseInt(Constants.getEnv("SDKS_PORT"));
this.port = "".equals(Constants.getEnv("SDK_PORT")) ?
Constants.SDK_PORT :
Integer.parseInt(Constants.getEnv("SDK_PORT"));
}
public SdkServer(SnowFlake snowFlake, int port) {
this.snowFlake = snowFlake;
this.port = port;
}
@Override
@ -48,10 +54,6 @@ public class SdkServer extends BaseServer {
try {
channelFuture = serverBootstrap.bind(port).sync();
logger.info("SdkServer start success, port is:{}", port);
// channelFuture = serverBootstrap.bind().sync();
// InetSocketAddress addr = (InetSocketAddress) channelFuture.channel().localAddress();
// logger.info("SdkServer start success, port is:{}", addr.getPort());
} catch (InterruptedException e) {
logger.error("SdkServer start fail,", e);
}

1
did-server/src/main/java/cn/ceres/did/server/sdk/SdkServerDecoder.java
View File

@ -1,6 +1,7 @@
package cn.ceres.did.server.sdk;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.common.SdkProto;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;

1
did-server/src/main/java/cn/ceres/did/server/sdk/SdkServerEncoder.java
View File

@ -1,6 +1,7 @@
package cn.ceres.did.server.sdk;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.common.SdkProto;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;

3
did-server/src/main/java/cn/ceres/did/server/sdk/SdkServerHandler.java
View File

@ -2,6 +2,7 @@ package cn.ceres.did.server.sdk;
import cn.ceres.did.common.Constants;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.common.SdkProto;
import cn.ceres.did.core.SnowFlake;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
@ -23,7 +24,7 @@ public class SdkServerHandler extends SimpleChannelInboundHandler<SdkProto> {
/**
* 通过信号量来控制流量
*/
private final Semaphore semaphore = new Semaphore(Constants.HANDLE_SDKS_TPS);
private final Semaphore semaphore = new Semaphore(Constants.HANDLE_SDK_TPS);
private final SnowFlake snowFlake;
SdkServerHandler(SnowFlake snowFlake) {