alsace/wu-lazy-cloud-network
1
0
Fork 0
mirror of https://gitee.com/wujiawei1207537021/wu-lazy-cloud-network.git synced 2025-06-06 13:27:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
wujiawei 2023-02-12 10:34:46 +08:00
parent fae329ea3e
commit 6cb17fbd58
20 changed files with 837 additions and 831 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
netty-proxy-client/pom.xml
View File

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>netty-proxy</artifactId>

3
netty-proxy-client/src/main/java/com/luck/client/ClientStart.java
View File

@ -1,8 +1,5 @@
package com.luck.client;
import org.springframework.boot.SpringApplication;
public class ClientStart {
public static void main(String[] args) throws Exception {

150
netty-proxy-client/src/main/java/com/luck/client/Constant.java
View File

@ -1,75 +1,87 @@
package com.luck.client;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
import io.netty.util.internal.StringUtil;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class Constant {
/** 代理服务channel */
public static Channel proxyChannel = null;
/** 绑定访客id */
public static final AttributeKey<String> VID = AttributeKey.newInstance("vid");
/** 访客，代理服务channel */
public static Map<String, Channel> vpc = new ConcurrentHashMap<>();
/** 访客，真实服务channel */
public static Map<String, Channel> vrc = new ConcurrentHashMap<>();
/** 真实服务端口 */
public static int realPort = 8080;
/** 服务端口 */
public static int serverPort = 16001;
/** 服务IP */
public static String serverIp = "127.0.0.1";
/**
* 清除连接
*
* @param vid 访客ID
*/
public static void clearvpcvrc(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel clientChannel = vpc.get(vid);
if (null != clientChannel) {
clientChannel.attr(VID).set(null);
vpc.remove(vid);
}
Channel visitorChannel = vrc.get(vid);
if (null != visitorChannel) {
visitorChannel.attr(VID).set(null);
vrc.remove(vid);
}
}
/**
* 清除关闭连接
*
* @param vid 访客ID
*/
public static void clearvpcvrcAndClose(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel clientChannel = vpc.get(vid);
if (null != clientChannel) {
clientChannel.attr(VID).set(null);
vpc.remove(vid);
clientChannel.close();
}
Channel visitorChannel = vrc.get(vid);
if (null != visitorChannel) {
visitorChannel.attr(VID).set(null);
vrc.remove(vid);
visitorChannel.close();
}
}
/**
* 绑定访客id
*/
public static final AttributeKey<String> VID = AttributeKey.newInstance("vid");
/**
* 代理服务channel
*/
public static Channel proxyChannel = null;
/**
* 访客，代理服务channel
*/
public static Map<String, Channel> vpc = new ConcurrentHashMap<>();
/**
* 访客，真实服务channel
*/
public static Map<String, Channel> vrc = new ConcurrentHashMap<>();
/**
* 真实服务端口
*/
public static int realPort = 8080;
/**
* 服务端口
*/
public static int serverPort = 16001;
/**
* 服务IP
*/
public static String serverIp = "127.0.0.1";
/**
* 清除连接
*
* @param vid 访客ID
*/
public static void clearvpcvrc(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel clientChannel = vpc.get(vid);
if (null != clientChannel) {
clientChannel.attr(VID).set(null);
vpc.remove(vid);
}
Channel visitorChannel = vrc.get(vid);
if (null != visitorChannel) {
visitorChannel.attr(VID).set(null);
vrc.remove(vid);
}
}
/**
* 清除关闭连接
*
* @param vid 访客ID
*/
public static void clearvpcvrcAndClose(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel clientChannel = vpc.get(vid);
if (null != clientChannel) {
clientChannel.attr(VID).set(null);
vpc.remove(vid);
clientChannel.close();
}
Channel visitorChannel = vrc.get(vid);
if (null != visitorChannel) {
visitorChannel.attr(VID).set(null);
vrc.remove(vid);
visitorChannel.close();
}
}
}

175
netty-proxy-client/src/main/java/com/luck/client/ProxyHandler.java
View File

@ -1,7 +1,6 @@
package com.luck.client;
import com.luck.msg.MyMsg;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
@ -9,91 +8,91 @@ import io.netty.channel.ChannelOption;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.internal.StringUtil;
public class ProxyHandler extends SimpleChannelInboundHandler<MyMsg> {
@Override
public void channelRead0(ChannelHandlerContext ctx, MyMsg myMsg) {
// 客户端读取到代理过来的数据了
byte type = myMsg.getType();
String vid = new String(myMsg.getData());
switch (type) {
case MyMsg.TYPE_HEARTBEAT:
break;
case MyMsg.TYPE_CONNECT:
RealSocket.connectRealServer(vid);
break;
case MyMsg.TYPE_DISCONNECT:
Constant.clearvpcvrcAndClose(vid);
break;
case MyMsg.TYPE_TRANSFER:
// 把数据转到真实服务
ByteBuf buf = ctx.alloc().buffer(myMsg.getData().length);
buf.writeBytes(myMsg.getData());
String visitorId = ctx.channel().attr(Constant.VID).get();
Channel rchannel = Constant.vrc.get(visitorId);
if (null != rchannel) {
rchannel.writeAndFlush(buf);
}
break;
default:
// 操作有误
}
// 客户端发数据到真实服务了
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelWritabilityChanged(ctx);
return;
}
Channel realChannel = Constant.vrc.get(vid);
if (realChannel != null) {
realChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelInactive(ctx);
return;
}
Channel realChannel = Constant.vrc.get(vid);
if (realChannel != null && realChannel.isActive()) {
realChannel.close();
}
super.channelInactive(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
cause.printStackTrace();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
switch (event.state()) {
case READER_IDLE:
ctx.channel().close();
break;
case WRITER_IDLE:
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_HEARTBEAT);
ctx.channel().writeAndFlush(myMsg);
break;
case ALL_IDLE:
break;
}
}
}
@Override
public void channelRead0(ChannelHandlerContext ctx, MyMsg myMsg) {
// 客户端读取到代理过来的数据了
byte type = myMsg.getType();
String vid = new String(myMsg.getData());
switch (type) {
case MyMsg.TYPE_HEARTBEAT:
break;
case MyMsg.TYPE_CONNECT:
RealSocket.connectRealServer(vid);
break;
case MyMsg.TYPE_DISCONNECT:
Constant.clearvpcvrcAndClose(vid);
break;
case MyMsg.TYPE_TRANSFER:
// 把数据转到真实服务
ByteBuf buf = ctx.alloc().buffer(myMsg.getData().length);
buf.writeBytes(myMsg.getData());
String visitorId = ctx.channel().attr(Constant.VID).get();
Channel rchannel = Constant.vrc.get(visitorId);
if (null != rchannel) {
rchannel.writeAndFlush(buf);
}
break;
default:
// 操作有误
}
// 客户端发数据到真实服务了
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelWritabilityChanged(ctx);
return;
}
Channel realChannel = Constant.vrc.get(vid);
if (realChannel != null) {
realChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelInactive(ctx);
return;
}
Channel realChannel = Constant.vrc.get(vid);
if (realChannel != null && realChannel.isActive()) {
realChannel.close();
}
super.channelInactive(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
cause.printStackTrace();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
switch (event.state()) {
case READER_IDLE:
ctx.channel().close();
break;
case WRITER_IDLE:
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_HEARTBEAT);
ctx.channel().writeAndFlush(myMsg);
break;
case ALL_IDLE:
break;
}
}
}
}

