tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-19 22:24:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat[net]: 增加Jprotobuf的服务器启动支持

Browse Source
This commit is contained in:
jaysunxiao
2022-01-12 11:28:38 +08:00
parent 1853254f00
commit 0f8870a794
12 changed files with 502 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/pom.xml
+13
View File
@@ -134,6 +134,19 @@
<version>${netty.version}</version>
</dependency>
<!-- Protobuf -->
<dependency>
<groupId>com.baidu</groupId>
<artifactId>jprotobuf</artifactId>
<version>${jprotobuf.version}</version>
<exclusions>
<exclusion>
<artifactId>slf4j-api</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<!-- 动态生成二进制字节码的javassist类库 -->
<dependency>
<groupId>org.javassist</groupId>
net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java
+53
View File
@@ -0,0 +1,53 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.core.jprotobuf;
import com.zfoo.net.core.AbstractClient;
import com.zfoo.net.handler.ClientRouteHandler;
import com.zfoo.net.handler.codec.jprotobuf.JProtobufTcpCodecHandler;
import com.zfoo.net.handler.idle.ClientIdleHandler;
import com.zfoo.util.net.HostAndPort;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
/**
* @author jaysunxiao
* @version 3.0
*/
public class JProtobufTcpClient extends AbstractClient {
public JProtobufTcpClient(HostAndPort host) {
super(host);
}
@Override
public ChannelInitializer<? extends Channel> channelChannelInitializer() {
return new ChannelHandlerInitializer();
}
private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel channel) {
channel.pipeline().addLast(new IdleStateHandler(0, 0, 60));
channel.pipeline().addLast(new ClientIdleHandler());
channel.pipeline().addLast(new JProtobufTcpCodecHandler());
channel.pipeline().addLast(new ClientRouteHandler());
}
}
}
net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTpcServer.java
+50
View File
@@ -0,0 +1,50 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.core.jprotobuf;
import com.zfoo.net.core.AbstractServer;
import com.zfoo.net.handler.ServerRouteHandler;
import com.zfoo.net.handler.codec.jprotobuf.JProtobufTcpCodecHandler;
import com.zfoo.net.handler.idle.ServerIdleHandler;
import com.zfoo.util.net.HostAndPort;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
/**
* @author jaysunxiao
* @version 3.0
*/
public class JProtobufTpcServer extends AbstractServer {
public JProtobufTpcServer(HostAndPort host) {
super(host);
}
@Override
public ChannelInitializer<SocketChannel> channelChannelInitializer() {
return new ChannelHandlerInitializer();
}
private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel channel) {
channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
channel.pipeline().addLast(new ServerIdleHandler());
channel.pipeline().addLast(new JProtobufTcpCodecHandler());
channel.pipeline().addLast(new ServerRouteHandler());
}
}
}
net/src/main/java/com/zfoo/net/handler/codec/jprotobuf/JProtobufTcpCodecHandler.java
+142
View File
@@ -0,0 +1,142 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.handler.codec.jprotobuf;
import com.baidu.bjf.remoting.protobuf.Codec;
import com.baidu.bjf.remoting.protobuf.ProtobufProxy;
import com.zfoo.net.packet.model.DecodedPacketInfo;
import com.zfoo.net.packet.model.EncodedPacketInfo;
import com.zfoo.net.packet.service.PacketService;
import com.zfoo.net.router.attachment.IAttachment;
import com.zfoo.net.util.SessionUtils;
import com.zfoo.protocol.IPacket;
import com.zfoo.protocol.ProtocolManager;
import com.zfoo.protocol.buffer.ByteBufUtils;
import com.zfoo.protocol.util.StringUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.List;
import static com.zfoo.protocol.ProtocolManager.protocols;
/**
* header(4byte) + protocolId(2byte) + packet
* header = body(bytes.length) + protocolId.length(2byte)
*
* @author jaysunxiao
* @version 3.0
*/
public class JProtobufTcpCodecHandler extends ByteToMessageCodec<EncodedPacketInfo> {
private static final Logger logger = LoggerFactory.getLogger(JProtobufTcpCodecHandler.class);
public static DecodedPacketInfo read(ByteBuf buffer) throws IOException {
var protocolId = ByteBufUtils.readShort(buffer);
var protocolRegistration = ProtocolManager.getProtocol(protocolId);
var protocolClass = protocols[protocolId].protocolConstructor().getDeclaringClass();
var protobufCodec = ProtobufProxy.create(protocolClass);
var bytes = ByteBufUtils.readAllBytes(buffer);
var packet = protobufCodec.decode(bytes);
// 解析包的附加包
var hasAttachment = ByteBufUtils.tryReadBoolean(buffer);
IAttachment attachment = null;
if (hasAttachment) {
}
return DecodedPacketInfo.valueOf((IPacket) packet, attachment);
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
// 不够读一个int
if (in.readableBytes() <= PacketService.PACKET_HEAD_LENGTH) {
return;
}
in.markReaderIndex();
var length = in.readInt();
// 如果长度非法，则抛出异常断开连接
if (length < 0) {
throw new IllegalArgumentException(StringUtils.format("[session:{}]的包头长度[length:{}]非法"
, SessionUtils.sessionInfo(ctx), length));
}
// ByteBuf里的数据太小
if (in.readableBytes() < length) {
in.resetReaderIndex();
return;
}
ByteBuf tmpByteBuf = null;
try {
tmpByteBuf = in.readRetainedSlice(length);
DecodedPacketInfo packetInfo = read(tmpByteBuf);
out.add(packetInfo);
} catch (Exception e) {
logger.error("[session:{}]解码exception异常", SessionUtils.sessionInfo(ctx), e);
} catch (Throwable t) {
logger.error("[session:{}]解码throwable错误", SessionUtils.sessionInfo(ctx), t);
} finally {
ReferenceCountUtil.release(tmpByteBuf);
}
}
@Override
protected void encode(ChannelHandlerContext ctx, EncodedPacketInfo packetInfo, ByteBuf out) {
try {
write(out, packetInfo.getPacket(), packetInfo.getAttachment());
} catch (Exception e) {
logger.error("[session:{}][{}]编码exception异常", SessionUtils.sessionInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), e);
} catch (Throwable t) {
logger.error("[session:{}][{}]编码throwable错误", SessionUtils.sessionInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), t);
}
}
public void write(ByteBuf buffer, IPacket packet, IAttachment attachment) throws IOException {
if (packet == null) {
logger.error("packet is null and can not be sent.");
return;
}
// 预留写入包的长度，一个int字节大小
buffer.writeInt(PacketService.PACKET_HEAD_LENGTH);
var protocolId = packet.protocolId();
// 写入协议号
ByteBufUtils.writeShort(buffer, protocolId);
var protobufCodec = (Codec<IPacket>) ProtobufProxy.create(packet.getClass());
byte[] bytes = protobufCodec.encode(packet);
buffer.writeBytes(bytes);
int length = buffer.readableBytes();
int packetLength = length - PacketService.PACKET_HEAD_LENGTH;
buffer.writerIndex(0);
buffer.writeInt(packetLength);
buffer.writerIndex(length);
}
}
net/src/main/java/com/zfoo/net/packet/service/PacketService.java
+1 -1
View File
@@ -134,7 +134,7 @@ public class PacketService implements IPacketService {
// 解析包体
var packet = ProtocolManager.read(buffer);
// 解析包的附加包
var hasAttachment = ByteBufUtils.readBoolean(buffer);
var hasAttachment = ByteBufUtils.tryReadBoolean(buffer);
var attachment = hasAttachment ? ((IAttachment) ProtocolManager.read(buffer)) : null;
return DecodedPacketInfo.valueOf(packet, attachment);
}
net/src/test/java/com/zfoo/net/core/jprotobuf/client/TcpClientTest.java
+56
View File
@@ -0,0 +1,56 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.core.jprotobuf.client;
import com.zfoo.net.NetContext;
import com.zfoo.net.core.jprotobuf.JProtobufTcpClient;
import com.zfoo.net.packet.jprotobuf.JProtobufHelloRequest;
import com.zfoo.net.packet.jprotobuf.JProtobufHelloResponse;
import com.zfoo.protocol.util.JsonUtils;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author jaysunxiao
* @version 3.0
*/
@Ignore
public class TcpClientTest {
private static final Logger logger = LoggerFactory.getLogger(TcpClientTest.class);
@Test
public void startClient() throws Exception {
var context = new ClassPathXmlApplicationContext("config.xml");
var client = new JProtobufTcpClient(HostAndPort.valueOf("127.0.0.1:9000"));
var session = client.start();
for (int i = 0; i < 1000; i++) {
var ask = new JProtobufHelloRequest();
ask.setMessage("Hello, this is jprotobuf client!");
var answer = NetContext.getRouter().syncAsk(session, ask, JProtobufHelloResponse.class, null).packet();
logger.info("同步请求收到结果[{}]", JsonUtils.object2String(answer));
ThreadUtils.sleep(1000);
}
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpController.java
+44
View File
@@ -0,0 +1,44 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed 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 com.zfoo.net.core.jprotobuf.server;
import com.zfoo.net.NetContext;
import com.zfoo.net.packet.jprotobuf.JProtobufHelloRequest;
import com.zfoo.net.packet.jprotobuf.JProtobufHelloResponse;
import com.zfoo.net.router.receiver.PacketReceiver;
import com.zfoo.net.session.model.Session;
import com.zfoo.protocol.util.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/**
* @author jaysunxiao
* @version 3.0
*/
@Component
public class JProtobufTcpController {
private static final Logger logger = LoggerFactory.getLogger(JProtobufTcpController.class);
@PacketReceiver
public void atJProtobufHelloRequest(Session session, JProtobufHelloRequest ask) {
logger.info("receive [packet:{}] from client", JsonUtils.object2String(ask));
var answer = new JProtobufHelloResponse();
answer.setMessage("Hello, this is the jprotobuf tcp server!");
NetContext.getRouter().send(session, answer);
}
}
net/src/test/java/com/zfoo/net/core/jprotobuf/server/JProtobufTcpServerTest.java
+40
View File
@@ -0,0 +1,40 @@
/*
* Copyright (C) 2020 The zfoo Authors
*
* Licensed 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 com.zfoo.net.core.jprotobuf.server;
import com.zfoo.net.core.jprotobuf.JProtobufTpcServer;
import com.zfoo.util.ThreadUtils;
import com.zfoo.util.net.HostAndPort;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author jaysunxiao
* @version 3.0
*/
@Ignore
public class JProtobufTcpServerTest {
@Test
public void startServer() {
var context = new ClassPathXmlApplicationContext("config.xml");
var server = new JProtobufTpcServer(HostAndPort.valueOf("127.0.0.1:9000"));
server.start();
ThreadUtils.sleep(Long.MAX_VALUE);
}
}
net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloRequest.java
+47
View File
@@ -0,0 +1,47 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed 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 com.zfoo.net.packet.jprotobuf;
import com.baidu.bjf.remoting.protobuf.annotation.Protobuf;
import com.zfoo.protocol.IPacket;
/**
* @author jaysunxiao
* @version 3.0
*/
public class JProtobufHelloRequest implements IPacket {
public static final transient short PROTOCOL_ID = 1500;
@Protobuf(order = 1)
private String message;
public static JProtobufHelloRequest valueOf(String message) {
var request = new JProtobufHelloRequest();
request.message = message;
return request;
}
@Override
public short protocolId() {
return PROTOCOL_ID;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
net/src/test/java/com/zfoo/net/packet/jprotobuf/JProtobufHelloResponse.java
+42
View File
@@ -0,0 +1,42 @@
/*
* Copyright (C) 2020 The zfoo Authors
* Licensed 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 com.zfoo.net.packet.jprotobuf;
import com.baidu.bjf.remoting.protobuf.annotation.Protobuf;
import com.zfoo.protocol.IPacket;
/**
* @author jaysunxiao
* @version 3.0
*/
public class JProtobufHelloResponse implements IPacket {
public static final transient short PROTOCOL_ID = 1501;
@Protobuf(order = 1)
private String message;
@Override
public short protocolId() {
return PROTOCOL_ID;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
net/src/test/resources/protocol.xml
+3
View File
@@ -66,6 +66,9 @@
<protocol id="1400" location="com.zfoo.net.packet.websocket.WebsocketHelloRequest"/>
<protocol id="1401" location="com.zfoo.net.packet.websocket.WebsocketHelloResponse"/>
<protocol id="1500" location="com.zfoo.net.packet.jprotobuf.JProtobufHelloRequest"/>
<protocol id="1501" location="com.zfoo.net.packet.jprotobuf.JProtobufHelloResponse"/>
</module>
<module id="4" name="js" minId="2000" maxId="3000" version="1.0.0">
protocol/src/main/java/com/zfoo/protocol/buffer/ByteBufUtils.java
+11
View File
@@ -46,6 +46,10 @@ public abstract class ByteBufUtils {
return byteBuf.readBoolean();
}
public static boolean tryReadBoolean(ByteBuf byteBuf) {
return byteBuf.isReadable() && readBoolean(byteBuf);
}
public static void writeBooleanBox(ByteBuf byteBuf, Boolean value) {
byteBuf.writeBoolean(value != null && value);
}
@@ -71,6 +75,13 @@ public abstract class ByteBufUtils {
return byteBuf.readByte();
}
public static byte[] readAllBytes(ByteBuf byteBuf) {
var readableBytes = byteBuf.readableBytes();
var bytes = new byte[readableBytes];
byteBuf.readBytes(bytes);
return bytes;
}
//---------------------------------short--------------------------------------
public static void writeShort(ByteBuf byteBuf, short value) {
byteBuf.writeShort(value);