autumn/jsowell-charger-web
1
0
Fork 0
mirror of https://codeup.aliyun.com/67c68d4e484ca2f0a13ac3c1/ydc/jsowell-charger-web.git synced 2026-04-20 19:15:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

同步获取响应数据

Browse Source
This commit is contained in:
Guoqs
2024-08-01 18:03:21 +08:00
parent eb40bd5798
commit 64028f04f3
20 changed files with 99 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
jsowell-netty/src/main/java/com/jsowell/netty/rpc/ClientChannelInitializer.java
View File

@@ -1,19 +0,0 @@
package com.jsowell.netty.rpc;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new MessageEncode());
pipeline.addLast(new MessageDecode());
pipeline.addLast(new RpcResponseHandler());
}
}

11
jsowell-netty/src/main/java/com/jsowell/netty/rpc/Message.java
View File

@@ -1,11 +0,0 @@
package com.jsowell.netty.rpc;
import lombok.Data;
@Data
public abstract class Message {
protected Byte messageType;
}

22
jsowell-netty/src/main/java/com/jsowell/netty/rpc/MessageConstant.java
View File

@@ -1,22 +0,0 @@
package com.jsowell.netty.rpc;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class MessageConstant {
public final static Byte rpcRequest = 1;
public final static Byte rpcResponse = 2;
public static Map<Byte, Class<? extends Message>> messageTypeMap = new ConcurrentHashMap<>();
static {
messageTypeMap.put(rpcRequest, RpcRequest.class);
messageTypeMap.put(rpcResponse, RpcResponse.class);
}
public static Class<? extends Message> getMessageClass(Byte messageType){
return messageTypeMap.get(messageType);
}
}

45
jsowell-netty/src/main/java/com/jsowell/netty/rpc/MessageDecode.java
View File

@@ -1,45 +0,0 @@
package com.jsowell.netty.rpc;
import com.jsowell.common.util.bean.SerializationUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import java.util.List;
public class MessageDecode extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
// 由于数据包的前4个字节用于记录总数据大小，如果数据不够4个字节，不进行读
if(byteBuf.readableBytes() < 4) {
return;
}
// 标记开始读的位置
byteBuf.markReaderIndex();
// 前四个字节记录了数据大小
int dataSize = byteBuf.readInt();
// 查看剩余可读字节是否足够，如果不是，重置读取位置，等待下一次解析
if(byteBuf.readableBytes() < dataSize) {
byteBuf.resetReaderIndex();
return;
}
// 读取消息类型
byte messageType = byteBuf.readByte();
// 读取数据, 数组大小需要剔除1个字节的消息类型
byte[] data = new byte[dataSize -1];
byteBuf.readBytes(data);
Message message = SerializationUtil.deserialize(MessageConstant.getMessageClass(messageType), data);
list.add(message);
}
}

23
jsowell-netty/src/main/java/com/jsowell/netty/rpc/MessageEncode.java
View File

@@ -1,23 +0,0 @@
package com.jsowell.netty.rpc;
import com.jsowell.common.util.bean.SerializationUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
public class MessageEncode extends MessageToByteEncoder<Message> {
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf byteBuf) throws Exception {
// 将对象进行序列化
byte[] data = SerializationUtil.serialize(message);
// 写数据长度，前4个字节用于记录数据总长度（对象 + 类型（1个字节））
byteBuf.writeInt(data.length + 1);
// 写记录消息类型，用于反序列选择类的类型
byteBuf.writeByte(message.getMessageType());
// 写对象
byteBuf.writeBytes(data);
}
}

52
jsowell-netty/src/main/java/com/jsowell/netty/rpc/RpcClient.java
View File

@@ -1,52 +0,0 @@
package com.jsowell.netty.rpc;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
public class RpcClient {
public Channel connect(String host, Integer port) {
EventLoopGroup worker = new NioEventLoopGroup();
Channel channel = null;
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(worker)
.channel(NioSocketChannel.class)
.option(ChannelOption.AUTO_READ, true)
.handler(new ClientChannelInitializer());
ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
System.out.println("客户端启动");
channel = channelFuture.channel();
// 添加关闭监听器
channel.closeFuture().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
System.out.println("关闭客户端");
worker.shutdownGracefully();
}
});
} catch (Exception e) {
e.printStackTrace();
if(channel == null || !channel.isActive()) {
worker.shutdownGracefully();
} else {
channel.close();
}
}
return channel;
}
}

21
jsowell-netty/src/main/java/com/jsowell/netty/rpc/RpcRequest.java
View File

@@ -1,21 +0,0 @@
package com.jsowell.netty.rpc;
import lombok.Data;
import lombok.ToString;
import java.util.UUID;
@Data
@ToString
public class RpcRequest extends Message{
private String id;
private String param;
public RpcRequest() {
this.id = UUID.randomUUID().toString();
super.messageType = MessageConstant.rpcRequest;
}
}

36
jsowell-netty/src/main/java/com/jsowell/netty/rpc/RpcRequestHandler.java
View File