184
netty-proxy-client/src/main/java/com/luck/client/ProxySocket.java
View File

@ -1,107 +1,101 @@
package com.luck.client;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import com.luck.msg.MyMsg;
import com.luck.msg.MyMsgDecoder;
import com.luck.msg.MyMsgEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.StringUtil;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ProxySocket {
private static EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
/** 重连代理服务 */
private static final ScheduledExecutorService reconnectExecutor = Executors.newSingleThreadScheduledExecutor();
public static Channel connectProxyServer() throws Exception {
reconnectExecutor.scheduleAtFixedRate(() -> {
try {
connectProxyServer(null);
} catch (Exception e) {
e.printStackTrace();
}
}, 3, 3, TimeUnit.SECONDS);
return connectProxyServer(null);
}
public static Channel connectProxyServer(String vid) throws Exception {
if (StringUtil.isNullOrEmpty(vid)) {
if (Constant.proxyChannel == null || !Constant.proxyChannel.isActive()) {
newConnect(null);
}
return null;
} else {
Channel channel = Constant.vpc.get(vid);
if (null == channel) {
newConnect(vid);
channel = Constant.vpc.get(vid);
}
return channel;
}
}
private static void newConnect(String vid) throws InterruptedException {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMsgDecoder(Integer.MAX_VALUE, 0, 4, -4, 0));
pipeline.addLast(new MyMsgEncoder());
pipeline.addLast(new IdleStateHandler(40, 8, 0));
pipeline.addLast(new ProxyHandler());
}
});
bootstrap.connect(Constant.serverIp, Constant.serverPort).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
// 客户端链接代理服务器成功
Channel channel = future.channel();
if (StringUtil.isNullOrEmpty(vid)) {
// 告诉服务端这条连接是client的连接
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_CONNECT);
myMsg.setData("client".getBytes());
channel.writeAndFlush(myMsg);
Constant.proxyChannel = channel;
} else {
// 告诉服务端这条连接是vid的连接
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_CONNECT);
myMsg.setData(vid.getBytes());
channel.writeAndFlush(myMsg);
// 客户端绑定通道关系
Constant.vpc.put(vid, channel);
channel.attr(Constant.VID).set(vid);
Channel realChannel = Constant.vrc.get(vid);
if (null != realChannel) {
realChannel.config().setOption(ChannelOption.AUTO_READ, true);
}
}
}
}
});
}
/**
* 重连代理服务
*/
private static final ScheduledExecutorService reconnectExecutor = Executors.newSingleThreadScheduledExecutor();
private static EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
public static Channel connectProxyServer() throws Exception {
reconnectExecutor.scheduleAtFixedRate(() -> {
try {
connectProxyServer(null);
} catch (Exception e) {
e.printStackTrace();
}
}, 3, 3, TimeUnit.SECONDS);
return connectProxyServer(null);
}
public static Channel connectProxyServer(String vid) throws Exception {
if (StringUtil.isNullOrEmpty(vid)) {
if (Constant.proxyChannel == null || !Constant.proxyChannel.isActive()) {
newConnect(null);
}
return null;
} else {
Channel channel = Constant.vpc.get(vid);
if (null == channel) {
newConnect(vid);
channel = Constant.vpc.get(vid);
}
return channel;
}
}
private static void newConnect(String vid) throws InterruptedException {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMsgDecoder(Integer.MAX_VALUE, 0, 4, -4, 0));
pipeline.addLast(new MyMsgEncoder());
pipeline.addLast(new IdleStateHandler(40, 8, 0));
pipeline.addLast(new ProxyHandler());
}
});
bootstrap.connect(Constant.serverIp, Constant.serverPort).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
// 客户端链接代理服务器成功
Channel channel = future.channel();
if (StringUtil.isNullOrEmpty(vid)) {
// 告诉服务端这条连接是client的连接
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_CONNECT);
myMsg.setData("client".getBytes());
channel.writeAndFlush(myMsg);
Constant.proxyChannel = channel;
} else {
// 告诉服务端这条连接是vid的连接
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_CONNECT);
myMsg.setData(vid.getBytes());
channel.writeAndFlush(myMsg);
// 客户端绑定通道关系
Constant.vpc.put(vid, channel);
channel.attr(Constant.VID).set(vid);
Channel realChannel = Constant.vrc.get(vid);
if (null != realChannel) {
realChannel.config().setOption(ChannelOption.AUTO_READ, true);
}
}
}
}
});
}
}

