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

Optimized code

master
ehlxr 2021-01-19 14:45:49 +08:00
parent 2cc021a74f
commit d34eddd492
11 changed files with 76 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
did-common/pom.xml
View File

@ -23,13 +23,4 @@
<artifactId>logback-classic</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

4
did-sdk/pom.xml
View File

@ -27,10 +27,6 @@
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>

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

@ -1,7 +1,7 @@
package cn.ceres.did.client;
import cn.ceres.did.sdk.SdkProto;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.sdk.SdkProto;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
@ -17,7 +17,7 @@ import java.util.concurrent.atomic.AtomicInteger;
* @author ehlxr
*/
public abstract class AbstractClient implements Client {
private Logger logger = LoggerFactory.getLogger(AbstractClient.class);
private final Logger logger = LoggerFactory.getLogger(AbstractClient.class);
private final Semaphore asyncSemaphore = new Semaphore(100000);
private final Semaphore onewaySemaphore = new Semaphore(100000);
@ -30,7 +30,7 @@ public abstract class AbstractClient implements Client {
public void init() {
asyncResponse = new ConcurrentHashMap<>(16);
workGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 10, new ThreadFactory() {
private AtomicInteger index = new AtomicInteger(0);
private final AtomicInteger index = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
@ -57,20 +57,17 @@ public abstract class AbstractClient implements Client {
try {
final ResponseFuture responseFuture = new ResponseFuture(rqid, timeoutMillis, null, null);
asyncResponse.put(rqid, responseFuture);
channel.writeAndFlush(sdkProto).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) {
if (channelFuture.isSuccess()) {
//发送成功后立即跳出
responseFuture.setIsSendStateOk(true);
return;
}
// 代码执行到此说明发送失败，需要释放资源
asyncResponse.remove(rqid);
responseFuture.putResponse(null);
responseFuture.setCause(channelFuture.cause());
logger.warn("send a request command to channel <" + NettyUtil.parseRemoteAddr(channel) + "> failed.");
channel.writeAndFlush(sdkProto).addListener((ChannelFutureListener) channelFuture -> {
if (channelFuture.isSuccess()) {
//发送成功后立即跳出
responseFuture.setIsSendStateOk(true);
return;
}
// 代码执行到此说明发送失败，需要释放资源
asyncResponse.remove(rqid);
responseFuture.putResponse(null);
responseFuture.setCause(channelFuture.cause());
logger.warn("send a request command to channel <" + NettyUtil.parseRemoteAddr(channel) + "> failed.");
});
// 阻塞等待响应
SdkProto resultProto = responseFuture.waitResponse(timeoutMillis);
@ -104,27 +101,24 @@ public abstract class AbstractClient implements Client {
final ResponseFuture responseFuture = new ResponseFuture(rqid, timeoutMillis, invokeCallback, asyncSemaphore);
asyncResponse.put(rqid, responseFuture);
try {
channelFuture.channel().writeAndFlush(sdkProto).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) {
if (channelFuture.isSuccess()) {
responseFuture.setIsSendStateOk(true);
return;
}
// 代码执行到些说明发送失败，需要释放资源
asyncResponse.remove(rqid);
responseFuture.setCause(channelFuture.cause());
responseFuture.putResponse(null);
try {
responseFuture.executeInvokeCallback();
} catch (Exception e) {
logger.warn("excute callback in writeAndFlush addListener, and callback throw", e);
} finally {
responseFuture.release();
}
logger.warn("send a request command to channel <" + NettyUtil.parseRemoteAddr(channel) + "> failed.", channelFuture.cause());
channelFuture.channel().writeAndFlush(sdkProto).addListener((ChannelFutureListener) channelFuture -> {
if (channelFuture.isSuccess()) {
responseFuture.setIsSendStateOk(true);
return;
}
// 代码执行到些说明发送失败，需要释放资源
asyncResponse.remove(rqid);
responseFuture.setCause(channelFuture.cause());
responseFuture.putResponse(null);
try {
responseFuture.executeInvokeCallback();
} catch (Exception e) {
logger.warn("excute callback in writeAndFlush addListener, and callback throw", e);
} finally {
responseFuture.release();
}
logger.warn("send a request command to channel <" + NettyUtil.parseRemoteAddr(channel) + "> failed.", channelFuture.cause());
});
} catch (Exception e) {
responseFuture.release();
@ -151,13 +145,10 @@ public abstract class AbstractClient implements Client {
boolean acquired = onewaySemaphore.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
if (acquired) {
try {
channelFuture.channel().writeAndFlush(sdkProto).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) {
onewaySemaphore.release();
if (!channelFuture.isSuccess()) {
logger.warn("send a request command to channel <" + NettyUtil.parseRemoteAddr(channel) + "> failed.");
}
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) {
@ -167,7 +158,8 @@ public abstract class AbstractClient implements Client {
asyncResponse.remove(rqid);
}
} else {
String info = String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread " + "nums: %d semaphoreAsyncValue: %d", timeoutMillis, this.asyncSemaphore.getQueueLength(), this.asyncSemaphore.availablePermits());
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);
}

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

@ -1,10 +1,10 @@
package cn.ceres.did.client;
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 cn.ceres.did.common.Constants;
import cn.ceres.did.common.NettyUtil;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
@ -17,7 +17,7 @@ import java.net.InetSocketAddress;
* @author ehlxr
*/
public class SdkClient extends AbstractClient {
private Logger logger = LoggerFactory.getLogger(SdkClient.class);
private final Logger logger = LoggerFactory.getLogger(SdkClient.class);
private String host;
private int port;
@ -47,12 +47,9 @@ 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.channel().closeFuture().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) {
logger.warn("client channel close.", channelFuture.cause());
shutdown();
}
channelFuture.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> {
logger.warn("client channel close.", channelFuture.cause());
shutdown();
});
InetSocketAddress address = (InetSocketAddress) channelFuture.channel().remoteAddress();

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

@ -6,7 +6,7 @@ import java.util.concurrent.atomic.AtomicInteger;
* @author ehlxr
*/
public class SdkProto {
private static final AtomicInteger requestId = new AtomicInteger(0);
private static final AtomicInteger REQUEST_ID = new AtomicInteger(0);
/**
* 请求的ID
@ -18,7 +18,7 @@ public class SdkProto {
private long did;
public SdkProto() {
rqid = requestId.incrementAndGet();
rqid = REQUEST_ID.incrementAndGet();
did = 0;
}

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

@ -1,7 +1,5 @@
package cn.ceres.did;
import cn.ceres.did.client.InvokeCallback;
import cn.ceres.did.client.ResponseFuture;
import cn.ceres.did.client.SdkClient;
import cn.ceres.did.sdk.SdkProto;
import org.junit.Test;
@ -36,13 +34,10 @@ public class DidSdkTest {
for (int i = 0; i < NUM; i++) {
final SdkProto sdkProto = new SdkProto();
final int finalI = i;
client.invokeAsync(sdkProto, 2000, new InvokeCallback() {
@Override
public void operationComplete(ResponseFuture responseFuture) {
System.out.println(finalI + " sendProto: " + sdkProto.toString());
countDownLatch.countDown();
System.out.println(finalI + " resultProto: " + responseFuture.getSdkProto().toString());
}
client.invokeAsync(sdkProto, 2000, responseFuture -> {
System.out.println(finalI + " sendProto: " + sdkProto.toString());
countDownLatch.countDown();
System.out.println(finalI + " resultProto: " + responseFuture.getSdkProto().toString());
});
}
countDownLatch.await(10, TimeUnit.SECONDS);

4
did-server/pom.xml
View File

@ -23,10 +23,6 @@
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>

15
did-server/src/main/java/cn/ceres/did/ServerStarter.java
View File

@ -1,9 +1,9 @@
package cn.ceres.did;
import cn.ceres.did.common.Constants;
import cn.ceres.did.core.SnowFlake;
import cn.ceres.did.server.http.HttpServer;
import cn.ceres.did.server.sdk.SdkServer;
import cn.ceres.did.common.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -40,13 +40,10 @@ public class ServerStarter {
sdkServer.init();
sdkServer.start();
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
httpServer.shutdown();
sdkServer.shutdown();
System.exit(0);
}
});
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
httpServer.shutdown();
sdkServer.shutdown();
System.exit(0);
}));
}
}

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

@ -1,8 +1,8 @@
package cn.ceres.did.server.sdk;
import cn.ceres.did.common.Constants;
import cn.ceres.did.server.BaseServer;
import cn.ceres.did.core.SnowFlake;
import cn.ceres.did.server.BaseServer;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.SocketChannel;
@ -18,7 +18,7 @@ public class SdkServer extends BaseServer {
public SdkServer(SnowFlake snowFlake) {
this.snowFlake = snowFlake;
this.port = "".equals(Constants.getEnv("SDKS_PORT")) ? Constants.SDKS_PORT : Integer.valueOf(Constants.getEnv("SDKS_PORT"));
this.port = "".equals(Constants.getEnv("SDKS_PORT")) ? Constants.SDKS_PORT : Integer.parseInt(Constants.getEnv("SDKS_PORT"));
}
@Override

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

@ -3,7 +3,10 @@ package cn.ceres.did.server.sdk;
import cn.ceres.did.common.Constants;
import cn.ceres.did.common.NettyUtil;
import cn.ceres.did.core.SnowFlake;
import io.netty.channel.*;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -34,12 +37,7 @@ public class SdkServerHandler extends SimpleChannelInboundHandler<SdkProto> {
if (semaphore.tryAcquire(Constants.ACQUIRE_TIMEOUTMILLIS, TimeUnit.MILLISECONDS)) {
try {
sdkProto.setDid(snowFlake.nextId());
ctx.channel().writeAndFlush(sdkProto).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) {
semaphore.release();
}
});
ctx.channel().writeAndFlush(sdkProto).addListener((ChannelFutureListener) channelFuture -> semaphore.release());
} catch (Exception e) {
semaphore.release();
logger.error("SdkServerhandler error", e);

27
pom.xml
View File

@ -16,22 +16,34 @@
<name>did</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<!--目标编译的 Java 版本可以通过以下属性指定，不用配置 maven-compiler-plugin 插件-->
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<netty.version>4.1.6.Final</netty.version>
<logback.version>1.1.7</logback.version>
<junit.version>4.11</junit.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.6.Final</version>
<version>${netty.version}</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.1.7</version>
<version>${logback.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<version>${junit.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
@ -40,15 +52,6 @@
<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.5.1</version>
<configuration>
<source>1.7</source>
<target>1.7</target>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>