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

优化sncp

Browse Source
This commit is contained in:
redkale
2023-01-30 16:32:36 +08:00
parent 7dc57c67ed
commit 6d1cf4b6bf
16 changed files with 138 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/org/redkale/boot/Application.java
View File

@@ -612,7 +612,7 @@ public final class Application {
DefaultAnyValue tarnsportConf = DefaultAnyValue.create(TransportFactory.NAME_POOLMAXCONNS, System.getProperty("redkale.net.transport.pool.maxconns", "100"))
.addValue(TransportFactory.NAME_PINGINTERVAL, System.getProperty("redkale.net.transport.ping.interval", "30"))
.addValue(TransportFactory.NAME_CHECKINTERVAL, System.getProperty("redkale.net.transport.check.interval", "30"));
this.sncpTransportFactory.init(tarnsportConf, Sncp.PING_BUFFER, Sncp.PONG_BUFFER.remaining());
this.sncpTransportFactory.init(tarnsportConf, ByteBuffer.wrap(Sncp.getPingBytes()).asReadOnlyBuffer(), Sncp.getPongBytes().length);
this.clusterAgent = cluster;
this.messageAgents = mqs;
if (compileMode || this.classLoader instanceof RedkaleClassLoader.RedkaleCacheClassLoader) {

2
src/main/java/org/redkale/convert/ConvertFactory.java
View File

@@ -1102,6 +1102,8 @@ public abstract class ConvertFactory<R extends Reader, W extends Writer> {
decoder = createStreamDecoder(type);
} else if (Map.class.isAssignableFrom(clazz)) {
decoder = createMapDecoder(type);
} else if (CompletionHandler.class.isAssignableFrom(clazz)) {
decoder = CompletionHandlerSimpledCoder.instance;
} else if (Optional.class == clazz) {
decoder = new OptionalCoder(this, type);
} else if (clazz == Object.class) {

6
src/main/java/org/redkale/mq/MessageRecord.java
View File

@@ -12,7 +12,7 @@ import org.redkale.convert.*;
import org.redkale.convert.bson.BsonConvert;
import org.redkale.convert.json.JsonConvert;
import org.redkale.net.http.HttpSimpleRequest;
import org.redkale.net.sncp.SncpRequest;
import org.redkale.net.sncp.Sncp;
/**
* 存在MQ里面的数据结构<p>
@@ -318,8 +318,8 @@ public class MessageRecord implements Serializable {
sb.append(",\"respTopic\":\"").append(this.respTopic).append("\"");
}
if (this.content != null) {
if (this.ctype == CTYPE_BSON_RESULT && this.content.length > SncpRequest.HEADER_SIZE) {
int offset = SncpRequest.HEADER_SIZE + 1; //循环占位符
if (this.ctype == CTYPE_BSON_RESULT && this.content.length > Sncp.HEADER_SIZE) {
int offset = Sncp.HEADER_SIZE + 1; //循环占位符
Object rs = BsonConvert.root().convertFrom(Object.class, this.content, offset, this.content.length - offset);
sb.append(",\"content\":").append(rs);
} else if (this.ctype == CTYPE_HTTP_REQUEST) {

5
src/main/java/org/redkale/mq/SncpMessageResponse.java
View File

@@ -6,7 +6,6 @@
package org.redkale.mq;
import org.redkale.convert.bson.BsonWriter;
import static org.redkale.net.sncp.SncpRequest.HEADER_SIZE;
import org.redkale.net.sncp.*;
import org.redkale.util.ByteArray;
@@ -51,14 +50,14 @@ public class SncpMessageResponse extends SncpResponse {
callback.run();
}
if (out == null) {
final ByteArray result = new ByteArray(SncpRequest.HEADER_SIZE);
final ByteArray result = new ByteArray(Sncp.HEADER_SIZE);
fillHeader(result, 0, retcode);
producer.apply(messageClient.createMessageRecord(message.getSeqid(), message.getRespTopic(), null, (byte[]) null));
return;
}
final int respBodyLength = out.count(); //body总长度
final ByteArray result = out.toByteArray();
fillHeader(result, respBodyLength - HEADER_SIZE, retcode);
fillHeader(result, respBodyLength - Sncp.HEADER_SIZE, retcode);
producer.apply(messageClient.createMessageRecord(message.getSeqid(), message.getRespTopic(), null, result.getBytes()));
}
}

24
src/main/java/org/redkale/net/sncp/Sncp.java
View File

@@ -11,7 +11,6 @@ import static java.lang.annotation.RetentionPolicy.RUNTIME;
import java.lang.annotation.*;
import java.lang.reflect.*;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.CompletionHandler;
import java.util.*;
import org.redkale.annotation.*;
@@ -38,9 +37,20 @@ import org.redkale.util.*;
*/
public abstract class Sncp {
public static final ByteBuffer PING_BUFFER = ByteBuffer.wrap("PING".getBytes()).asReadOnlyBuffer();
public static final int HEADER_SIZE = 60;
public static final ByteBuffer PONG_BUFFER = ByteBuffer.wrap("PONG".getBytes()).asReadOnlyBuffer();
private static final byte[] PING_BYTES = new ByteArray(HEADER_SIZE)
.putLong(0L) //8 seqid
.putChar((char) HEADER_SIZE) //2 headerSize
.putUint128(Uint128.ZERO) //16 serviceid
.putInt(0) //4 serviceVersion
.putUint128(Uint128.ZERO) //16 actionid
.put(new byte[6]) //6 addr
.putInt(0) //4 bodyLength
.putInt(0) //4 retcode
.getBytes();
private static final byte[] PONG_BYTES = Arrays.copyOf(PING_BYTES, PING_BYTES.length);
static final String FIELDPREFIX = "_redkale";
@@ -70,6 +80,14 @@ public abstract class Sncp {
private Sncp() {
}
public static byte[] getPingBytes() {
return Arrays.copyOf(PING_BYTES, PING_BYTES.length);
}
public static byte[] getPongBytes() {
return Arrays.copyOf(PONG_BYTES, PONG_BYTES.length);
}
public static Uint128 actionid(final RpcAction action) {
return hash(action.name());
}

4
src/main/java/org/redkale/net/sncp/SncpClient.java
View File

@@ -18,6 +18,7 @@ import org.redkale.convert.bson.*;
import org.redkale.convert.json.*;
import org.redkale.mq.*;
import org.redkale.net.*;
import static org.redkale.net.sncp.Sncp.HEADER_SIZE;
import org.redkale.net.sncp.Sncp.SncpDyn;
import static org.redkale.net.sncp.SncpRequest.*;
import static org.redkale.net.sncp.SncpResponse.fillRespHeader;
@@ -62,9 +63,6 @@ public final class SncpClient {
protected final String topic;
@Resource
protected JsonConvert jsonConvert;
@Resource
protected BsonConvert bsonConvert;

4
src/main/java/org/redkale/net/sncp/SncpDispatcherServlet.java
View File

@@ -21,6 +21,8 @@ public class SncpDispatcherServlet extends DispatcherServlet<Uint128, SncpContex
private final Object sncplock = new Object();
private final byte[] pongBytes = Sncp.getPongBytes();
@Override
public void addServlet(SncpServlet servlet, Object attachment, AnyValue conf, Uint128... mappings) {
synchronized (sncplock) {
@@ -70,7 +72,7 @@ public class SncpDispatcherServlet extends DispatcherServlet<Uint128, SncpContex
@Override
public void execute(SncpRequest request, SncpResponse response) throws IOException {
if (request.isPing()) {
response.finishBuffer(false, Sncp.PONG_BUFFER.duplicate());
response.finish(pongBytes);
return;
}
SncpServlet servlet = (SncpServlet) mappingServlet(request.getServiceid());

20
src/main/java/org/redkale/net/sncp/SncpRequest.java
View File

@@ -11,6 +11,7 @@ import java.nio.ByteBuffer;
import java.util.logging.*;
import org.redkale.convert.bson.BsonConvert;
import org.redkale.net.Request;
import static org.redkale.net.sncp.Sncp.HEADER_SIZE;
import org.redkale.util.Uint128;
/**
@@ -22,8 +23,6 @@ import org.redkale.util.Uint128;
*/
public class SncpRequest extends Request<SncpContext> {
public static final int HEADER_SIZE = 60;
public static final byte[] DEFAULT_HEADER = new byte[HEADER_SIZE];
protected static final int READ_STATE_ROUTE = 1;
@@ -63,14 +62,6 @@ public class SncpRequest extends Request<SncpContext> {
@Override //request.header与response.header数据格式保持一致
protected int readHeader(ByteBuffer buffer, Request last) {
if (buffer.remaining() == Sncp.PING_BUFFER.remaining()) {
if (buffer.hasRemaining()) {
buffer.get(new byte[buffer.remaining()]);
}
this.ping = true; //Sncp.PING_BUFFER
this.readState = READ_STATE_END;
return 0;
}
//---------------------head----------------------------------
if (this.readState == READ_STATE_ROUTE) {
if (buffer.remaining() < HEADER_SIZE) {
@@ -89,7 +80,7 @@ public class SncpRequest extends Request<SncpContext> {
buffer.get(addrBytes); //ipaddr //6
this.bodyLength = buffer.getInt(); //4
if (buffer.getInt() != 0) { //4
if (buffer.getInt() != 0) { //4 retcode
if (context.getLogger().isLoggable(Level.FINEST)) {
context.getLogger().finest("sncp buffer header.retcode not 0");
}
@@ -100,6 +91,13 @@ public class SncpRequest extends Request<SncpContext> {
}
//---------------------body----------------------------------
if (this.readState == READ_STATE_BODY) {
if (this.bodyLength == 0) {
this.readState = READ_STATE_END;
if (this.seqid == 0 && this.serviceid == Uint128.ZERO && this.actionid == Uint128.ZERO) {
this.ping = true;
}
return 0;
}
int len = Math.min(this.bodyLength, buffer.remaining());
buffer.get(body, 0, len);
this.bodyOffset = len;

16
src/main/java/org/redkale/net/sncp/SncpResponse.java
View File

@@ -5,10 +5,9 @@
*/
package org.redkale.net.sncp;
import java.nio.ByteBuffer;
import org.redkale.convert.bson.BsonWriter;
import org.redkale.net.Response;
import static org.redkale.net.sncp.SncpRequest.HEADER_SIZE;
import static org.redkale.net.sncp.Sncp.HEADER_SIZE;
import org.redkale.util.*;
/**
@@ -67,14 +66,9 @@ public class SncpResponse extends Response<SncpContext, SncpRequest> {
return super.recycle();
}
@Override
protected void finishBuffer(boolean kill, ByteBuffer buffer) {
super.finishBuffer(kill, buffer);
}
public void finish(final int retcode, final BsonWriter out) {
if (out == null) {
final ByteArray buffer = new ByteArray(SncpRequest.HEADER_SIZE);
final ByteArray buffer = new ByteArray(HEADER_SIZE);
fillHeader(buffer, 0, retcode);
finish(buffer);
return;
@@ -96,13 +90,13 @@ public class SncpResponse extends Response<SncpContext, SncpRequest> {
int offset = 0;
buffer.putLong(offset, seqid);
offset += 8;
buffer.putChar(offset, (char) SncpRequest.HEADER_SIZE);
buffer.putChar(offset, (char) HEADER_SIZE);
offset += 2;
Uint128.write(buffer, offset, serviceid);
buffer.putUint128(offset, serviceid);
offset += 16;
buffer.putInt(offset, serviceVersion);
offset += 4;
Uint128.write(buffer, offset, actionid);
buffer.putUint128(offset, actionid);
offset += 16;
buffer.put(offset, addrBytes);
offset += addrBytes.length; //4

39
src/main/java/org/redkale/util/ByteArray.java
View File

@@ -250,6 +250,22 @@ public final class ByteArray implements ByteTuple {
return Double.longBitsToDouble(getLong(offset));
}
/**
* 获取指定位置的Uint128值,须确保0 &lt;= offset+16 &lt; length
*
* @param offset 位置
*
* @return Uint128值
*/
public Uint128 getUint128(int offset) {
return Uint128.create(new byte[]{
content[offset], content[offset + 1], content[offset + 2], content[offset + 3],
content[offset + 4], content[offset + 5], content[offset + 6], content[offset + 7],
content[offset + 8], content[offset + 9], content[offset + 10], content[offset + 11],
content[offset + 12], content[offset + 13], content[offset + 14], content[offset + 15]
});
}
/**
* 获取最后一个字节值,调用前须保证count大于0
*
@@ -751,6 +767,29 @@ public final class ByteArray implements ByteTuple {
return this.putLong(offset, Double.doubleToLongBits(value));
}
/**
* 写入一个Uint128值
*
* @param value Uint128值
*
* @return ByteArray
*/
public ByteArray putUint128(Uint128 value) {
return this.put(value.directBytes());
}
/**
* 指定位置写入一个Uint128值， content.length 必须不能小于offset+16
*
* @param offset 偏移量
* @param value Uint128值
*
* @return ByteArray
*/
public ByteArray putUint128(int offset, Uint128 value) {
return this.put(offset, value.directBytes());
}
public ByteArray putByte(short value) {
return put((byte) value);
}

28
src/main/java/org/redkale/util/Uint128.java
View File

@@ -29,7 +29,7 @@ public final class Uint128 extends Number implements Comparable<Uint128> {
(byte) (v2 >> 24), (byte) (v2 >> 16), (byte) (v2 >> 8), (byte) v2};
}
protected Uint128(byte[] bytes) {
private Uint128(byte[] bytes) {
if (bytes == null || bytes.length != 16) {
throw new NumberFormatException("Not 16 length bytes");
}
@@ -44,13 +44,25 @@ public final class Uint128 extends Number implements Comparable<Uint128> {
return value;
}
public static Uint128 create(byte[] bytes) {
return new Uint128(bytes);
public static Uint128 create(byte[] bs) {
if (bs[15] == 0 && bs[14] == 0 && bs[13] == 0 && bs[12] == 0
&& bs[11] == 0 && bs[10] == 0 && bs[9] == 0 && bs[8] == 0
&& bs[7] == 0 && bs[6] == 0 && bs[5] == 0 && bs[4] == 0
&& bs[3] == 0 && bs[2] == 0 && bs[1] == 0 && bs[0] == 0) {
return ZERO;
}
return new Uint128(bs);
}
public static Uint128 read(ByteBuffer buffer) {
byte[] bs = new byte[16];
buffer.get(bs);
if (bs[15] == 0 && bs[14] == 0 && bs[13] == 0 && bs[12] == 0
&& bs[11] == 0 && bs[10] == 0 && bs[9] == 0 && bs[8] == 0
&& bs[7] == 0 && bs[6] == 0 && bs[5] == 0 && bs[4] == 0
&& bs[3] == 0 && bs[2] == 0 && bs[1] == 0 && bs[0] == 0) {
return ZERO;
}
return new Uint128(bs);
}
@@ -59,16 +71,6 @@ public final class Uint128 extends Number implements Comparable<Uint128> {
return buffer;
}
public static ByteArray write(ByteArray array, Uint128 dlong) {
array.put(dlong.value);
return array;
}
public static ByteArray write(ByteArray array, int offset, Uint128 dlong) {
array.put(offset, dlong.value);
return array;
}
public boolean equals(byte[] bytes) {
return Arrays.equals(this.value, bytes);
}

3
src/test/java/org/redkale/test/net/TransportTest.java
View File

@@ -6,6 +6,7 @@
package org.redkale.test.net;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import org.redkale.net.*;
@@ -41,7 +42,7 @@ public class TransportTest {
Thread.sleep(1000);
TransportFactory factory = TransportFactory.create(asyncGroup, 0, 0);
DefaultAnyValue conf = DefaultAnyValue.create(TransportFactory.NAME_PINGINTERVAL, 5);
factory.init(conf, Sncp.PING_BUFFER, Sncp.PONG_BUFFER.remaining());
factory.init(conf, ByteBuffer.wrap(Sncp.getPingBytes()).asReadOnlyBuffer(), Sncp.getPingBytes().length);
Transport transport = factory.createTransportTCP("", null, addrs);
System.out.println(String.format(format, System.currentTimeMillis()));
try {

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

@@ -5,14 +5,14 @@
*/
package org.redkale.test.sncp;
import java.io.*;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.logging.LogManager;
import org.redkale.boot.LoggingBaseHandler;
import org.redkale.convert.bson.*;
import org.redkale.net.*;
import org.redkale.net.sncp.*;
@@ -33,22 +33,18 @@ public class SncpTest {
private static final int port2 = 4240;
private static final String protocol = "SNCP.UDP";
private static final String protocol = "SNCP.TCP";
private static final ResourceFactory factory = ResourceFactory.create();
public static void main(String[] args) throws Exception {
ByteArrayOutputStream out = new ByteArrayOutputStream();
final PrintStream ps = new PrintStream(out);
ps.println("handlers = java.util.logging.ConsoleHandler");
ps.println(".handlers = java.util.logging.ConsoleHandler");
ps.println(".level = FINEST");
ps.println("java.util.logging.ConsoleHandler.level = FINEST");
LogManager.getLogManager().readConfiguration(new ByteArrayInputStream(out.toByteArray()));
LoggingBaseHandler.initDebugLogConfig();
factory.register("", BsonConvert.class, BsonFactory.root().getConvert());
if (System.getProperty("client") == null) {
runServer();
if (port2 > 0) runServer2();
if (port2 > 0) {
runServer2();
}
}
if (System.getProperty("server") == null) {
runClient();
@@ -72,7 +68,9 @@ public class SncpTest {
public static ObjectPool<ByteBuffer> newBufferPool() {
return ObjectPool.createSafePool(new LongAdder(), new LongAdder(), 16,
(Object... params) -> ByteBuffer.allocateDirect(8192), null, (e) -> {
if (e == null || e.isReadOnly() || e.capacity() != 8192) return false;
if (e == null || e.isReadOnly() || e.capacity() != 8192) {
return false;
}
e.clear();
return true;
});
@@ -82,7 +80,9 @@ public class SncpTest {
InetSocketAddress addr = new InetSocketAddress(myhost, port);
Set<InetSocketAddress> set = new LinkedHashSet<>();
set.add(addr);
if (port2 > 0) set.add(new InetSocketAddress(myhost, port2));
if (port2 > 0) {
set.add(new InetSocketAddress(myhost, port2));
}
final AsyncIOGroup asyncGroup = new AsyncIOGroup(8192, 16);
asyncGroup.start();
final TransportFactory transFactory = TransportFactory.create(asyncGroup, protocol.endsWith(".UDP") ? "UDP" : "TCP", 0, 0);
@@ -102,25 +102,26 @@ public class SncpTest {
callbean.setContent("数据X");
service.queryLongResult("f", 3, 33L);
service.insert(callbean);
System.out.println("bean.id应该会被修改(id不会是1)： " + callbean);
callbean = service.insert(callbean);
System.out.println("bean： " + callbean);
System.out.println("---------------------------------------------------");
final int count = 10;
final CountDownLatch cld = new CountDownLatch(count);
final AtomicInteger ai = new AtomicInteger();
long s = System.currentTimeMillis();
for (int i = 0; i < count; i++) {
final int k = i + 1;
new Thread() {
@Override
public void run() {
try {
Thread.sleep(k);
//Thread.sleep(k);
SncpTestBean bean = new SncpTestBean();
bean.setId(k);
bean.setContent("数据: " + (k < 10 ? "0" : "") + k);
StringBuilder sb = new StringBuilder();
sb.append(k).append("------");
for (int i = 0; i < 1200; i++) {
for (int i = 0; i < 12; i++) {
sb.append("_").append(i).append("_").append(k).append("_0123456789");
}
bean.setContent(sb.toString());
@@ -138,11 +139,13 @@ public class SncpTest {
}.start();
}
cld.await();
System.out.println("---并发" + count + "次耗时: " + (System.currentTimeMillis() - s) / 1000.0 + "s");
final CountDownLatch cld2 = new CountDownLatch(1);
long s2 = System.currentTimeMillis();
final CompletableFuture<String> future = service.queryResultAsync(callbean);
future.whenComplete((v, e) -> {
System.out.println("异步执行完毕: " + v + ", 异常为: " + e + ", 耗时: " + (System.currentTimeMillis() - s2) / 1000.0 + "s");
cld2.countDown();
System.out.println("异步执行完毕: " + v + ", 异常为: " + e);
});
cld2.await();
System.out.println("---全部运行完毕---");
@@ -168,7 +171,9 @@ public class SncpTest {
conf.addValue("protocol", protocol);
SncpServer server = new SncpServer(null, System.currentTimeMillis(), conf, factory);
Set<InetSocketAddress> set = new LinkedHashSet<>();
if (port2 > 0) set.add(new InetSocketAddress(myhost, port2));
if (port2 > 0) {
set.add(new InetSocketAddress(myhost, port2));
}
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");

11
src/test/java/org/redkale/test/sncp/SncpTestBean.java
View File

@@ -6,10 +6,10 @@
package org.redkale.test.sncp;
import org.redkale.convert.bson.BsonFactory;
import org.redkale.convert.json.JsonConvert;
import org.redkale.persistence.Id;
import org.redkale.util.Utility;
import org.redkale.source.FilterBean;
import org.redkale.convert.json.*;
import org.redkale.util.Utility;
/**
*
@@ -23,13 +23,14 @@ public class SncpTestBean implements FilterBean {
private String content;
public static void main(String[] args) throws Exception {
SncpTestBean bean = JsonConvert.root().convertFrom(SncpTestBean.class, "{\"content\":\"数据: 01\",\"id\":1}");
String json = "{\"content\":\"数据: 01\",\"id\":1}";
SncpTestBean bean = JsonConvert.root().convertFrom(SncpTestBean.class, json);
System.out.println(bean);
byte[] bs = BsonFactory.root().getConvert().convertTo(bean);
Utility.println("---------", bs);
Utility.println("---------", bs);
System.out.println(BsonFactory.root().getConvert().convertFrom(SncpTestBean.class, bs).toString());
}
@Override
public String toString() {
return JsonConvert.root().convertTo(this);

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

@@ -22,7 +22,7 @@ public interface SncpTestIService extends Service {
public CompletableFuture<String> queryResultAsync(SncpTestBean bean);
public void insert(SncpTestBean... beans);
public SncpTestBean insert(SncpTestBean bean);
public String updateBean(SncpTestBean bean);
}

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

@@ -30,7 +30,7 @@ public class SncpTestServiceImpl implements SncpTestIService {
public void run() {
try {
Thread.sleep(1000);
System.out.println(Thread.currentThread().getName() + " 运行了异步方法-----------queryResultAsync方法");
System.out.println(Thread.currentThread().getName() + " sleep 1秒后运行了异步方法-----------queryResultAsync方法");
future.complete("异步result: " + bean);
} catch (Exception e) {
e.printStackTrace();
@@ -83,10 +83,9 @@ public class SncpTestServiceImpl implements SncpTestIService {
}
@Override
public void insert(SncpTestBean... beans) {
for (SncpTestBean bean : beans) {
bean.setId(System.currentTimeMillis());
}
public SncpTestBean insert(SncpTestBean bean) {
bean.setId(System.currentTimeMillis());
return bean;
}
@Override