129
netty-proxy-client/src/main/java/com/luck/client/RealHandler.java
View File

@ -1,75 +1,74 @@
package com.luck.client;
import com.luck.msg.MyMsg;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.internal.StringUtil;
public class RealHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
public void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) {
// 客户读取到真实服务数据了
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_TRANSFER);
myMsg.setData(bytes);
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel proxyChannel = Constant.vpc.get(vid);
if (null != proxyChannel) {
proxyChannel.writeAndFlush(myMsg);
}
// 客户端发送真实数据到代理了
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
super.channelActive(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelInactive(ctx);
return;
}
Channel proxyChannel = Constant.vpc.get(vid);
if (proxyChannel != null) {
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_DISCONNECT);
myMsg.setData(vid.getBytes());
proxyChannel.writeAndFlush(myMsg);
}
super.channelInactive(ctx);
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelWritabilityChanged(ctx);
return;
}
Channel proxyChannel = Constant.vpc.get(vid);
if (proxyChannel != null) {
proxyChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
@Override
public void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) {
// 客户读取到真实服务数据了
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_TRANSFER);
myMsg.setData(bytes);
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel proxyChannel = Constant.vpc.get(vid);
if (null != proxyChannel) {
proxyChannel.writeAndFlush(myMsg);
}
// 客户端发送真实数据到代理了
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
super.channelActive(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelInactive(ctx);
return;
}
Channel proxyChannel = Constant.vpc.get(vid);
if (proxyChannel != null) {
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_DISCONNECT);
myMsg.setData(vid.getBytes());
proxyChannel.writeAndFlush(myMsg);
}
super.channelInactive(ctx);
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelWritabilityChanged(ctx);
return;
}
Channel proxyChannel = Constant.vpc.get(vid);
if (proxyChannel != null) {
proxyChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
}