@@ -1,36 +0,0 @@
package com.jsowell.netty.rpc;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.DefaultEventLoopGroup;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
public class RpcRequestHandler extends SimpleChannelInboundHandler<RpcRequest> {
private final static EventLoopGroup worker = new DefaultEventLoopGroup(Runtime.getRuntime().availableProcessors() + 1);
@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) throws Exception {
// 为避免占用网络io，此处异步进行处理
worker.submit(() -> {
System.out.println("[RpcRequestHandler] "+ Thread.currentThread().getName() +" 处理请求，msg： " + msg);
// 模拟处理耗时
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
RpcResponse rpcResponse = new RpcResponse();
rpcResponse.setId(msg.getId());
rpcResponse.setResult("处理" + msg.getParam());
ctx.writeAndFlush(rpcResponse);
});
}
}

18
jsowell-netty/src/main/java/com/jsowell/netty/rpc/RpcResponse.java
View File

@@ -1,18 +0,0 @@
package com.jsowell.netty.rpc;
import lombok.Data;
import lombok.ToString;
@Data
@ToString
public class RpcResponse extends Message{
private String id;
private String result;
public RpcResponse() {
super.messageType = MessageConstant.rpcResponse;
}
}

25
jsowell-netty/src/main/java/com/jsowell/netty/rpc/RpcResponseHandler.java
View File

@@ -1,25 +0,0 @@
package com.jsowell.netty.rpc;
import com.jsowell.pile.rpc.RpcUtil;
import com.jsowell.pile.rpc.SyncPromise;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
public class RpcResponseHandler extends SimpleChannelInboundHandler<RpcResponse> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {
// 根据请求id，在集合中找到与外部线程通信的SyncPromise对象
SyncPromise syncPromise = RpcUtil.getSyncPromiseMap().get(msg.getId());
if(syncPromise != null) {
// 设置响应结果
// syncPromise.setRpcResponse(msg);
// 唤醒外部线程
syncPromise.wake();
}
}
}

55
jsowell-netty/src/main/java/com/jsowell/netty/rpc/RpcServer.java
View File

@@ -1,55 +0,0 @@
package com.jsowell.netty.rpc;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class RpcServer {
public void bind(Integer port) {
EventLoopGroup parent = new NioEventLoopGroup();
EventLoopGroup child = new NioEventLoopGroup();
Channel channel = null;
try{
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(parent, child)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ServerChannelInitializer());
ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
System.out.println("server启动");
// 非阻塞等待关闭
channelFuture.channel().closeFuture().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
System.out.println("server关闭");
parent.shutdownGracefully();
child.shutdownGracefully();
}
});
channel = channelFuture.channel();
} catch (Exception e) {
e.printStackTrace();
if(channel == null || !channel.isActive()) {
System.out.println("server关闭");
parent.shutdownGracefully();
child.shutdownGracefully();
} else {
channel.close();
}
}
}
}

19
jsowell-netty/src/main/java/com/jsowell/netty/rpc/ServerChannelInitializer.java
View File

@@ -1,19 +0,0 @@
package com.jsowell.netty.rpc;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new MessageEncode());
pipeline.addLast(new MessageDecode());
pipeline.addLast(new RpcRequestHandler());
}
}

50
jsowell-netty/src/main/java/com/jsowell/netty/rpc/TestRpcClient.java
View File

@@ -1,50 +0,0 @@
package com.jsowell.netty.rpc;
import com.jsowell.pile.rpc.RpcUtil;
import java.util.concurrent.TimeUnit;
public class TestRpcClient {
public static void main(String[] args) throws Exception{
//Channel channel = new RpcClient().connect("127.0.0.1", 8888);
Thread thread1 = new Thread(new Runnable() {
@Override
public void run() {
RpcRequest rpcRequest = new RpcRequest();
rpcRequest.setParam("参数1");
try {
// System.out.println("thread1发送请求");
// RpcResponse rpcResponse = RpcUtil.send(rpcRequest, 5, TimeUnit.SECONDS);
// System.out.println("thread1处理结果：" + rpcResponse);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
});
Thread thread2 = new Thread(new Runnable() {
@Override
public void run() {
RpcRequest rpcRequest2 = new RpcRequest();
rpcRequest2.setParam("参数2");
try {
// System.out.println("thread2发送请求");
// RpcResponse rpcResponse = RpcUtil.send(rpcRequest2, 2, TimeUnit.SECONDS);
// System.out.println("thread2处理结果：" + rpcResponse);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
});
// 休眠一下，等待客户端与服务端进行连接
Thread.sleep(1000);
thread1.start();
thread2.start();
}
}

7
jsowell-netty/src/main/java/com/jsowell/netty/rpc/TestRpcServer.java
View File

@@ -1,7 +0,0 @@
package com.jsowell.netty.rpc;
public class TestRpcServer {
public static void main(String[] args) {
new RpcServer().bind(8888);
}
}

5
jsowell-netty/src/main/java/com/jsowell/netty/server/yunkuaichong/NettyServerHandler.java
View File

@@ -1,14 +1,13 @@
package com.jsowell.netty.server.yunkuaichong;
import com.alibaba.fastjson2.JSON;
import com.google.common.collect.Lists;
import com.jsowell.common.core.domain.ykc.YKCFrameTypeCode;
import com.jsowell.common.enums.ykc.PileChannelEntity;
import com.jsowell.common.util.BytesUtil;
import com.jsowell.common.util.StringUtils;
import com.jsowell.common.util.YKCUtils;
import com.jsowell.pile.rpc.RpcUtil;
import com.jsowell.pile.rpc.SyncPromise;
import com.jsowell.common.util.RpcUtil;
import com.jsowell.common.protocol.SyncPromise;
import com.jsowell.netty.service.yunkuaichong.YKCBusinessService;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;