lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化SncpClient的新实现

Browse Source
This commit is contained in:
redkale
2023-03-22 16:02:17 +08:00
parent 358ad80f21
commit 039ed0f569
41 changed files with 2030 additions and 1673 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
src/test/java/org/redkale/test/service/ABMainService.java
View File

@@ -5,21 +5,23 @@
*/
package org.redkale.test.service;
import java.io.*;
import java.net.*;
import java.nio.*;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.logging.*;
import org.redkale.annotation.*;
import java.util.logging.Level;
import org.redkale.annotation.Resource;
import org.redkale.boot.*;
import org.redkale.convert.bson.*;
import org.redkale.convert.json.*;
import org.redkale.net.*;
import org.redkale.convert.bson.BsonConvert;
import org.redkale.convert.json.JsonConvert;
import org.redkale.net.AsyncIOGroup;
import org.redkale.net.client.ClientAddress;
import org.redkale.net.http.*;
import org.redkale.net.sncp.*;
import org.redkale.service.*;
import org.redkale.service.Service;
import org.redkale.util.AnyValue.DefaultAnyValue;
import org.redkale.util.*;
@@ -30,85 +32,96 @@ import org.redkale.util.*;
@RestService(name = "abmain")
public class ABMainService implements Service {
private static final int abport = 8866;
@Resource
private BCService bcService;
public static void remote(String[] args) throws Throwable {
System.out.println("------------------- 远程模式调用 -----------------------------------");
final Application application = Application.create(true);
final int abport = 8888;
final AsyncIOGroup asyncGroup = new AsyncIOGroup(8192, 16);
asyncGroup.start();
ResourceFactory resFactory = ResourceFactory.create();
ExecutorService executor = Executors.newSingleThreadExecutor();
final TransportFactory transFactory = TransportFactory.create(asyncGroup, 0, 0);
transFactory.addGroupInfo("g77", new InetSocketAddress("127.0.0.1", 5577));
transFactory.addGroupInfo("g88", new InetSocketAddress("127.0.0.1", 5588));
transFactory.addGroupInfo("g99", new InetSocketAddress("127.0.0.1", 5599));
InetSocketAddress sncpAddress = new InetSocketAddress("127.0.0.1", abport);
final SncpClient client = new SncpClient("", asyncGroup, sncpAddress, new ClientAddress(sncpAddress), "TCP", 16, 100);
final ResourceFactory resFactory = ResourceFactory.create();
resFactory.register(JsonConvert.root());
resFactory.register(BsonConvert.root());
final SncpRpcGroups rpcGroups = application.getSncpRpcGroups();
rpcGroups.computeIfAbsent("g77", "TCP").putAddress(new InetSocketAddress("127.0.0.1", 5577));
rpcGroups.computeIfAbsent("g88", "TCP").putAddress(new InetSocketAddress("127.0.0.1", 5588));
rpcGroups.computeIfAbsent("g99", "TCP").putAddress(new InetSocketAddress("127.0.0.1", 5599));
//------------------------ 初始化 CService ------------------------------------
CService cservice = Sncp.createSimpleLocalService(CService.class, null, resFactory, transFactory, new InetSocketAddress("127.0.0.1", 5577), "g77");
CService cservice = Sncp.createSimpleLocalService(CService.class, resFactory);
SncpServer cserver = new SncpServer();
cserver.getResourceFactory().register(application);
cserver.getLogger().setLevel(Level.WARNING);
//cserver.getLogger().setLevel(Level.WARNING);
cserver.addSncpServlet(cservice);
cserver.init(DefaultAnyValue.create("port", 5577));
cserver.start();
//------------------------ 初始化 BCService ------------------------------------
BCService bcservice = Sncp.createSimpleLocalService(BCService.class, null, resFactory, transFactory, new InetSocketAddress("127.0.0.1", 5588), "g88");
CService remoteCService = Sncp.createSimpleRemoteService(CService.class, null, transFactory, new InetSocketAddress("127.0.0.1", 5588), "g77");
resFactory.inject(remoteCService);
resFactory.register("", remoteCService);
BCService bcservice = Sncp.createSimpleLocalService(BCService.class, resFactory);
CService remoteCService = Sncp.createSimpleRemoteService(CService.class, resFactory, rpcGroups, client, "g77");
if (remoteCService != null) {
resFactory.inject(remoteCService);
resFactory.register("", remoteCService);
}
SncpServer bcserver = new SncpServer();
bcserver.getResourceFactory().register(application);
bcserver.getLogger().setLevel(Level.WARNING);
//bcserver.getLogger().setLevel(Level.WARNING);
bcserver.addSncpServlet(bcservice);
bcserver.init(DefaultAnyValue.create("port", 5588));
bcserver.start();
//------------------------ 初始化 ABMainService ------------------------------------
ABMainService service = Sncp.createSimpleLocalService(ABMainService.class, null, resFactory, transFactory, new InetSocketAddress("127.0.0.1", 5599), "g99");
BCService remoteBCService = Sncp.createSimpleRemoteService(BCService.class, null, transFactory, new InetSocketAddress("127.0.0.1", 5599), "g88");
resFactory.inject(remoteBCService);
resFactory.register("", remoteBCService);
HttpServer server = new HttpServer();
server.getResourceFactory().register(application);
server.getLogger().setLevel(Level.WARNING);
server.addRestServlet(null, service, null, HttpServlet.class, "/pipes");
ABMainService service = Sncp.createSimpleLocalService(ABMainService.class, resFactory);
BCService remoteBCService = Sncp.createSimpleRemoteService(BCService.class, resFactory, rpcGroups, client, "g88");
if (remoteBCService != null) {
resFactory.inject(remoteBCService);
resFactory.register("", remoteBCService);
}
resFactory.inject(cservice);
resFactory.inject(bcservice);
resFactory.inject(service);
HttpServer server = new HttpServer();
server.getResourceFactory().register(application);
//server.getLogger().setLevel(Level.WARNING);
server.init(DefaultAnyValue.create("port", abport));
server.addRestServlet(null, service, null, HttpServlet.class, "/pipes");
server.start();
Thread.sleep(100);
System.out.println("开始请求");
//不声明一个新的HttpClient会导致Utility.postHttpContent操作
//同一url在Utility里的httpClient会缓存导致调用是吧，应该是httpClient的bug
java.net.http.HttpClient httpClient = java.net.http.HttpClient.newHttpClient();
System.out.println("httpclient类: " + httpClient.getClass().getName());
//同步方法
String url = "http://127.0.0.1:" + abport + "/pipes/abmain/syncabtime/张先生";
System.out.println(Utility.postHttpContent(url));
String url = "http://127.0.0.1:" + abport + "/pipes/abmain/sab/张先生";
System.out.println(Utility.postHttpContentAsync(httpClient, url, StandardCharsets.UTF_8, null).join());
//异步方法
url = "http://127.0.0.1:" + abport + "/pipes/abmain/asyncabtime/张先生";
System.out.println(Utility.postHttpContent(url));
url = "http://127.0.0.1:" + abport + "/pipes/abmain/abc/张先生";
System.out.println(Utility.postHttpContentAsync(httpClient, url, StandardCharsets.UTF_8, null).join());
//异步方法
url = "http://127.0.0.1:" + abport + "/pipes/abmain/asyncabtime2/张先生";
System.out.println(Utility.postHttpContent(url));
url = "http://127.0.0.1:" + abport + "/pipes/abmain/abc2/张先生";
System.out.println(Utility.postHttpContentAsync(httpClient, url, StandardCharsets.UTF_8, null).join());
server.shutdown();
bcserver.shutdown();
cserver.shutdown();
}
public static void main(String[] args) throws Throwable {
LoggingBaseHandler.initDebugLogConfig();
System.out.println("------------------- 本地模式调用 -----------------------------------");
final Application application = Application.create(true);
final int abport = 8888;
ResourceFactory factory = ResourceFactory.create();
ABMainService service = new ABMainService();
@@ -131,16 +144,17 @@ public class ABMainService implements Service {
server.start();
Thread.sleep(100);
System.out.println("开始请求");
//同步方法
String url = "http://127.0.0.1:" + abport + "/pipes/abmain/syncabtime/张先生";
String url = "http://127.0.0.1:" + abport + "/pipes/abmain/sab/张先生";
System.out.println(Utility.postHttpContent(url));
//异步方法
url = "http://127.0.0.1:" + abport + "/pipes/abmain/asyncabtime/张先生";
url = "http://127.0.0.1:" + abport + "/pipes/abmain/abc/张先生";
System.out.println(Utility.postHttpContent(url));
//异步方法
url = "http://127.0.0.1:" + abport + "/pipes/abmain/asyncabtime2/张先生";
url = "http://127.0.0.1:" + abport + "/pipes/abmain/abc2/张先生";
System.out.println(Utility.postHttpContent(url));
server.shutdown();
@@ -170,16 +184,17 @@ public class ABMainService implements Service {
});
}
@RestMapping(name = "syncabtime")
@RestMapping(name = "sab")
public String abCurrentTime(@RestParam(name = "#") final String name) {
String rs = "同步abCurrentTime: " + bcService.bcCurrentTime(name);
System.out.println("准备执行ABMainService.sab方法");
String rs = "同步abCurrentTime: " + bcService.bcCurrentTime1(name);
System.out.println("执行了 ABMainService.abCurrentTime++++同步方法");
return rs;
}
@RestMapping(name = "asyncabtime")
@RestMapping(name = "abc")
public void abCurrentTime(final CompletionHandler<String, Void> handler, @RestParam(name = "#") final String name) {
bcService.bcCurrentTime(Utility.createAsyncHandler((v, a) -> {
bcService.bcCurrentTime2(Utility.createAsyncHandler((v, a) -> {
System.out.println("执行了 ABMainService.abCurrentTime----异步方法");
String rs = "异步abCurrentTime: " + v;
if (handler != null) {
@@ -192,9 +207,9 @@ public class ABMainService implements Service {
}), name);
}
@RestMapping(name = "asyncabtime2")
@RestMapping(name = "abc2")
public void abCurrentTime(final MyAsyncHandler<String, Void> handler, @RestParam(name = "#") final String name) {
bcService.bcCurrentTime(new MyAsyncHandler<String, Void>() {
bcService.bcCurrentTime3(new MyAsyncHandler<String, Void>() {
@Override
public int id() {
return 1;

29
src/test/java/org/redkale/test/service/BCService.java
View File

@@ -5,10 +5,10 @@
*/
package org.redkale.test.service;
import java.nio.channels.*;
import org.redkale.annotation.*;
import java.nio.channels.CompletionHandler;
import org.redkale.annotation.Resource;
import org.redkale.service.*;
import org.redkale.util.*;
import org.redkale.util.Utility;
/**
*
@@ -33,16 +33,17 @@ public class BCService implements Service {
return serviceType;
}
public String bcCurrentTime(final String name) {
String rs = "同步bcCurrentTime: " + cService.ccCurrentTime(name).getResult();
System.out.println("执行了 BCService.bcCurrentTime++++同步方法");
public String bcCurrentTime1(final String name) {
System.out.println("准备执行BCService.bcCurrentTime1方法");
String rs = "同步bcCurrentTime1: " + cService.ccCurrentTime1(name).getResult();
System.out.println("执行了 BCService.bcCurrentTime1++++同步方法1");
return rs;
}
public void bcCurrentTime(final CompletionHandler<String, Void> handler, final String name) {
cService.ccCurrentTime(Utility.createAsyncHandler((v, a) -> {
System.out.println("执行了 BCService.bcCurrentTime----异步方法");
String rs = "异步bcCurrentTime: " + (v == null ? null : v.getResult());
public void bcCurrentTime2(final CompletionHandler<String, Void> handler, final String name) {
cService.ccCurrentTime2(Utility.createAsyncHandler((v, a) -> {
System.out.println("执行了 BCService.bcCurrentTime2----异步方法2");
String rs = "异步bcCurrentTime2: " + (v == null ? null : v.getResult());
if (handler != null) {
handler.completed(rs, null);
}
@@ -53,8 +54,8 @@ public class BCService implements Service {
}), name);
}
public void bcCurrentTime(final MyAsyncHandler<String, Void> handler, final String name) {
cService.mcCurrentTime(new MyAsyncHandler<RetResult<String>, Void>() {
public void bcCurrentTime3(final MyAsyncHandler<String, Void> handler, final String name) {
cService.mcCurrentTime3(new MyAsyncHandler<RetResult<String>, Void>() {
@Override
public int id() {
return 1;
@@ -62,8 +63,8 @@ public class BCService implements Service {
@Override
public void completed(RetResult<String> v, Void a) {
System.out.println("执行了 BCService.bcCurrentTime----异步方法2");
String rs = "异步bcCurrentTime: " + (v == null ? null : v.getResult());
System.out.println("执行了 BCService.bcCurrentTime3----异步方法3");
String rs = "异步bcCurrentTime3: " + (v == null ? null : v.getResult());
if (handler != null) {
handler.completed(rs, null);
}

24
src/test/java/org/redkale/test/service/CService.java
View File

@@ -5,10 +5,10 @@
*/
package org.redkale.test.service;
import java.nio.channels.*;
import org.redkale.annotation.*;
import java.nio.channels.CompletionHandler;
import org.redkale.annotation.Resource;
import org.redkale.service.*;
import org.redkale.util.*;
import org.redkale.util.Utility;
/**
*
@@ -30,21 +30,21 @@ public class CService implements Service {
return serviceType;
}
public RetResult<String> ccCurrentTime(final String name) {
String rs = "同步ccCurrentTime: " + name + ": " + Utility.formatTime(System.currentTimeMillis());
System.out.println("执行了 CService.ccCurrentTime++++同步方法");
public RetResult<String> ccCurrentTime1(final String name) {
String rs = "同步ccCurrentTime1: " + name + ": " + Utility.formatTime(System.currentTimeMillis());
System.out.println("执行了 CService.ccCurrentTime1++++同步方法1");
return new RetResult(rs);
}
public void ccCurrentTime(final CompletionHandler<RetResult<String>, Void> handler, final String name) {
String rs = "异步ccCurrentTime: " + name + ": " + Utility.formatTime(System.currentTimeMillis());
System.out.println("执行了 CService.ccCurrentTime----异步方法");
public void ccCurrentTime2(final CompletionHandler<RetResult<String>, Void> handler, final String name) {
String rs = "异步ccCurrentTime2: " + name + ": " + Utility.formatTime(System.currentTimeMillis());
System.out.println("执行了 CService.ccCurrentTime2----异步方法2");
if (handler != null) handler.completed(new RetResult(rs), null);
}
public void mcCurrentTime(final MyAsyncHandler<RetResult<String>, Void> handler, final String name) {
String rs = "异步mcCurrentTime: " + name + ": " + Utility.formatTime(System.currentTimeMillis());
System.out.println("执行了 CService.mcCurrentTime----异步方法2");
public void mcCurrentTime3(final MyAsyncHandler<RetResult<String>, Void> handler, final String name) {
String rs = "异步mcCurrentTime3: " + name + ": " + Utility.formatTime(System.currentTimeMillis());
System.out.println("执行了 CService.mcCurrentTime3----异步方法3");
if (handler != null) handler.completed(new RetResult(rs), null);
}
}

8
src/test/java/org/redkale/test/sncp/SncpClientCodecTest.java
View File

@@ -49,11 +49,11 @@ public class SncpClientCodecTest {
SncpHeader header = new SncpHeader(sncpAddress, Uint128.ZERO, Uint128.ZERO);
SncpClientRequest request = new SncpClientRequest();
ByteArray writeArray = new ByteArray();
request.prepare(header, 1, "", new byte[20]);
request.prepare(header, 1, "", new ByteArray().putPlaceholder(20));
System.out.println("request.1 = " + request);
writeArray.put(new byte[SncpHeader.HEADER_SIZE]);
request.writeTo(conn, writeArray);
request.prepare(header, 2, "", new byte[25]);
request.prepare(header, 2, "", new ByteArray().putPlaceholder(25));
System.out.println("request.2 = " + request);
writeArray.put(new byte[SncpHeader.HEADER_SIZE]);
request.writeTo(conn, writeArray);
@@ -63,8 +63,6 @@ public class SncpClientCodecTest {
System.out.println("sncp.realBuf = " + realBuf.remaining());
codec.decodeMessages(realBuf, new ByteArray());
System.out.println("respResults.size = " + respResults.size());
if (!main) {
Assertions.assertEquals(2, respResults.size());
}
Assertions.assertEquals(2, respResults.size());
}
}

8
src/test/java/org/redkale/test/sncp/SncpTest.java
View File

@@ -14,7 +14,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.redkale.boot.*;
import org.redkale.convert.bson.*;
import org.redkale.net.*;
import org.redkale.net.sncp.*;
import org.redkale.net.sncp.SncpServer;
import org.redkale.service.Service;
import org.redkale.util.*;
@@ -76,7 +76,7 @@ public class SncpTest {
asyncGroup.start();
final TransportFactory transFactory = TransportFactory.create(asyncGroup, protocol.endsWith(".UDP") ? "UDP" : "TCP", 0, 0);
transFactory.addGroupInfo("client", set);
final SncpTestIService service = Sncp.createSimpleRemoteService(SncpTestIService.class, null, transFactory, addr, "client");
final SncpTestIService service = null;//Sncp.createSimpleRemoteService(SncpTestIService.class, null, transFactory, addr, "client");
factory.inject(service);
// SncpTestBean bean = new SncpTestBean();
@@ -171,7 +171,7 @@ public class SncpTest {
}
final TransportFactory transFactory = TransportFactory.create(asyncGroup, protocol.endsWith(".UDP") ? "UDP" : "TCP", 0, 0);
transFactory.addGroupInfo("server", set);
SncpTestIService service = Sncp.createSimpleLocalService(SncpTestServiceImpl.class, null, factory, transFactory, addr, "server");
SncpTestIService service = null;//Sncp.createSimpleLocalService(SncpTestServiceImpl.class, null, factory, transFactory, addr, "server");
factory.inject(service);
server.addSncpServlet(service);
System.out.println(service);
@@ -211,7 +211,7 @@ public class SncpTest {
final TransportFactory transFactory = TransportFactory.create(asyncGroup, protocol.endsWith(".UDP") ? "UDP" : "TCP", 0, 0);
transFactory.addGroupInfo("server", set);
Service service = Sncp.createSimpleLocalService(SncpTestServiceImpl.class, null, factory, transFactory, addr, "server");
Service service = null;//Sncp.createSimpleLocalService(SncpTestServiceImpl.class, null, factory, transFactory, addr, "server");
server.addSncpServlet(service);
server.init(conf);
server.start();

7
src/test/java/org/redkale/test/sncp/SncpTestServiceImpl.java
View File

@@ -13,7 +13,6 @@ import org.redkale.annotation.ResourceType;
import org.redkale.net.*;
import org.redkale.net.sncp.Sncp;
import org.redkale.service.*;
import org.redkale.util.ResourceFactory;
/**
*
@@ -89,7 +88,7 @@ public class SncpTestServiceImpl implements SncpTestIService {
final TransportFactory transFactory = TransportFactory.create(asyncGroup, 0, 0);
transFactory.addGroupInfo("g70", new InetSocketAddress("127.0.0.1", 7070));
Service service = Sncp.createSimpleLocalService(SncpTestServiceImpl.class, null, ResourceFactory.create(), transFactory, new InetSocketAddress("127.0.0.1", 7070), "g70");
Service service = null;// Sncp.createSimpleLocalService(SncpTestServiceImpl.class, null, ResourceFactory.create(), transFactory, new InetSocketAddress("127.0.0.1", 7070), "g70");
for (Method method : service.getClass().getDeclaredMethods()) {
System.out.println(method);
}
@@ -98,7 +97,7 @@ public class SncpTestServiceImpl implements SncpTestIService {
System.out.println(method);
}
System.out.println("-----------------------------------");
service = Sncp.createSimpleRemoteService(SncpTestServiceImpl.class, null, transFactory, new InetSocketAddress("127.0.0.1", 7070), "g70");
service = null;//Sncp.createSimpleRemoteService(SncpTestServiceImpl.class, null, transFactory, new InetSocketAddress("127.0.0.1", 7070), "g70");
for (Method method : service.getClass().getDeclaredMethods()) {
System.out.println(method);
}
@@ -107,7 +106,7 @@ public class SncpTestServiceImpl implements SncpTestIService {
System.out.println(method);
}
System.out.println("-----------------------------------");
service = Sncp.createSimpleRemoteService(SncpTestIService.class, null, transFactory, new InetSocketAddress("127.0.0.1", 7070), "g70");
service = null;//Sncp.createSimpleRemoteService(SncpTestIService.class, null, transFactory, new InetSocketAddress("127.0.0.1", 7070), "g70");
for (Method method : service.getClass().getDeclaredMethods()) {
System.out.println(method);
}

2
src/test/java/org/redkale/test/sncp/TestService.java
View File

@@ -8,8 +8,8 @@ import java.nio.channels.CompletionHandler;
import java.util.concurrent.CompletableFuture;
import org.redkale.annotation.ResourceType;
import org.redkale.convert.*;
import org.redkale.net.sncp.SncpDynServlet.SncpActionServlet;
import org.redkale.net.sncp.*;
import org.redkale.net.sncp.SncpServlet.SncpActionServlet;
import org.redkale.service.Service;
import org.redkale.test.util.TestBean;
import org.redkale.util.Uint128;