108
netty-proxy-client/src/main/java/com/luck/client/RealSocket.java
View File

@ -1,65 +1,59 @@
package com.luck.client;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.internal.StringUtil;
public class RealSocket {
static EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
/**
* 连接真实服务
*
* @param vid 访客ID
* @return
*/
public static Channel connectRealServer(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return null;
}
Channel channel = Constant.vrc.get(vid);
if (null == channel) {
newConnect(vid);
channel = Constant.vrc.get(vid);
}
return channel;
}
private static void newConnect(String vid) {
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new RealHandler());
}
});
bootstrap.connect("127.0.0.1", Constant.realPort).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
// 客户端链接真实服务成功
future.channel().config().setOption(ChannelOption.AUTO_READ, false);
future.channel().attr(Constant.VID).set(vid);
Constant.vrc.put(vid, future.channel());
ProxySocket.connectProxyServer(vid);
}
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
static EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
/**
* 连接真实服务
*
* @param vid 访客ID
* @return
*/
public static Channel connectRealServer(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return null;
}
Channel channel = Constant.vrc.get(vid);
if (null == channel) {
newConnect(vid);
channel = Constant.vrc.get(vid);
}
return channel;
}
private static void newConnect(String vid) {
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new RealHandler());
}
});
bootstrap.connect("127.0.0.1", Constant.realPort).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
// 客户端链接真实服务成功
future.channel().config().setOption(ChannelOption.AUTO_READ, false);
future.channel().attr(Constant.VID).set(vid);
Constant.vrc.put(vid, future.channel());
ProxySocket.connectProxyServer(vid);
}
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
}

2
netty-proxy-client/src/main/java/com/luck/client/controller/ClientController.java
View File

@ -11,7 +11,7 @@ import org.springframework.web.bind.annotation.GetMapping;
public class ClientController {
@GetMapping("/version")
public Result version(){
public Result version() {
return ResultFactory.successOf("客户端版本");
}
}

4
netty-proxy-common/pom.xml
View File

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>netty-proxy</artifactId>

96
netty-proxy-common/src/main/java/com/luck/msg/MyMsg.java
View File

@ -1,46 +1,58 @@
package com.luck.msg;
import java.util.Arrays;
public class MyMsg {
/** 心跳 */
public static final byte TYPE_HEARTBEAT = 0X00;
/** 连接成功 */
public static final byte TYPE_CONNECT = 0X01;
/** 数据传输 */
public static final byte TYPE_TRANSFER = 0X02;
/** 连接断开 */
public static final byte TYPE_DISCONNECT = 0X09;
/** 数据类型 */
private byte type;
/** 消息传输数据 */
private byte[] data;
public byte getType() {
return type;
}
public void setType(byte type) {
this.type = type;
}
public byte[] getData() {
return data;
}
public void setData(byte[] data) {
this.data = data;
}
@Override
public String toString() {
return "MyMsg [type=" + type + ", data=" + Arrays.toString(data) + "]";
}
/**
* 心跳
*/
public static final byte TYPE_HEARTBEAT = 0X00;
/**
* 连接成功
*/
public static final byte TYPE_CONNECT = 0X01;
/**
* 数据传输
*/
public static final byte TYPE_TRANSFER = 0X02;
/**
* 连接断开
*/
public static final byte TYPE_DISCONNECT = 0X09;
/**
* 数据类型
*/
private byte type;
/**
* 消息传输数据
*/
private byte[] data;
public byte getType() {
return type;
}
public void setType(byte type) {
this.type = type;
}
public byte[] getData() {
return data;
}
public void setData(byte[] data) {
this.data = data;
}
@Override
public String toString() {
return "MyMsg [type=" + type + ", data=" + Arrays.toString(data) + "]";
}
}

70
netty-proxy-common/src/main/java/com/luck/msg/MyMsgDecoder.java
View File

@ -1,41 +1,41 @@
package com.luck.msg;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
public class MyMsgDecoder extends LengthFieldBasedFrameDecoder {
public MyMsgDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment,
int initialBytesToStrip) {
super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
}
public MyMsgDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment,
int initialBytesToStrip, boolean failFast) {
super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, failFast);
}
@Override
protected MyMsg decode(ChannelHandlerContext ctx, ByteBuf in2) throws Exception {
ByteBuf in = (ByteBuf) super.decode(ctx, in2);
if (in == null) {
return null;
}
if (in.readableBytes() < 4) {
return null;
}
MyMsg myMsg = new MyMsg();
int dataLength = in.readInt();
byte type = in.readByte();
myMsg.setType(type);
byte[] data = new byte[dataLength - 5];
in.readBytes(data);
myMsg.setData(data);
in.release();
return myMsg;
}
public MyMsgDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment,
int initialBytesToStrip) {
super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
}
public MyMsgDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment,
int initialBytesToStrip, boolean failFast) {
super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, failFast);
}
@Override
protected MyMsg decode(ChannelHandlerContext ctx, ByteBuf in2) throws Exception {
ByteBuf in = (ByteBuf) super.decode(ctx, in2);
if (in == null) {
return null;
}
if (in.readableBytes() < 4) {
return null;
}
MyMsg myMsg = new MyMsg();
int dataLength = in.readInt();
byte type = in.readByte();
myMsg.setType(type);
byte[] data = new byte[dataLength - 5];
in.readBytes(data);
myMsg.setData(data);
in.release();
return myMsg;
}
}

