From df2ced99a0287096adffaed646fbf5c70fb2c1f7 Mon Sep 17 00:00:00 2001 From: aoshiguchen <1052045476@qq.com> Date: Wed, 26 Oct 2022 22:00:13 +0800 Subject: [PATCH] =?UTF-8?q?1=E3=80=81=E6=9C=8D=E5=8A=A1=E7=AB=AF=E9=83=A8?= =?UTF-8?q?=E5=88=86=E4=BB=A3=E7=A0=81=E9=87=8D=E6=9E=84=202=E3=80=81?= =?UTF-8?q?=E6=B5=81=E9=87=8F=E6=8A=A5=E8=A1=A8=E5=AD=97=E6=AE=B5=E8=B0=83?= =?UTF-8?q?=E6=95=B4=203=E3=80=81=E5=A2=9E=E5=8A=A0=E6=B5=81=E9=87=8F?= =?UTF-8?q?=E7=BB=9F=E8=AE=A1=E7=9B=B8=E5=85=B3=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 2 +- docs/Channel.MD | 4 +- .../client/core/RealServerChannelHandler.java | 6 +- .../neutrino/proxy/core/ProxyMessage.java | 4 +- .../server/dal/FlowReportMinuteMapper.java | 2 + .../proxy/server/dal/LicenseMapper.java | 5 +- .../server/dal/entity/FlowReportMinuteDO.java | 10 +-- .../server/job/FlowReportForMinuteJob.java | 39 ++++++++- .../proxy/core/VisitorChannelHandler.java | 27 ++++--- .../domain/VisitorChannelAttachInfo.java | 4 + .../handler/ProxyMessageAuthHandler.java | 13 ++- .../handler/ProxyMessageConnectHandler.java | 10 +-- .../handler/ProxyMessageTransferHandler.java | 14 +++- .../server/service/FlowReportService.java | 79 +++++++++++++++++++ .../neutrino/proxy/server/util/ProxyUtil.java | 25 +++--- .../src/main/resources/sql/init-structure.sql | 4 - 16 files changed, 191 insertions(+), 57 deletions(-) create mode 100644 neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/FlowReportService.java diff --git a/.gitignore b/.gitignore index 83909f75..44013869 100644 --- a/.gitignore +++ b/.gitignore @@ -3,7 +3,7 @@ *.class data.db* .neutrino-proxy.license -.neutrino-proxy-client.json +.neutrino-proxy-client.json* lib/* diff --git a/docs/Channel.MD b/docs/Channel.MD index b28985cc..5c31b33e 100644 --- a/docs/Channel.MD +++ b/docs/Channel.MD @@ -18,9 +18,9 @@ ## 代理数据传输的通道(ProxyChannel) - 该channel负责完成内网被代理服务与代理服务端之间的数据转发任务。 -- 每个客户端维护一个`ProxyChannel`的缓存队列，需要时从该队列中取，当取不到是，直接新建一个`ProxyChannel`返回。 +- 每个客户端维护一个`ProxyChannel`的缓存队列，需要时从该队列中取，当取不到时，直接新建一个`ProxyChannel`返回。 当一个`ProxyChannel`实例用完后，需要归还到缓存队列中(`ProxyChannel`收到`DisConnect`指令时)。 -- 当`RealServerChannel`建立完成后，就会获取一个`ProxyChannel`,并绑定对应的`RealServerChannel`,并设置`RealServerChannel`为 +- 当`RealServerChannel`建立完成后，就会获取相关联的`ProxyChannel`,并与其绑定。设置`RealServerChannel`为 可读状态。然后通过`ProxyChannel`向服务端发送`Connect`指令。 diff --git a/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/RealServerChannelHandler.java b/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/RealServerChannelHandler.java index 24ef9aa4..056a52c0 100755 --- a/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/RealServerChannelHandler.java +++ b/neutrino-proxy-client/src/main/java/fun/asgc/neutrino/proxy/client/core/RealServerChannelHandler.java @@ -42,15 +42,15 @@ public class RealServerChannelHandler extends SimpleChannelInboundHandler list(); + @ResultType(LicenseDO.class) + @Select("select * from license") + List listAll(); + /** * 新增license * @param license diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/dal/entity/FlowReportMinuteDO.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/dal/entity/FlowReportMinuteDO.java index 1c729a51..437d3899 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/dal/entity/FlowReportMinuteDO.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/dal/entity/FlowReportMinuteDO.java @@ -36,7 +36,7 @@ import java.util.Date; @ToString @Accessors(chain = true) @Data -@Table("job_info") +@Table("flow_report_minute") public class FlowReportMinuteDO { @Id private Integer id; @@ -48,18 +48,14 @@ public class FlowReportMinuteDO { * licenseId */ private Integer licenseId; - /** - * ip - */ - private String ip; /** * 写入字节数 */ - private Long writeBytes; + private Integer writeBytes; /** * 读取字节数 */ - private Long readBytes; + private Integer readBytes; /** * 报表统计时间 * yyyy-MM-dd HH:mm diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/job/FlowReportForMinuteJob.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/job/FlowReportForMinuteJob.java index c5956bb0..fef65ec4 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/job/FlowReportForMinuteJob.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/job/FlowReportForMinuteJob.java @@ -21,12 +21,23 @@ */ package fun.asgc.neutrino.proxy.server.job; +import fun.asgc.neutrino.core.annotation.Autowired; import fun.asgc.neutrino.core.annotation.Component; import fun.asgc.neutrino.core.annotation.NonIntercept; import fun.asgc.neutrino.core.quartz.IJobHandler; import fun.asgc.neutrino.core.quartz.annotation.JobHandler; +import fun.asgc.neutrino.core.util.CollectionUtil; +import fun.asgc.neutrino.core.util.DateUtil; +import fun.asgc.neutrino.proxy.server.dal.FlowReportMinuteMapper; +import fun.asgc.neutrino.proxy.server.dal.LicenseMapper; +import fun.asgc.neutrino.proxy.server.dal.entity.FlowReportMinuteDO; +import fun.asgc.neutrino.proxy.server.dal.entity.LicenseDO; +import fun.asgc.neutrino.proxy.server.service.FlowReportService; import lombok.extern.slf4j.Slf4j; +import java.util.Date; +import java.util.List; + /** * 流量统计报表 - 分钟级别 * @author: aoshiguchen @@ -38,8 +49,34 @@ import lombok.extern.slf4j.Slf4j; @JobHandler(name = "FlowReportForMinuteJob", cron = "0 */1 * * * ?", param = "") public class FlowReportForMinuteJob implements IJobHandler { + @Autowired + private FlowReportService flowReportService; + @Autowired + private LicenseMapper licenseMapper; + @Autowired + private FlowReportMinuteMapper flowReportMinuteMapper; + @Override public void execute(String param) throws Exception { - // TODO aoshiguchen + List list = licenseMapper.listAll(); + if (CollectionUtil.isEmpty(list)) { + return; + } + Date now = new Date(); + for (LicenseDO item : list) { + Integer writeBytes = flowReportService.getAndResetWriteByte(item.getId()); + Integer readBytes = flowReportService.getAndResetReadByte(item.getId()); + if (writeBytes == 0 && readBytes == 0) { + continue; + } + FlowReportMinuteDO flowReportMinuteDO = new FlowReportMinuteDO(); + flowReportMinuteDO.setUserId(item.getUserId()); + flowReportMinuteDO.setLicenseId(item.getId()); + flowReportMinuteDO.setWriteBytes(writeBytes); + flowReportMinuteDO.setReadBytes(readBytes); + flowReportMinuteDO.setDate(DateUtil.format(now, "yyyy-MM-dd HH:mm")); + flowReportMinuteDO.setCreateTime(now); + // TODO insert + } } } diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/VisitorChannelHandler.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/VisitorChannelHandler.java index 0ca26919..de829dca 100755 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/VisitorChannelHandler.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/core/VisitorChannelHandler.java @@ -22,8 +22,11 @@ package fun.asgc.neutrino.proxy.server.proxy.core; +import fun.asgc.neutrino.core.util.BeanManager; import fun.asgc.neutrino.proxy.core.Constants; import fun.asgc.neutrino.proxy.core.ProxyMessage; +import fun.asgc.neutrino.proxy.server.proxy.domain.VisitorChannelAttachInfo; +import fun.asgc.neutrino.proxy.server.service.FlowReportService; import fun.asgc.neutrino.proxy.server.util.ProxyUtil; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; @@ -41,7 +44,7 @@ import java.util.concurrent.atomic.AtomicLong; */ public class VisitorChannelHandler extends SimpleChannelInboundHandler { - private static AtomicLong userIdProducer = new AtomicLong(0); + private static AtomicLong visitorIdProducer = new AtomicLong(0); @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { @@ -65,25 +68,29 @@ public class VisitorChannelHandler extends SimpleChannelInboundHandler buf.readBytes(bytes); String visitorId = ProxyUtil.getVisitorIdByChannel(visitorChannel); proxyChannel.writeAndFlush(ProxyMessage.buildTransferMessage(visitorId, bytes)); + + // 增加流量计数 + VisitorChannelAttachInfo visitorChannelAttachInfo = ProxyUtil.getAttachInfo(visitorChannel); + BeanManager.getBean(FlowReportService.class).addWriteByte(visitorChannelAttachInfo.getLicenseId(), buf.readableBytes()); } } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { - Channel userChannel = ctx.channel(); - InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress(); + Channel visitorChannel = ctx.channel(); + InetSocketAddress sa = (InetSocketAddress) visitorChannel.localAddress(); Channel cmdChannel = ProxyUtil.getCmdChannelByServerPort(sa.getPort()); if (cmdChannel == null) { // 该端口还没有代理客户端 ctx.channel().close(); } else { - String userId = newUserId(); + String visitorId = newVisitorId(); String lanInfo = ProxyUtil.getClientLanInfoByServerPort(sa.getPort()); // 用户连接到代理服务器时，设置用户连接不可读，等待代理后端服务器连接成功后再改变为可读状态 - userChannel.config().setOption(ChannelOption.AUTO_READ, false); - ProxyUtil.addUserChannelToCmdChannel(cmdChannel, userId, userChannel); - cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(userId).setData(lanInfo.getBytes())); + visitorChannel.config().setOption(ChannelOption.AUTO_READ, false); + ProxyUtil.addVisitorChannelToCmdChannel(cmdChannel, visitorId, visitorChannel); + cmdChannel.writeAndFlush(ProxyMessage.buildConnectMessage(visitorId).setData(lanInfo.getBytes())); } super.channelActive(ctx); @@ -144,11 +151,11 @@ public class VisitorChannelHandler extends SimpleChannelInboundHandler } /** - * 为用户连接产生ID + * 为访问者连接产生ID * * @return */ - private static String newUserId() { - return String.valueOf(userIdProducer.incrementAndGet()); + private static String newVisitorId() { + return String.valueOf(visitorIdProducer.incrementAndGet()); } } diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/VisitorChannelAttachInfo.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/VisitorChannelAttachInfo.java index 8cca08a0..eefa1c05 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/VisitorChannelAttachInfo.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/domain/VisitorChannelAttachInfo.java @@ -34,6 +34,10 @@ import lombok.experimental.Accessors; public class VisitorChannelAttachInfo { private String visitorId; private String lanInfo; + /** + * licenseId + */ + private Integer licenseId; /** * ip地址 */ diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java index af38e833..51de8825 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageAuthHandler.java @@ -29,19 +29,16 @@ import fun.asgc.neutrino.core.annotation.NonIntercept; import fun.asgc.neutrino.core.util.CollectionUtil; import fun.asgc.neutrino.core.util.StringUtil; import fun.asgc.neutrino.proxy.core.*; -import fun.asgc.neutrino.proxy.server.constant.*; import fun.asgc.neutrino.proxy.server.base.proxy.ProxyConfig; -import fun.asgc.neutrino.proxy.server.proxy.core.BytesMetricsHandler; -import fun.asgc.neutrino.proxy.server.proxy.core.VisitorChannelHandler; +import fun.asgc.neutrino.proxy.server.constant.EnableStatusEnum; import fun.asgc.neutrino.proxy.server.dal.entity.LicenseDO; import fun.asgc.neutrino.proxy.server.dal.entity.PortMappingDO; import fun.asgc.neutrino.proxy.server.dal.entity.UserDO; +import fun.asgc.neutrino.proxy.server.proxy.core.BytesMetricsHandler; +import fun.asgc.neutrino.proxy.server.proxy.core.VisitorChannelHandler; import fun.asgc.neutrino.proxy.server.proxy.domain.CmdChannelAttachInfo; import fun.asgc.neutrino.proxy.server.proxy.domain.ProxyMapping; -import fun.asgc.neutrino.proxy.server.service.LicenseService; -import fun.asgc.neutrino.proxy.server.service.PortMappingService; -import fun.asgc.neutrino.proxy.server.service.ProxyMutualService; -import fun.asgc.neutrino.proxy.server.service.UserService; +import fun.asgc.neutrino.proxy.server.service.*; import fun.asgc.neutrino.proxy.server.util.ProxyUtil; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; @@ -80,6 +77,8 @@ public class ProxyMessageAuthHandler implements ProxyMessageHandler { private PortMappingService portMappingService; @Autowired private ProxyMutualService proxyMutualService; + @Autowired + private FlowReportService flowReportService; @Override public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) { diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageConnectHandler.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageConnectHandler.java index 6d36f234..3f56e8e1 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageConnectHandler.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageConnectHandler.java @@ -97,14 +97,14 @@ public class ProxyMessageConnectHandler implements ProxyMessageHandler { return; } - Channel userChannel = ProxyUtil.getUserChannel(cmdChannel, visitorId); - if (userChannel != null) { + Channel visitorChannel = ProxyUtil.getVisitorChannel(cmdChannel, visitorId); + if (visitorChannel != null) { ctx.channel().attr(Constants.VISITOR_ID).set(visitorId); ctx.channel().attr(Constants.LICENSE_ID).set(licenseDO.getId()); - ctx.channel().attr(Constants.NEXT_CHANNEL).set(userChannel); - userChannel.attr(Constants.NEXT_CHANNEL).set(ctx.channel()); + ctx.channel().attr(Constants.NEXT_CHANNEL).set(visitorChannel); + visitorChannel.attr(Constants.NEXT_CHANNEL).set(ctx.channel()); // 代理客户端与后端服务器连接成功，修改用户连接为可读状态 - userChannel.config().setOption(ChannelOption.AUTO_READ, true); + visitorChannel.config().setOption(ChannelOption.AUTO_READ, true); } } diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageTransferHandler.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageTransferHandler.java index edb176d5..367e07fd 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageTransferHandler.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/proxy/handler/ProxyMessageTransferHandler.java @@ -25,10 +25,14 @@ package fun.asgc.neutrino.proxy.server.proxy.handler; import fun.asgc.neutrino.core.annotation.Component; import fun.asgc.neutrino.core.annotation.Match; import fun.asgc.neutrino.core.annotation.NonIntercept; +import fun.asgc.neutrino.core.util.BeanManager; import fun.asgc.neutrino.proxy.core.Constants; import fun.asgc.neutrino.proxy.core.ProxyDataTypeEnum; import fun.asgc.neutrino.proxy.core.ProxyMessage; import fun.asgc.neutrino.proxy.core.ProxyMessageHandler; +import fun.asgc.neutrino.proxy.server.proxy.domain.VisitorChannelAttachInfo; +import fun.asgc.neutrino.proxy.server.service.FlowReportService; +import fun.asgc.neutrino.proxy.server.util.ProxyUtil; import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; @@ -45,11 +49,15 @@ public class ProxyMessageTransferHandler implements ProxyMessageHandler { @Override public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) { - Channel userChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get(); - if (null != userChannel) { + Channel visitorChannel = ctx.channel().attr(Constants.NEXT_CHANNEL).get(); + if (null != visitorChannel) { ByteBuf buf = ctx.alloc().buffer(proxyMessage.getData().length); buf.writeBytes(proxyMessage.getData()); - userChannel.writeAndFlush(buf); + visitorChannel.writeAndFlush(buf); + + // 增加流量计数 + VisitorChannelAttachInfo visitorChannelAttachInfo = ProxyUtil.getAttachInfo(visitorChannel); + BeanManager.getBean(FlowReportService.class).addReadByte(visitorChannelAttachInfo.getLicenseId(), buf.readableBytes()); } } diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/FlowReportService.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/FlowReportService.java new file mode 100644 index 00000000..fdc2750a --- /dev/null +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/FlowReportService.java @@ -0,0 +1,79 @@ +/** + * Copyright (c) 2022 aoshiguchen + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ +package fun.asgc.neutrino.proxy.server.service; + +import fun.asgc.neutrino.core.annotation.Component; +import fun.asgc.neutrino.core.annotation.NonIntercept; +import fun.asgc.neutrino.core.util.LockUtil; +import lombok.extern.slf4j.Slf4j; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * 流量报表服务 + * @author: aoshiguchen + * @date: 2022/10/26 + */ +@Slf4j +@NonIntercept +@Component +public class FlowReportService { + private Map writeByteMap = new HashMap<>(); + private Map readByteMap = new HashMap<>(); + + private AtomicInteger getWriteByte(Integer licenseId) { + return LockUtil.doubleCheckProcessForNoException(() -> !writeByteMap.containsKey(licenseId), + licenseId, + () -> { + writeByteMap.put(licenseId, new AtomicInteger()); + }, + () -> writeByteMap.get(licenseId)); + } + + private AtomicInteger getReadByte(Integer licenseId) { + return LockUtil.doubleCheckProcessForNoException(() -> !readByteMap.containsKey(licenseId), + licenseId, + () -> { + readByteMap.put(licenseId, new AtomicInteger()); + }, + () -> readByteMap.get(licenseId)); + } + + public void addWriteByte(Integer licenseId, Integer writeByte) { + getWriteByte(licenseId).addAndGet(writeByte); + } + + public void addReadByte(Integer licenseId, Integer readByte) { + getReadByte(licenseId).addAndGet(readByte); + } + + public Integer getAndResetWriteByte(Integer licenseId) { + return getWriteByte(licenseId).getAndSet(0); + } + + public Integer getAndResetReadByte(Integer licenseId) { + return getReadByte(licenseId).getAndSet(0); + } + +} diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java index ce1fc2b5..531f2dad 100644 --- a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/util/ProxyUtil.java @@ -171,20 +171,23 @@ public class ProxyUtil { /** * 增加用户连接与代理客户端连接关系 * - * @param userId - * @param userChannel + * @param visitorId + * @param visitorChannel */ - public static void addUserChannelToCmdChannel(Channel cmdChannel, String userId, Channel userChannel) { - InetSocketAddress sa = (InetSocketAddress) userChannel.localAddress(); + public static void addVisitorChannelToCmdChannel(Channel cmdChannel, String visitorId, Channel visitorChannel) { + InetSocketAddress sa = (InetSocketAddress) visitorChannel.localAddress(); String lanInfo = getClientLanInfoByServerPort(sa.getPort()); - setAttachInfo(userChannel, new VisitorChannelAttachInfo() - .setVisitorId(userId) + CmdChannelAttachInfo cmdChannelAttachInfo = getAttachInfo(cmdChannel); + + setAttachInfo(visitorChannel, new VisitorChannelAttachInfo() + .setVisitorId(visitorId) .setLanInfo(lanInfo) - .setIp(ChannelUtil.getIP(userChannel)) + .setLicenseId(cmdChannelAttachInfo.getLicenseId()) + .setIp(ChannelUtil.getIP(visitorChannel)) ); userChannelMapLock.writeLock().lock(); try { - ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getVisitorChannelMap().put(userId, userChannel); + cmdChannelAttachInfo.getVisitorChannelMap().put(visitorId, visitorChannel); } finally { userChannelMapLock.writeLock().unlock(); } @@ -206,14 +209,14 @@ public class ProxyUtil { /** * 根据代理客户端连接与用户编号获取用户连接 * - * @param userId + * @param visitorId * @return */ - public static Channel getUserChannel(Channel cmdChannel, String userId) { + public static Channel getVisitorChannel(Channel cmdChannel, String visitorId) { if (null == cmdChannel || null == getAttachInfo(cmdChannel)) { return null; } - return ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getVisitorChannelMap().get(userId); + return ((CmdChannelAttachInfo)getAttachInfo(cmdChannel)).getVisitorChannelMap().get(visitorId); } /** diff --git a/neutrino-proxy-server/src/main/resources/sql/init-structure.sql b/neutrino-proxy-server/src/main/resources/sql/init-structure.sql index 7a19aa02..4d855f07 100644 --- a/neutrino-proxy-server/src/main/resources/sql/init-structure.sql +++ b/neutrino-proxy-server/src/main/resources/sql/init-structure.sql @@ -137,7 +137,6 @@ CREATE TABLE IF NOT EXISTS `flow_report_minute` ( `id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, `user_id` INTEGER(20) NOT NULL, `license_id` INTEGER(20) NOT NULL, - `ip` VARCHAR(50) NOT NULL, `write_bytes` INTEGER(20) NOT NULL, `read_bytes` INTEGER(20) NOT NULL, `date` VARCHAR(20) NOT NULL, @@ -153,7 +152,6 @@ CREATE TABLE IF NOT EXISTS `flow_report_hour` ( `id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, `user_id` INTEGER(20) NOT NULL, `license_id` INTEGER(20) NOT NULL, - `ip` VARCHAR(50) NOT NULL, `write_bytes` INTEGER(20) NOT NULL, `read_bytes` INTEGER(20) NOT NULL, `date` VARCHAR(20) NOT NULL, @@ -169,7 +167,6 @@ CREATE TABLE IF NOT EXISTS `flow_report_day` ( `id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, `user_id` INTEGER(20) NOT NULL, `license_id` INTEGER(20) NOT NULL, - `ip` VARCHAR(50) NOT NULL, `write_bytes` INTEGER(20) NOT NULL, `read_bytes` INTEGER(20) NOT NULL, `date` VARCHAR(20) NOT NULL, @@ -185,7 +182,6 @@ CREATE TABLE IF NOT EXISTS `flow_report_month` ( `id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT, `user_id` INTEGER(20) NOT NULL, `license_id` INTEGER(20) NOT NULL, - `ip` VARCHAR(50) NOT NULL, `write_bytes` INTEGER(20) NOT NULL, `read_bytes` INTEGER(20) NOT NULL, `date` VARCHAR(20) NOT NULL,