parent
26ccc04ee0
commit
a27f0b24e3
@ -0,0 +1,73 @@
|
||||
package com.baiye.case5.client;
|
||||
|
||||
|
||||
import com.baiye.case5.client.codec.ClientProtocolDecoder;
|
||||
import com.baiye.case5.client.codec.ClientProtocolEncoder;
|
||||
import com.baiye.case5.client.codec.OrderFrameDecoder;
|
||||
import com.baiye.case5.client.codec.OrderFrameEncoder;
|
||||
import com.baiye.case5.client.handler.ClientHandler;
|
||||
import com.baiye.case5.common.RequestMessage;
|
||||
import com.baiye.case5.common.order.OrderOperation;
|
||||
import com.baiye.case5.util.IdUtil;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelPipeline;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
import io.netty.handler.logging.LoggingHandler;
|
||||
|
||||
/**
|
||||
* v1 版本 - 客户端
|
||||
*/
|
||||
public class OrderClientV0 {
|
||||
|
||||
public static void main(String[] args) {
|
||||
|
||||
NioEventLoopGroup boss = new NioEventLoopGroup();
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
final ClientHandler clientHandler = new ClientHandler();
|
||||
LoggingHandler loggingHandler = new LoggingHandler(LogLevel.INFO);
|
||||
|
||||
try {
|
||||
bootstrap.group(boss);
|
||||
bootstrap.channel(NioSocketChannel.class);
|
||||
// bootstrap.option(NioChannelOption.TCP_NODELAY, true);
|
||||
// 客户端连接服务端超时设置
|
||||
// bootstrap.option(NioChannelOption.CONNECT_TIMEOUT_MILLIS, 10 * 1000);
|
||||
bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(NioSocketChannel ch) throws Exception {
|
||||
ChannelPipeline p = ch.pipeline();
|
||||
|
||||
p.addLast(loggingHandler);
|
||||
|
||||
p.addLast(new OrderFrameDecoder());
|
||||
p.addLast(new OrderFrameEncoder());
|
||||
p.addLast(new ClientProtocolDecoder());
|
||||
p.addLast(new ClientProtocolEncoder());
|
||||
|
||||
// ... 这里是考虑客户端不需要业务处理的 handler
|
||||
p.addLast(clientHandler);
|
||||
}
|
||||
});
|
||||
|
||||
// 建立连接 - sync
|
||||
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8090).sync();
|
||||
// 构建消息体
|
||||
OrderOperation orderOperation = new OrderOperation(1, "fish");
|
||||
// 生成Id
|
||||
long id = IdUtil.nextId();
|
||||
// 发送消息
|
||||
channelFuture.channel().writeAndFlush(new RequestMessage(id, orderOperation));
|
||||
|
||||
channelFuture.channel().closeFuture().get();
|
||||
|
||||
}catch (Exception e){
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
boss.shutdownGracefully();
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,78 @@
|
||||
package com.baiye.case5.client;
|
||||
|
||||
|
||||
import com.baiye.case5.client.codec.*;
|
||||
import com.baiye.case5.client.handler.ClientHandler;
|
||||
import com.baiye.case5.client.handler.dispatcher.ResponseMessagePendingCenter;
|
||||
import com.baiye.case5.common.order.OrderOperation;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelPipeline;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
import io.netty.handler.logging.LoggingHandler;
|
||||
|
||||
/**
|
||||
* 引入 OperationToRequestMessageEncoder 来改进请求发送的封装
|
||||
*
|
||||
* @author q
|
||||
*/
|
||||
public class OrderClientV1 {
|
||||
|
||||
public static void main(String[] args) {
|
||||
final ClientHandler clientHandler = new ClientHandler();
|
||||
LoggingHandler loggingHandler = new LoggingHandler(LogLevel.INFO);
|
||||
final ResponseMessagePendingCenter center = new ResponseMessagePendingCenter();
|
||||
NioEventLoopGroup boss = new NioEventLoopGroup();
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
|
||||
try {
|
||||
bootstrap.group(boss);
|
||||
bootstrap.channel(NioSocketChannel.class);
|
||||
bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(NioSocketChannel ch) throws Exception {
|
||||
ChannelPipeline pipeline = ch.pipeline();
|
||||
// 编解码
|
||||
pipeline.addLast(new OrderFrameDecoder());
|
||||
pipeline.addLast(new OrderFrameEncoder());
|
||||
pipeline.addLast(new ClientProtocolDecoder());
|
||||
pipeline.addLast(new ClientProtocolEncoder());
|
||||
// 请求内容 Encode
|
||||
pipeline.addLast(new OperationToRequestMessageEncoder());
|
||||
// Netty 日志 - handler
|
||||
pipeline.addLast(loggingHandler);
|
||||
|
||||
// pipeline.addLast(new ResponseMessagePromiseHandler(center));
|
||||
// pipeline.addLast(clientHandler);
|
||||
}
|
||||
});
|
||||
// 建立连接 - sync
|
||||
ChannelFuture ch = bootstrap.connect("127.0.0.1", 8090).sync();
|
||||
// 因为上面封装了 OperationToRequestMessageEncoder - 这里直接可以用 OrderOperation 替代下面的 fix
|
||||
|
||||
// fixme
|
||||
/* RequestMessage requestMessage = new RequestMessage(IdUtil.nextId(), new OrderOperation(1002, "ji"));
|
||||
ch.channel().writeAndFlush(requestMessage);*/
|
||||
|
||||
OrderOperation orderOperation = new OrderOperation(1001, "tudou");
|
||||
// 发送消息
|
||||
ch.channel().writeAndFlush(orderOperation);
|
||||
// long id = IdUtil.nextId();
|
||||
// ResponseMessagePromise promise = center.addPromise(id);
|
||||
// ch.channel().writeAndFlush(new RequestMessage(id, orderOperation));
|
||||
|
||||
// ResponseMessage responseMessage = promise.get();
|
||||
// System.out.println("阻塞等待回复：" + responseMessage);
|
||||
|
||||
ch.channel().closeFuture().get();
|
||||
|
||||
}catch (Exception e){
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
boss.shutdownGracefully();
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,78 @@
|
||||
package com.baiye.case5.client;
|
||||
|
||||
import com.baiye.case5.client.codec.*;
|
||||
import com.baiye.case5.client.handler.dispatcher.OperationResultFuture;
|
||||
import com.baiye.case5.client.handler.dispatcher.RequestPendingCenter;
|
||||
import com.baiye.case5.client.handler.dispatcher.ResponseDispatcherHandler;
|
||||
import com.baiye.case5.common.OperationResult;
|
||||
import com.baiye.case5.common.order.OrderOperation;
|
||||
import com.baiye.case5.util.IdUtil;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelPipeline;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
import io.netty.handler.logging.LoggingHandler;
|
||||
|
||||
/**
|
||||
* 对应 future -> streamId 的方式
|
||||
*/
|
||||
public class OrderClientV3 {
|
||||
|
||||
public static void main(String[] args) {
|
||||
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
LoggingHandler loggingHandler = new LoggingHandler(LogLevel.INFO);
|
||||
|
||||
NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
|
||||
|
||||
RequestPendingCenter requestPendingCenter = new RequestPendingCenter();
|
||||
|
||||
try {
|
||||
bootstrap.group(nioEventLoopGroup);
|
||||
bootstrap.channel(NioSocketChannel.class);
|
||||
bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(NioSocketChannel channel) throws Exception {
|
||||
ChannelPipeline pipeline = channel.pipeline();
|
||||
// 编解码
|
||||
pipeline.addLast(new OrderFrameDecoder());
|
||||
pipeline.addLast(new OrderFrameEncoder());
|
||||
pipeline.addLast(new ClientProtocolDecoder());
|
||||
pipeline.addLast(new ClientProtocolEncoder());
|
||||
// future 对应的 handler
|
||||
pipeline.addLast(new ResponseDispatcherHandler(requestPendingCenter));
|
||||
// 请求内容 Encode
|
||||
pipeline.addLast(new OperationToRequestMessageEncoder());
|
||||
// Netty 日志 - handler
|
||||
pipeline.addLast(loggingHandler);
|
||||
|
||||
}
|
||||
});
|
||||
// 建立连接 - sync
|
||||
ChannelFuture ch = bootstrap.connect("127.0.0.1", 8090).sync();
|
||||
// 构建消息体
|
||||
OrderOperation orderOperation = new OrderOperation(1001, "tudou");
|
||||
// 生成 streamId
|
||||
long streamId = IdUtil.nextId();
|
||||
// 构建请求对应的 future
|
||||
OperationResultFuture operationResultFuture = new OperationResultFuture();
|
||||
// 发送消息之前把 streamId + future 放入 center 中
|
||||
requestPendingCenter.add(streamId, operationResultFuture);
|
||||
// 发送消息
|
||||
// TODO: 2022/8/23 当发送消息失败的时候 future 也应该处理
|
||||
ch.channel().writeAndFlush(orderOperation);
|
||||
// 使用 future 来等待请求的结果
|
||||
OperationResult operationResult = operationResultFuture.get();
|
||||
// channel 关闭
|
||||
ch.channel().closeFuture().get();
|
||||
|
||||
}catch (Exception e){
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
nioEventLoopGroup.shutdownGracefully();
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,84 @@
|
||||
package com.baiye.case5.client;
|
||||
|
||||
import com.baiye.case5.client.codec.*;
|
||||
import com.baiye.case5.client.handler.dispatcher.OperationResultFuture;
|
||||
import com.baiye.case5.client.handler.dispatcher.RequestPendingCenter;
|
||||
import com.baiye.case5.client.handler.dispatcher.ResponseDispatcherHandler;
|
||||
import com.baiye.case5.common.OperationResult;
|
||||
import com.baiye.case5.common.order.OrderOperation;
|
||||
import com.baiye.case5.util.IdUtil;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelPipeline;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioChannelOption;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
import io.netty.handler.logging.LoggingHandler;
|
||||
|
||||
/**
|
||||
* 参数调优版本
|
||||
*
|
||||
* @author q
|
||||
*/
|
||||
public class OrderClientV4 {
|
||||
|
||||
public static void main(String[] args) {
|
||||
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
LoggingHandler loggingHandler = new LoggingHandler(LogLevel.INFO);
|
||||
|
||||
NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
|
||||
|
||||
RequestPendingCenter requestPendingCenter = new RequestPendingCenter();
|
||||
|
||||
try {
|
||||
bootstrap.group(nioEventLoopGroup);
|
||||
bootstrap.channel(NioSocketChannel.class);
|
||||
// 参数调优1 超时 10s
|
||||
bootstrap.option(NioChannelOption.CONNECT_TIMEOUT_MILLIS, 10 * 1000);
|
||||
bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(NioSocketChannel channel) throws Exception {
|
||||
ChannelPipeline pipeline = channel.pipeline();
|
||||
// 编解码
|
||||
pipeline.addLast(new OrderFrameDecoder());
|
||||
pipeline.addLast(new OrderFrameEncoder());
|
||||
pipeline.addLast(new ClientProtocolDecoder());
|
||||
pipeline.addLast(new ClientProtocolEncoder());
|
||||
// future 对应的 handler
|
||||
pipeline.addLast(new ResponseDispatcherHandler(requestPendingCenter));
|
||||
// 请求内容 Encode
|
||||
pipeline.addLast(new OperationToRequestMessageEncoder());
|
||||
// Netty 日志 - handler
|
||||
pipeline.addLast(loggingHandler);
|
||||
|
||||
}
|
||||
});
|
||||
// 建立连接 - sync
|
||||
ChannelFuture ch = bootstrap.connect("127.0.0.1", 8090).sync();
|
||||
// 构建消息体
|
||||
OrderOperation orderOperation = new OrderOperation(1001, "tudou");
|
||||
// 生成 streamId
|
||||
long streamId = IdUtil.nextId();
|
||||
// 构建请求对应的 future
|
||||
OperationResultFuture operationResultFuture = new OperationResultFuture();
|
||||
// 发送消息之前把 streamId + future 放入 center 中
|
||||
requestPendingCenter.add(streamId, operationResultFuture);
|
||||
// 发送消息
|
||||
// TODO: 2022/8/23 当发送消息失败的时候 future 也应该处理
|
||||
ch.channel().writeAndFlush(orderOperation);
|
||||
// 使用 future 来等待请求的结果
|
||||
OperationResult operationResult = operationResultFuture.get();
|
||||
// channel 关闭
|
||||
ch.channel().closeFuture().get();
|
||||
|
||||
}catch (Exception e){
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
nioEventLoopGroup.shutdownGracefully();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,39 @@
|
||||
package com.baiye.case5.client.codec;
|
||||
|
||||
import com.baiye.case5.common.ResponseMessage;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.MessageToMessageDecoder;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 4. ByteBuf -> msg
|
||||
*/
|
||||
public class ClientProtocolDecoder extends MessageToMessageDecoder<ByteBuf> {
|
||||
|
||||
@Override
|
||||
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
|
||||
// byte[] res = new byte[in.readableBytes()];
|
||||
// in.copy().readBytes(res);
|
||||
//// System.out.println(" in: " + new String(Base64.getEncoder().encode(res)));
|
||||
// System.out.println(" in: " + bytesToHex(res));
|
||||
|
||||
ResponseMessage responseMessage = new ResponseMessage();
|
||||
responseMessage.decode(in);
|
||||
out.add(responseMessage);
|
||||
}
|
||||
|
||||
|
||||
public static String bytesToHex(byte[] bytes) {
|
||||
StringBuffer sb = new StringBuffer();
|
||||
for(int i = 0; i < bytes.length; i++) {
|
||||
String hex = Integer.toHexString(bytes[i] & 0xFF);
|
||||
if(hex.length() < 2){
|
||||
sb.append(0);
|
||||
}
|
||||
sb.append(hex);
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,23 @@
|
||||
package com.baiye.case5.client.codec;
|
||||
|
||||
|
||||
import com.baiye.case5.common.RequestMessage;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.MessageToMessageEncoder;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 1. 客户端 RequestMessage -> 二进制的字节流
|
||||
*/
|
||||
public class ClientProtocolEncoder extends MessageToMessageEncoder<RequestMessage> {
|
||||
|
||||
@Override
|
||||
protected void encode(ChannelHandlerContext ctx, RequestMessage msg, List<Object> out) throws Exception {
|
||||
ByteBuf buffer = ctx.alloc().buffer();
|
||||
msg.encode(buffer);
|
||||
out.add(buffer);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,19 @@
|
||||
package com.baiye.case5.client.codec;
|
||||
|
||||
|
||||
import com.baiye.case5.common.Operation;
|
||||
import com.baiye.case5.common.RequestMessage;
|
||||
import com.baiye.case5.util.IdUtil;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.MessageToMessageEncoder;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public class OperationToRequestMessageEncoder extends MessageToMessageEncoder<Operation> {
|
||||
@Override
|
||||
protected void encode(ChannelHandlerContext ctx, Operation operation, List<Object> out) throws Exception {
|
||||
RequestMessage requestMessage = new RequestMessage(IdUtil.nextId(), operation);
|
||||
|
||||
out.add(requestMessage);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
package com.baiye.case5.client.codec;
|
||||
|
||||
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
|
||||
|
||||
/**
|
||||
* 3. 请求进来 -> 解决 TCP协议的粘包和半包问题 [其他协议除外]
|
||||
*/
|
||||
public class OrderFrameDecoder extends LengthFieldBasedFrameDecoder {
|
||||
|
||||
/**
|
||||
*
|
||||
* 最大长度
|
||||
* 长度的位移
|
||||
* 长度
|
||||
* 需要进行调整
|
||||
* 跳过头长度
|
||||
*/
|
||||
public OrderFrameDecoder() {
|
||||
super(
|
||||
Integer.MAX_VALUE,
|
||||
0,
|
||||
2,
|
||||
0,
|
||||
2
|
||||
);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,13 @@
|
||||
package com.baiye.case5.client.codec;
|
||||
|
||||
|
||||
import io.netty.handler.codec.LengthFieldPrepender;
|
||||
|
||||
/**
|
||||
* 2. 处理半包问题
|
||||
*/
|
||||
public class OrderFrameEncoder extends LengthFieldPrepender {
|
||||
public OrderFrameEncoder() {
|
||||
super(2);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,20 @@
|
||||
package com.baiye.case5.client.handler;
|
||||
|
||||
import com.baiye.case5.common.ResponseMessage;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
|
||||
/**
|
||||
* @author lianghuarong
|
||||
* @date 2021/6/16 15:08
|
||||
*/
|
||||
@ChannelHandler.Sharable
|
||||
public class ClientHandler extends SimpleChannelInboundHandler<ResponseMessage> {
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, ResponseMessage msg) throws Exception {
|
||||
System.out.println("收到响应：" + msg);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,8 @@
|
||||
package com.baiye.case5.client.handler.dispatcher;
|
||||
|
||||
|
||||
import com.baiye.case5.common.OperationResult;
|
||||
import io.netty.util.concurrent.DefaultPromise;
|
||||
|
||||
public class OperationResultFuture extends DefaultPromise<OperationResult> {
|
||||
}
|
||||
@ -0,0 +1,45 @@
|
||||
package com.baiye.case5.client.handler.dispatcher;
|
||||
|
||||
|
||||
import com.baiye.case5.common.OperationResult;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* 存储 streamId -> Future 的关系
|
||||
*
|
||||
* @author q
|
||||
*/
|
||||
public class RequestPendingCenter {
|
||||
|
||||
private Map<Long, OperationResultFuture> map = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* 发送请求 构建 streamId 和 future 的映射
|
||||
*
|
||||
* @param streamId 请求id
|
||||
* @param future 对应该请求的 future
|
||||
*/
|
||||
public void add(Long streamId, OperationResultFuture future){
|
||||
this.map.put(streamId, future);
|
||||
}
|
||||
|
||||
/**
|
||||
* 响应 构建 streamId 对应 future 的结果
|
||||
*
|
||||
* @param streamId 请求id
|
||||
* @param operationResult 请求结果
|
||||
*/
|
||||
public void set(Long streamId, OperationResult operationResult){
|
||||
OperationResultFuture operationResultFuture = this.map.get(streamId);
|
||||
if (operationResultFuture != null) {
|
||||
// 找到, 设置为成功
|
||||
operationResultFuture.setSuccess(operationResult);
|
||||
// 找到结果, 把整个结果进行移除, map容量控制
|
||||
this.map.remove(streamId);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,35 @@
|
||||
package com.baiye.case5.client.handler.dispatcher;
|
||||
|
||||
|
||||
import com.baiye.case5.common.ResponseMessage;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
|
||||
/**
|
||||
* 把请求对应的结果对应到 Center 上去
|
||||
*
|
||||
* @author q
|
||||
*/
|
||||
public class ResponseDispatcherHandler extends SimpleChannelInboundHandler<ResponseMessage> {
|
||||
|
||||
/**
|
||||
* requestPendingCenter 作为成员变量
|
||||
*/
|
||||
private RequestPendingCenter requestPendingCenter;
|
||||
|
||||
/**
|
||||
* 堆外开放构造器注入
|
||||
*/
|
||||
public ResponseDispatcherHandler(RequestPendingCenter requestPendingCenter) {
|
||||
this.requestPendingCenter = requestPendingCenter;
|
||||
}
|
||||
|
||||
/**
|
||||
* 给 future 设置结果
|
||||
*/
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, ResponseMessage responseMessage) throws Exception {
|
||||
// streamId(header) + result(body)
|
||||
requestPendingCenter.set(responseMessage.getMessageHeader().getStreamId(), responseMessage.getMessageBody());
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,34 @@
|
||||
package com.baiye.case5.client.handler.dispatcher;
|
||||
|
||||
import com.baiye.case5.common.ResponseMessage;
|
||||
import io.netty.util.concurrent.DefaultPromise;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* @author lianghuarong
|
||||
* @date 2021/6/17 14:09
|
||||
*/
|
||||
public class ResponseMessagePendingCenter {
|
||||
|
||||
private Map<Long, ResponseMessagePromise> map = new ConcurrentHashMap<>();
|
||||
|
||||
public void setSuccess(long streamId, ResponseMessage responseMessage){
|
||||
if (map.containsKey(streamId)){
|
||||
DefaultPromise promise = map.get(streamId);
|
||||
promise.setSuccess(responseMessage);
|
||||
map.remove(streamId);
|
||||
}
|
||||
}
|
||||
|
||||
public ResponseMessagePromise addPromise(long streamId){
|
||||
if (map.containsKey(streamId)){
|
||||
return map.get(streamId);
|
||||
}
|
||||
|
||||
ResponseMessagePromise p = new ResponseMessagePromise();
|
||||
map.put(streamId, p);
|
||||
return p;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,11 @@
|
||||
package com.baiye.case5.client.handler.dispatcher;
|
||||
|
||||
import com.baiye.case5.common.ResponseMessage;
|
||||
import io.netty.util.concurrent.DefaultPromise;
|
||||
|
||||
/**
|
||||
* @author lianghuarong
|
||||
* @date 2021/6/17 14:04
|
||||
*/
|
||||
public class ResponseMessagePromise extends DefaultPromise<ResponseMessage> {
|
||||
}
|
||||
@ -0,0 +1,25 @@
|
||||
package com.baiye.case5.client.handler.dispatcher;
|
||||
|
||||
|
||||
import com.baiye.case5.common.ResponseMessage;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
|
||||
/**
|
||||
* @author lianghuarong
|
||||
* @date 2021/6/17 14:15
|
||||
*/
|
||||
public class ResponseMessagePromiseHandler extends SimpleChannelInboundHandler<ResponseMessage> {
|
||||
|
||||
private ResponseMessagePendingCenter pendingCenter;
|
||||
|
||||
public ResponseMessagePromiseHandler(ResponseMessagePendingCenter pendingCenter){
|
||||
this.pendingCenter = pendingCenter;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, ResponseMessage msg) throws Exception {
|
||||
pendingCenter.setSuccess(msg.getMessageHeader().getStreamId(), msg);
|
||||
ctx.fireChannelRead(msg);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,63 @@
|
||||
package com.baiye.case5.common;
|
||||
|
||||
import com.baiye.case5.util.JsonUtil;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import lombok.Data;
|
||||
|
||||
import java.nio.charset.Charset;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
import static cn.hutool.core.util.CharsetUtil.UTF_8;
|
||||
|
||||
/**
|
||||
* 自定义的一个简易的 编解码器
|
||||
* @param <T>
|
||||
*
|
||||
*/
|
||||
@Data
|
||||
public abstract class Message<T extends MessageBody> {
|
||||
private MessageHeader messageHeader;
|
||||
private T messageBody;
|
||||
|
||||
public T getMessageBody() {
|
||||
return messageBody;
|
||||
}
|
||||
|
||||
/**
|
||||
* 编码器
|
||||
*
|
||||
* @param byteBuf
|
||||
*/
|
||||
public void encode(ByteBuf byteBuf) {
|
||||
byteBuf.writeInt(messageHeader.getVersion());
|
||||
byteBuf.writeLong(messageHeader.getStreamId());
|
||||
byteBuf.writeInt(messageHeader.getOpCode());
|
||||
byteBuf.writeBytes(JsonUtil.toJson(messageBody).getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
|
||||
public abstract Class<T> getMessageBodyDecodeClass(int opCode);
|
||||
|
||||
|
||||
/**
|
||||
* 解码器
|
||||
*
|
||||
* @param msg
|
||||
*/
|
||||
public void decode(ByteBuf msg) {
|
||||
int version = msg.readInt();
|
||||
long streamId = msg.readLong();
|
||||
int opCode = msg.readInt();
|
||||
|
||||
MessageHeader messageHeader = new MessageHeader();
|
||||
messageHeader.setVersion(version);
|
||||
messageHeader.setStreamId(streamId);
|
||||
messageHeader.setOpCode(opCode);
|
||||
this.messageHeader = messageHeader;
|
||||
|
||||
Class<T> bodyClazz = getMessageBodyDecodeClass(opCode);
|
||||
T body = JsonUtil.fromJson(msg.toString(StandardCharsets.UTF_8), bodyClazz);
|
||||
this.messageBody = body;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,7 @@
|
||||
package com.baiye.case5.common;
|
||||
|
||||
/**
|
||||
* MessageBody 基类
|
||||
*/
|
||||
public abstract class MessageBody {
|
||||
}
|
||||
@ -0,0 +1,16 @@
|
||||
package com.baiye.case5.common;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class MessageHeader {
|
||||
|
||||
/**
|
||||
* 简单示范这里直接设置为 1
|
||||
*/
|
||||
private int version = 1;
|
||||
|
||||
private long streamId;
|
||||
|
||||
private int opCode;
|
||||
}
|
||||
@ -0,0 +1,6 @@
|
||||
package com.baiye.case5.common;
|
||||
|
||||
public abstract class Operation extends MessageBody{
|
||||
|
||||
public abstract OperationResult execute();
|
||||
}
|
||||
@ -0,0 +1,7 @@
|
||||
package com.baiye.case5.common;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public abstract class OperationResult extends MessageBody{
|
||||
}
|
||||
@ -0,0 +1,74 @@
|
||||
package com.baiye.case5.common;
|
||||
|
||||
import com.baiye.case5.common.auth.AuthOperation;
|
||||
import com.baiye.case5.common.auth.AuthOperationResult;
|
||||
import com.baiye.case5.common.keepalive.KeepaliveOperation;
|
||||
import com.baiye.case5.common.keepalive.KeepaliveOperationResult;
|
||||
import com.baiye.case5.common.order.OrderOperation;
|
||||
import com.baiye.case5.common.order.OrderOperationResult;
|
||||
|
||||
import java.util.function.Predicate;
|
||||
|
||||
public enum OperationType {
|
||||
/**
|
||||
* 权限
|
||||
*/
|
||||
AUTH(1, AuthOperation.class, AuthOperationResult.class),
|
||||
/**
|
||||
* 心跳
|
||||
*/
|
||||
KEEPALIVE(2, KeepaliveOperation.class, KeepaliveOperationResult.class),
|
||||
/**
|
||||
* 订单
|
||||
*/
|
||||
ORDER(3, OrderOperation.class, OrderOperationResult.class)
|
||||
;
|
||||
|
||||
|
||||
private int opCode;
|
||||
private Class<? extends Operation> operationClazz;
|
||||
private Class<? extends OperationResult> operationResultClazz;
|
||||
|
||||
OperationType(int opCode, Class<? extends Operation> operationClazz, Class<? extends OperationResult> operationResultClazz) {
|
||||
this.opCode = opCode;
|
||||
this.operationClazz = operationClazz;
|
||||
this.operationResultClazz = operationResultClazz;
|
||||
}
|
||||
|
||||
public Class<? extends Operation> getOperationClazz(){
|
||||
return operationClazz;
|
||||
}
|
||||
|
||||
public int getOpCode() {
|
||||
return opCode;
|
||||
}
|
||||
|
||||
public Class<? extends OperationResult> getOperationResultClazz() {
|
||||
return operationResultClazz;
|
||||
}
|
||||
|
||||
public static OperationType fromOpCode(int type){
|
||||
return getOperationType(requestType -> requestType.opCode == type);
|
||||
}
|
||||
|
||||
public static OperationType fromOperation(Operation operation){
|
||||
return getOperationType(requestType -> requestType.operationClazz == operation.getClass());
|
||||
}
|
||||
|
||||
public static OperationType fromOperationResult(OperationResult operationResult){
|
||||
return getOperationType(requestType -> requestType.operationResultClazz == operationResult.getClass());
|
||||
}
|
||||
|
||||
private static OperationType getOperationType(Predicate<OperationType> predicate){
|
||||
OperationType[] values = values();
|
||||
for (OperationType operationType : values) {
|
||||
if(predicate.test(operationType)){
|
||||
return operationType;
|
||||
}
|
||||
}
|
||||
|
||||
throw new AssertionError("no found type");
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,20 @@
|
||||
package com.baiye.case5.common;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
public class RequestMessage extends Message<Operation>{
|
||||
@Override
|
||||
public Class getMessageBodyDecodeClass(int opcode) {
|
||||
return OperationType.fromOpCode(opcode).getOperationClazz();
|
||||
}
|
||||
|
||||
public RequestMessage(){}
|
||||
|
||||
public RequestMessage(Long streamId, Operation operation){
|
||||
MessageHeader messageHeader = new MessageHeader();
|
||||
messageHeader.setStreamId(streamId);
|
||||
messageHeader.setOpCode(OperationType.fromOperation(operation).getOpCode());
|
||||
this.setMessageHeader(messageHeader);
|
||||
this.setMessageBody(operation);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,18 @@
|
||||
package com.baiye.case5.common;
|
||||
|
||||
public class ResponseMessage extends Message <OperationResult>{
|
||||
@Override
|
||||
public Class getMessageBodyDecodeClass(int opcode) {
|
||||
return OperationType.fromOpCode(opcode).getOperationResultClazz();
|
||||
}
|
||||
|
||||
public ResponseMessage(){}
|
||||
|
||||
public ResponseMessage(Long streamId, OperationResult operationResult){
|
||||
MessageHeader messageHeader = new MessageHeader();
|
||||
messageHeader.setStreamId(streamId);
|
||||
messageHeader.setOpCode(OperationType.fromOperationResult(operationResult).getOpCode());
|
||||
this.setMessageHeader(messageHeader);
|
||||
this.setMessageBody(operationResult);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,24 @@
|
||||
package com.baiye.case5.common.auth;
|
||||
|
||||
|
||||
import com.baiye.case5.common.Operation;
|
||||
import lombok.Data;
|
||||
import lombok.extern.java.Log;
|
||||
|
||||
@Data
|
||||
@Log
|
||||
public class AuthOperation extends Operation {
|
||||
|
||||
private final String userName;
|
||||
private final String password;
|
||||
|
||||
@Override
|
||||
public AuthOperationResult execute() {
|
||||
if("admin".equalsIgnoreCase(this.userName)){
|
||||
AuthOperationResult orderResponse = new AuthOperationResult(true);
|
||||
return orderResponse;
|
||||
}
|
||||
|
||||
return new AuthOperationResult(false);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,11 @@
|
||||
package com.baiye.case5.common.auth;
|
||||
|
||||
import com.baiye.case5.common.OperationResult;
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class AuthOperationResult extends OperationResult {
|
||||
|
||||
private final boolean passAuth;
|
||||
|
||||
}
|
||||
@ -0,0 +1,23 @@
|
||||
package com.baiye.case5.common.keepalive;
|
||||
|
||||
|
||||
import com.baiye.case5.common.Operation;
|
||||
import lombok.Data;
|
||||
import lombok.extern.java.Log;
|
||||
|
||||
@Data
|
||||
@Log
|
||||
public class KeepaliveOperation extends Operation {
|
||||
|
||||
private long time ;
|
||||
|
||||
public KeepaliveOperation() {
|
||||
this.time = System.nanoTime();
|
||||
}
|
||||
|
||||
@Override
|
||||
public KeepaliveOperationResult execute() {
|
||||
KeepaliveOperationResult orderResponse = new KeepaliveOperationResult(time);
|
||||
return orderResponse;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,11 @@
|
||||
package com.baiye.case5.common.keepalive;
|
||||
|
||||
import com.baiye.case5.common.OperationResult;
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class KeepaliveOperationResult extends OperationResult {
|
||||
|
||||
private final long time;
|
||||
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
package com.baiye.case5.common.order;
|
||||
|
||||
|
||||
import com.baiye.case5.common.Operation;
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class OrderOperation extends Operation {
|
||||
|
||||
private int tableId;
|
||||
private String dish;
|
||||
|
||||
public OrderOperation(int tableId, String dish) {
|
||||
this.tableId = tableId;
|
||||
this.dish = dish;
|
||||
}
|
||||
|
||||
@Override
|
||||
public OrderOperationResult execute() {
|
||||
System.out.println("order's executing startup with orderRequest: " + toString());
|
||||
//execute order logic
|
||||
System.out.println("order's executing complete");
|
||||
OrderOperationResult orderResponse = new OrderOperationResult(tableId, dish, true);
|
||||
return orderResponse;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,13 @@
|
||||
package com.baiye.case5.common.order;
|
||||
|
||||
import com.baiye.case5.common.OperationResult;
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class OrderOperationResult extends OperationResult {
|
||||
|
||||
private final int tableId;
|
||||
private final String dish;
|
||||
private final boolean complete;
|
||||
|
||||
}
|
||||
@ -0,0 +1,63 @@
|
||||
package com.baiye.case5.server;
|
||||
|
||||
import com.baiye.case5.server.codec.OrderFrameDecoder;
|
||||
import com.baiye.case5.server.codec.OrderFrameEncoder;
|
||||
import com.baiye.case5.server.codec.OrderProtocolDecoder;
|
||||
import com.baiye.case5.server.codec.OrderProtocolEncoder;
|
||||
import com.baiye.case5.server.handler.OrderServerProcessHandler;
|
||||
import io.netty.bootstrap.ServerBootstrap;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelPipeline;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
import io.netty.handler.logging.LoggingHandler;
|
||||
|
||||
import java.util.concurrent.ExecutionException;
|
||||
|
||||
/**
|
||||
* 服务端
|
||||
*/
|
||||
public class Server {
|
||||
|
||||
public static void main(String[] args) throws InterruptedException, ExecutionException {
|
||||
|
||||
ServerBootstrap serverBootstrap = new ServerBootstrap();
|
||||
// 设置 I/O 模式
|
||||
serverBootstrap.channel(NioServerSocketChannel.class);
|
||||
// 设置 日志级别
|
||||
serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
|
||||
// 设置 Reactor 方式
|
||||
NioEventLoopGroup group = new NioEventLoopGroup();
|
||||
try{
|
||||
serverBootstrap.group(group);
|
||||
serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(NioSocketChannel ch) throws Exception {
|
||||
ChannelPipeline pipeline = ch.pipeline();
|
||||
|
||||
pipeline.addLast(new OrderFrameDecoder());
|
||||
pipeline.addLast(new OrderFrameEncoder());
|
||||
|
||||
pipeline.addLast(new OrderProtocolEncoder());
|
||||
pipeline.addLast(new OrderProtocolDecoder());
|
||||
// 设置 日志级别
|
||||
pipeline.addLast(new LoggingHandler(LogLevel.INFO));
|
||||
|
||||
pipeline.addLast(new OrderServerProcessHandler());
|
||||
}
|
||||
});
|
||||
// 绑定端口启动
|
||||
ChannelFuture channelFuture = serverBootstrap.bind(8090).sync();
|
||||
|
||||
channelFuture.channel().closeFuture().sync();
|
||||
} finally {
|
||||
group.shutdownGracefully();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,71 @@
|
||||
package com.baiye.case5.server;
|
||||
|
||||
import com.baiye.case5.server.codec.OrderFrameDecoder;
|
||||
import com.baiye.case5.server.codec.OrderFrameEncoder;
|
||||
import com.baiye.case5.server.codec.OrderProtocolDecoder;
|
||||
import com.baiye.case5.server.codec.OrderProtocolEncoder;
|
||||
import com.baiye.case5.server.handler.OrderServerProcessHandler;
|
||||
import io.netty.bootstrap.ServerBootstrap;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelPipeline;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioChannelOption;
|
||||
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
import io.netty.handler.logging.LoggingHandler;
|
||||
|
||||
import java.util.concurrent.ExecutionException;
|
||||
|
||||
/**
|
||||
* 服务端 - 参数调优版本
|
||||
*
|
||||
* @author q
|
||||
*/
|
||||
public class ServerV1 {
|
||||
|
||||
public static void main(String[] args) throws InterruptedException, ExecutionException {
|
||||
|
||||
ServerBootstrap serverBootstrap = new ServerBootstrap();
|
||||
// 设置 I/O 模式
|
||||
serverBootstrap.channel(NioServerSocketChannel.class);
|
||||
// 设置 日志级别
|
||||
serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
|
||||
// 设置 Reactor 方式
|
||||
NioEventLoopGroup group = new NioEventLoopGroup();
|
||||
try{
|
||||
serverBootstrap.group(group);
|
||||
// 参数调优1 - 必调
|
||||
serverBootstrap.childOption(NioChannelOption.TCP_NODELAY, true);
|
||||
// 参数调优2 - 必调
|
||||
serverBootstrap.option(NioChannelOption.SO_BACKLOG, 1024);
|
||||
|
||||
serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(NioSocketChannel ch) throws Exception {
|
||||
ChannelPipeline pipeline = ch.pipeline();
|
||||
|
||||
pipeline.addLast(new OrderFrameDecoder());
|
||||
pipeline.addLast(new OrderFrameEncoder());
|
||||
|
||||
pipeline.addLast(new OrderProtocolEncoder());
|
||||
pipeline.addLast(new OrderProtocolDecoder());
|
||||
// 设置 日志级别
|
||||
pipeline.addLast(new LoggingHandler(LogLevel.INFO));
|
||||
|
||||
pipeline.addLast(new OrderServerProcessHandler());
|
||||
}
|
||||
});
|
||||
// 绑定端口启动
|
||||
ChannelFuture channelFuture = serverBootstrap.bind(8090).sync();
|
||||
|
||||
channelFuture.channel().closeFuture().sync();
|
||||
} finally {
|
||||
group.shutdownGracefully();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
package com.baiye.case5.server.codec;
|
||||
|
||||
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
|
||||
|
||||
/**
|
||||
* 1. 请求进来 -> 解决 TCP协议的粘包和半包问题 [其他协议除外]
|
||||
*/
|
||||
public class OrderFrameDecoder extends LengthFieldBasedFrameDecoder {
|
||||
|
||||
/**
|
||||
*
|
||||
* 最大长度
|
||||
* 长度的位移
|
||||
* 长度
|
||||
* 需要进行调整
|
||||
* 跳过头长度
|
||||
*/
|
||||
public OrderFrameDecoder() {
|
||||
super(
|
||||
Integer.MAX_VALUE,
|
||||
0,
|
||||
2,
|
||||
0,
|
||||
2
|
||||
);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,13 @@
|
||||
package com.baiye.case5.server.codec;
|
||||
|
||||
|
||||
import io.netty.handler.codec.LengthFieldPrepender;
|
||||
|
||||
/**
|
||||
* 5. 解决客户端的粘包和半包问题
|
||||
*/
|
||||
public class OrderFrameEncoder extends LengthFieldPrepender {
|
||||
public OrderFrameEncoder() {
|
||||
super(2);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,22 @@
|
||||
package com.baiye.case5.server.codec;
|
||||
|
||||
import com.baiye.case5.common.RequestMessage;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.MessageToMessageDecoder;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 2. 第二阶段的Decoder 二次解码器 ByteBuf -> RequestMessage
|
||||
*
|
||||
*/
|
||||
public class OrderProtocolDecoder extends MessageToMessageDecoder<ByteBuf> {
|
||||
@Override
|
||||
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> out) throws Exception {
|
||||
RequestMessage requestMessage = new RequestMessage();
|
||||
requestMessage.decode(byteBuf);
|
||||
|
||||
out.add(requestMessage);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,22 @@
|
||||
package com.baiye.case5.server.codec;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import com.baiye.case5.common.ResponseMessage;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.codec.MessageToMessageEncoder;
|
||||
|
||||
/**
|
||||
* 4. 业务 Handler writeAndFlush 处理后的信息 ResponseMessage -> ByteBuf
|
||||
*/
|
||||
public class OrderProtocolEncoder extends MessageToMessageEncoder<ResponseMessage> {
|
||||
|
||||
@Override
|
||||
protected void encode(ChannelHandlerContext ctx, ResponseMessage responseMessage, List<Object> out) throws Exception {
|
||||
ByteBuf buffer = ctx.alloc().buffer();
|
||||
responseMessage.encode(buffer);
|
||||
|
||||
out.add(buffer);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,28 @@
|
||||
package com.baiye.case5.server.handler;
|
||||
|
||||
import com.baiye.case5.common.Operation;
|
||||
import com.baiye.case5.common.OperationResult;
|
||||
import com.baiye.case5.common.RequestMessage;
|
||||
import com.baiye.case5.common.ResponseMessage;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
|
||||
/**
|
||||
* 3. 业务 Handler 进行处理
|
||||
*/
|
||||
public class OrderServerProcessHandler extends SimpleChannelInboundHandler<RequestMessage> {
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext channelHandlerContext, RequestMessage requestMessage) throws Exception {
|
||||
Operation operation = requestMessage.getMessageBody();
|
||||
OperationResult operationResult = operation.execute();
|
||||
|
||||
// 发送结果返回回去
|
||||
ResponseMessage responseMessage = new ResponseMessage();
|
||||
// 头信息原样返回
|
||||
responseMessage.setMessageHeader(requestMessage.getMessageHeader());
|
||||
responseMessage.setMessageBody(operationResult);
|
||||
|
||||
// 信息写出去
|
||||
channelHandlerContext.writeAndFlush(responseMessage);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,19 @@
|
||||
package com.baiye.case5.util;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
public final class IdUtil {
|
||||
|
||||
public static final AtomicLong IDX = new AtomicLong();
|
||||
|
||||
private IdUtil() {
|
||||
}
|
||||
|
||||
/**
|
||||
* 生成 streamId
|
||||
* @return streamId
|
||||
*/
|
||||
public static long nextId(){
|
||||
return IDX.incrementAndGet();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,25 @@
|
||||
package com.baiye.case5.util;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
|
||||
public final class JsonUtil {
|
||||
|
||||
private static final Gson GSON = new Gson();
|
||||
|
||||
private JsonUtil() {
|
||||
}
|
||||
|
||||
/**
|
||||
* 序列化
|
||||
*/
|
||||
public static <T> T fromJson(String jsonStr, Class<T> clazz){
|
||||
return GSON.fromJson(jsonStr, clazz);
|
||||
}
|
||||
|
||||
/**
|
||||
* 反序列化
|
||||
*/
|
||||
public static String toJson(Object obj) {
|
||||
return GSON.toJson(obj);
|
||||
}
|
||||
}
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 484 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 261 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 78 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 662 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 341 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 103 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 413 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 143 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 136 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 267 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 222 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 157 KiB |
Loading…
Reference in New Issue