44
netty-proxy-common/src/main/java/com/luck/msg/MyMsgEncoder.java
View File

@ -1,28 +1,28 @@
package com.luck.msg;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
public class MyMsgEncoder extends MessageToByteEncoder<MyMsg> {
public MyMsgEncoder() {
}
@Override
protected void encode(ChannelHandlerContext ctx, MyMsg msg, ByteBuf out) throws Exception {
int bodyLength = 5;
if (msg.getData() != null) {
bodyLength += msg.getData().length;
}
out.writeInt(bodyLength);
out.writeByte(msg.getType());
if (msg.getData() != null) {
out.writeBytes(msg.getData());
}
}
public MyMsgEncoder() {
}
@Override
protected void encode(ChannelHandlerContext ctx, MyMsg msg, ByteBuf out) throws Exception {
int bodyLength = 5;
if (msg.getData() != null) {
bodyLength += msg.getData().length;
}
out.writeInt(bodyLength);
out.writeByte(msg.getType());
if (msg.getData() != null) {
out.writeBytes(msg.getData());
}
}
}

4
netty-proxy-server/pom.xml
View File

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>netty-proxy</artifactId>

216
netty-proxy-server/src/main/java/com/luck/server/ClientHandler.java
View File

@ -1,119 +1,113 @@
package com.luck.server;
import com.luck.msg.MyMsg;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.internal.StringUtil;
public class ClientHandler extends SimpleChannelInboundHandler<MyMsg> {
@Override
public void channelRead0(ChannelHandlerContext ctx, MyMsg myMsg) {
// 代理服务器读到客户端数据了
byte type = myMsg.getType();
switch (type) {
case MyMsg.TYPE_HEARTBEAT:
MyMsg hb = new MyMsg();
hb.setType(MyMsg.TYPE_HEARTBEAT);
ctx.channel().writeAndFlush(hb);
break;
case MyMsg.TYPE_CONNECT:
String vid = new String(myMsg.getData());
if (StringUtil.isNullOrEmpty(vid) || "client".equals(vid)) {
Constant.clientChannel = ctx.channel();
} else {
// 绑定访客和客户端的连接
Channel visitorChannel = Constant.vvc.get(vid);
if (null != visitorChannel) {
ctx.channel().attr(Constant.VID).set(vid);
Constant.vcc.put(vid, ctx.channel());
// 通道绑定完成可以读取访客数据
visitorChannel.config().setOption(ChannelOption.AUTO_READ, true);
}
}
break;
case MyMsg.TYPE_DISCONNECT:
String disVid = new String(myMsg.getData());
Constant.clearVccVvcAndClose(disVid);
break;
case MyMsg.TYPE_TRANSFER:
// 把数据转到用户服务
ByteBuf buf = ctx.alloc().buffer(myMsg.getData().length);
buf.writeBytes(myMsg.getData());
String visitorId = ctx.channel().attr(Constant.VID).get();
Channel vchannel = Constant.vvc.get(visitorId);
if (null != vchannel) {
vchannel.writeAndFlush(buf);
}
break;
default:
// 操作有误
}
// 代理服务器发送数据到用户了
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if(StringUtil.isNullOrEmpty(vid)) {
super.channelWritabilityChanged(ctx);
return;
}
Channel visitorChannel = Constant.vvc.get(vid);
if (visitorChannel != null) {
visitorChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelInactive(ctx);
return;
}
Channel visitorChannel = Constant.vvc.get(vid);
if (visitorChannel != null && visitorChannel.isActive()) {
// 数据发送完成后再关闭连接，解决http1.0数据传输问题
visitorChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
visitorChannel.close();
} else {
ctx.channel().close();
}
Constant.clearVccVvc(vid);
super.channelInactive(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
switch (event.state()) {
case READER_IDLE:
ctx.channel().close();
break;
case WRITER_IDLE:
break;
case ALL_IDLE:
break;
}
}
}
@Override
public void channelRead0(ChannelHandlerContext ctx, MyMsg myMsg) {
// 代理服务器读到客户端数据了
byte type = myMsg.getType();
switch (type) {
case MyMsg.TYPE_HEARTBEAT:
MyMsg hb = new MyMsg();
hb.setType(MyMsg.TYPE_HEARTBEAT);
ctx.channel().writeAndFlush(hb);
break;
case MyMsg.TYPE_CONNECT:
String vid = new String(myMsg.getData());
if (StringUtil.isNullOrEmpty(vid) || "client".equals(vid)) {
Constant.clientChannel = ctx.channel();
} else {
// 绑定访客和客户端的连接
Channel visitorChannel = Constant.vvc.get(vid);
if (null != visitorChannel) {
ctx.channel().attr(Constant.VID).set(vid);
Constant.vcc.put(vid, ctx.channel());
// 通道绑定完成可以读取访客数据
visitorChannel.config().setOption(ChannelOption.AUTO_READ, true);
}
}
break;
case MyMsg.TYPE_DISCONNECT:
String disVid = new String(myMsg.getData());
Constant.clearVccVvcAndClose(disVid);
break;
case MyMsg.TYPE_TRANSFER:
// 把数据转到用户服务
ByteBuf buf = ctx.alloc().buffer(myMsg.getData().length);
buf.writeBytes(myMsg.getData());
String visitorId = ctx.channel().attr(Constant.VID).get();
Channel vchannel = Constant.vvc.get(visitorId);
if (null != vchannel) {
vchannel.writeAndFlush(buf);
}
break;
default:
// 操作有误
}
// 代理服务器发送数据到用户了
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelWritabilityChanged(ctx);
return;
}
Channel visitorChannel = Constant.vvc.get(vid);
if (visitorChannel != null) {
visitorChannel.config().setOption(ChannelOption.AUTO_READ, ctx.channel().isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelInactive(ctx);
return;
}
Channel visitorChannel = Constant.vvc.get(vid);
if (visitorChannel != null && visitorChannel.isActive()) {
// 数据发送完成后再关闭连接，解决http1.0数据传输问题
visitorChannel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
visitorChannel.close();
} else {
ctx.channel().close();
}
Constant.clearVccVvc(vid);
super.channelInactive(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
switch (event.state()) {
case READER_IDLE:
ctx.channel().close();
break;
case WRITER_IDLE:
break;
case ALL_IDLE:
break;
}
}
}
}

142
netty-proxy-server/src/main/java/com/luck/server/Constant.java
View File

@ -1,72 +1,82 @@
package com.luck.server;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
import io.netty.util.internal.StringUtil;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class Constant {
/** 客户端服务channel */
public static Channel clientChannel = null;
/** 绑定channel_id */
public static final AttributeKey<String> VID = AttributeKey.newInstance("vid");
/** 访客，客户服务channel */
public static Map<String, Channel> vcc = new ConcurrentHashMap<>();
/** 访客，访客服务channel */
public static Map<String, Channel> vvc = new ConcurrentHashMap<>();
/** 服务代理端口 */
public static int visitorPort = 16002;
/** 服务端口 */
public static int serverPort = 16001;
/**
* 清除连接
*
* @param vid 访客ID
*/
public static void clearVccVvc(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel clientChannel = vcc.get(vid);
if (null != clientChannel) {
clientChannel.attr(VID).set(null);
vcc.remove(vid);
}
Channel visitorChannel = vvc.get(vid);
if (null != visitorChannel) {
visitorChannel.attr(VID).set(null);
vvc.remove(vid);
}
}
/**
* 清除关闭连接
*
* @param vid 访客ID
*/
public static void clearVccVvcAndClose(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel clientChannel = vcc.get(vid);
if (null != clientChannel) {
clientChannel.attr(VID).set(null);
vcc.remove(vid);
clientChannel.close();
}
Channel visitorChannel = vvc.get(vid);
if (null != visitorChannel) {
visitorChannel.attr(VID).set(null);
vvc.remove(vid);
visitorChannel.close();
}
}
/**
* 绑定channel_id
*/
public static final AttributeKey<String> VID = AttributeKey.newInstance("vid");
/**
* 客户端服务channel
*/
public static Channel clientChannel = null;
/**
* 访客，客户服务channel
*/
public static Map<String, Channel> vcc = new ConcurrentHashMap<>();
/**
* 访客，访客服务channel
*/
public static Map<String, Channel> vvc = new ConcurrentHashMap<>();
/**
* 服务代理端口
*/
public static int visitorPort = 16002;
/**
* 服务端口
*/
public static int serverPort = 16001;
/**
* 清除连接
*
* @param vid 访客ID
*/
public static void clearVccVvc(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel clientChannel = vcc.get(vid);
if (null != clientChannel) {
clientChannel.attr(VID).set(null);
vcc.remove(vid);
}
Channel visitorChannel = vvc.get(vid);
if (null != visitorChannel) {
visitorChannel.attr(VID).set(null);
vvc.remove(vid);
}
}
/**
* 清除关闭连接
*
* @param vid 访客ID
*/
public static void clearVccVvcAndClose(String vid) {
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
Channel clientChannel = vcc.get(vid);
if (null != clientChannel) {
clientChannel.attr(VID).set(null);
vcc.remove(vid);
clientChannel.close();
}
Channel visitorChannel = vvc.get(vid);
if (null != visitorChannel) {
visitorChannel.attr(VID).set(null);
vvc.remove(vid);
visitorChannel.close();
}
}
}

106
netty-proxy-server/src/main/java/com/luck/server/ServerSocket.java
View File

@ -1,65 +1,61 @@
package com.luck.server;
import com.luck.msg.MyMsgDecoder;
import com.luck.msg.MyMsgEncoder;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
public class ServerSocket {
private static EventLoopGroup bossGroup = new NioEventLoopGroup();
private static EventLoopGroup workerGroup = new NioEventLoopGroup();
private static ChannelFuture channelFuture;
/**
* 启动服务端
* @throws Exception
*/
public static void startServer() throws Exception {
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMsgDecoder(Integer.MAX_VALUE, 0, 4, -4, 0));
pipeline.addLast(new MyMsgEncoder());
pipeline.addLast(new IdleStateHandler(40, 10, 0));
pipeline.addLast(new ClientHandler());
}
});
channelFuture = b.bind(Constant.serverPort).sync();
channelFuture.addListener((ChannelFutureListener) channelFuture -> {
// 服务器已启动
});
channelFuture.channel().closeFuture().sync();
} finally {
shutdown();
// 服务器已关闭
}
}
public static void shutdown() {
if (channelFuture != null) {
channelFuture.channel().close().syncUninterruptibly();
}
if ((bossGroup != null) && (!bossGroup.isShutdown())) {
bossGroup.shutdownGracefully();
}
if ((workerGroup != null) && (!workerGroup.isShutdown())) {
workerGroup.shutdownGracefully();
}
}
private static EventLoopGroup bossGroup = new NioEventLoopGroup();
private static EventLoopGroup workerGroup = new NioEventLoopGroup();
private static ChannelFuture channelFuture;
/**
* 启动服务端
*
* @throws Exception
*/
public static void startServer() throws Exception {
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMsgDecoder(Integer.MAX_VALUE, 0, 4, -4, 0));
pipeline.addLast(new MyMsgEncoder());
pipeline.addLast(new IdleStateHandler(40, 10, 0));
pipeline.addLast(new ClientHandler());
}
});
channelFuture = b.bind(Constant.serverPort).sync();
channelFuture.addListener((ChannelFutureListener) channelFuture -> {
// 服务器已启动
});
channelFuture.channel().closeFuture().sync();
} finally {
shutdown();
// 服务器已关闭
}
}
public static void shutdown() {
if (channelFuture != null) {
channelFuture.channel().close().syncUninterruptibly();
}
if ((bossGroup != null) && (!bossGroup.isShutdown())) {
bossGroup.shutdownGracefully();
}
if ((workerGroup != null) && (!workerGroup.isShutdown())) {
workerGroup.shutdownGracefully();
}
}
}

175
netty-proxy-server/src/main/java/com/luck/server/VisitorHandler.java
View File

@ -1,98 +1,97 @@
package com.luck.server;
import java.util.UUID;
import com.luck.msg.MyMsg;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOption;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.internal.StringUtil;
import java.util.UUID;
public class VisitorHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 访客连接上代理服务器了
Channel visitorChannel = ctx.channel();
// 先不读取访客数据
visitorChannel.config().setOption(ChannelOption.AUTO_READ, false);
// 生成访客ID
String vid = UUID.randomUUID().toString();
// 绑定访客通道
visitorChannel.attr(Constant.VID).set(vid);
Constant.vvc.put(vid, visitorChannel);
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_CONNECT);
myMsg.setData(vid.getBytes());
Constant.clientChannel.writeAndFlush(myMsg);
super.channelActive(ctx);
}
@Override
public void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_TRANSFER);
myMsg.setData(bytes);
// 代理服务器发送数据到客户端了
Channel clientChannel = Constant.vcc.get(vid);
clientChannel.writeAndFlush(myMsg);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelInactive(ctx);
return;
}
Channel clientChannel = Constant.vcc.get(vid);
if (clientChannel != null && clientChannel.isActive()) {
clientChannel.config().setOption(ChannelOption.AUTO_READ, true);
// 通知客户端，访客连接已经断开
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_DISCONNECT);
myMsg.setData(vid.getBytes());
clientChannel.writeAndFlush(myMsg);
}
Constant.clearVccVvc(vid);
super.channelInactive(ctx);
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
Channel visitorChannel = ctx.channel();
String vid = visitorChannel.attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelWritabilityChanged(ctx);
return;
}
Channel clientChannel = Constant.vcc.get(vid);
if (clientChannel != null) {
clientChannel.config().setOption(ChannelOption.AUTO_READ, visitorChannel.isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 访客连接上代理服务器了
Channel visitorChannel = ctx.channel();
// 先不读取访客数据
visitorChannel.config().setOption(ChannelOption.AUTO_READ, false);
// 生成访客ID
String vid = UUID.randomUUID().toString();
// 绑定访客通道
visitorChannel.attr(Constant.VID).set(vid);
Constant.vvc.put(vid, visitorChannel);
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_CONNECT);
myMsg.setData(vid.getBytes());
Constant.clientChannel.writeAndFlush(myMsg);
super.channelActive(ctx);
}
@Override
public void channelRead0(ChannelHandlerContext ctx, ByteBuf buf) {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
return;
}
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_TRANSFER);
myMsg.setData(bytes);
// 代理服务器发送数据到客户端了
Channel clientChannel = Constant.vcc.get(vid);
clientChannel.writeAndFlush(myMsg);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String vid = ctx.channel().attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelInactive(ctx);
return;
}
Channel clientChannel = Constant.vcc.get(vid);
if (clientChannel != null && clientChannel.isActive()) {
clientChannel.config().setOption(ChannelOption.AUTO_READ, true);
// 通知客户端，访客连接已经断开
MyMsg myMsg = new MyMsg();
myMsg.setType(MyMsg.TYPE_DISCONNECT);
myMsg.setData(vid.getBytes());
clientChannel.writeAndFlush(myMsg);
}
Constant.clearVccVvc(vid);
super.channelInactive(ctx);
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
Channel visitorChannel = ctx.channel();
String vid = visitorChannel.attr(Constant.VID).get();
if (StringUtil.isNullOrEmpty(vid)) {
super.channelWritabilityChanged(ctx);
return;
}
Channel clientChannel = Constant.vcc.get(vid);
if (clientChannel != null) {
clientChannel.config().setOption(ChannelOption.AUTO_READ, visitorChannel.isWritable());
}
super.channelWritabilityChanged(ctx);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}

52
netty-proxy-server/src/main/java/com/luck/server/VisitorSocket.java
View File

@ -1,5 +1,5 @@
package com.luck.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelInitializer;
@ -8,30 +8,30 @@ import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class VisitorSocket {
private static EventLoopGroup bossGroup = new NioEventLoopGroup();
private static EventLoopGroup workerGroup = new NioEventLoopGroup();
/**
* 启动服务代理
*
* @throws Exception
*/
public static void startServer() throws Exception {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new ChannelDuplexHandler());
pipeline.addLast(new VisitorHandler());
}
});
b.bind(Constant.visitorPort).get();
}
private static EventLoopGroup bossGroup = new NioEventLoopGroup();
private static EventLoopGroup workerGroup = new NioEventLoopGroup();
/**
* 启动服务代理
*
* @throws Exception
*/
public static void startServer() throws Exception {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new ChannelDuplexHandler());
pipeline.addLast(new VisitorHandler());
}
});
b.bind(Constant.visitorPort).get();
}
}

2
netty-proxy-server/src/main/java/com/luck/server/controller/ServerController.java
View File

@ -11,7 +11,7 @@ import org.springframework.web.bind.annotation.GetMapping;
public class ServerController {
@GetMapping("/version")
public Result version(){
public Result version() {
return ResultFactory.successOf("服务端端版本");
}
}

2
pom.xml
View File

@ -1,5 +1,5 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>