From 37ee8021becef8c2bdfbd01403fb043fb1a9db2f Mon Sep 17 00:00:00 2001 From: aoshiguchen <1052045476@qq.com> Date: Sun, 5 Feb 2023 12:02:54 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BD=BF=E7=94=A8=E4=B8=B4=E6=97=B6=E5=85=BC?= =?UTF-8?q?=E5=AE=B9=E6=96=B9=E6=A1=88=EF=BC=8C=E8=A7=A3=E5=86=B3jdbctempl?= =?UTF-8?q?ate=E6=96=B0=E5=A2=9E=E6=95=B0=E6=8D=AE=E6=97=B6=EF=BC=8C?= =?UTF-8?q?=E8=AE=BE=E7=BD=AE=E8=87=AA=E5=A2=9E=E4=B8=BB=E9=94=AE=E7=9A=84?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/db/template/JdbcOperations.java | 73 ++++++++++- .../core/db/template/JdbcTemplate.java | 32 ++++- .../handler/ProxyMessageAuthHandler.java | 58 +-------- .../server/service/VisitorChannelService.java | 114 ++++++++++++++++++ 4 files changed, 214 insertions(+), 63 deletions(-) create mode 100644 neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/VisitorChannelService.java diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/db/template/JdbcOperations.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/db/template/JdbcOperations.java index 483da757..4fe86a20 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/db/template/JdbcOperations.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/db/template/JdbcOperations.java @@ -21,15 +21,15 @@ */ package fun.asgc.neutrino.core.db.template; +import fun.asgc.neutrino.core.db.annotation.Id; +import fun.asgc.neutrino.core.util.CollectionUtil; import fun.asgc.neutrino.core.util.ReflectUtil; import fun.asgc.neutrino.core.util.TypeUtil; import java.lang.reflect.Field; import java.sql.*; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; /** * @@ -38,6 +38,8 @@ import java.util.Map; */ public class JdbcOperations { private static final JdbcOperations instance = new JdbcOperations(); + + private static final Map, Field> generateIdFieldMap = new ConcurrentHashMap<>(); private JdbcOperations() { @@ -85,6 +87,45 @@ public class JdbcOperations { }); } + /** + * 执行更新操作 + * 临时兼容返设主键问题 + * @param conn + * @param sql + * @param params + * @return + */ + public int executeUpdateByModel(final Connection conn , final String sql, final Object model, final Object[] params) throws SQLException { + return this.execute(new PreparedStatementJdbcCallback(){ + @Override + public Integer execute(PreparedStatement ps) throws SQLException { + Integer res = ps.executeUpdate(); + if (null != model) { + ResultSet resultSet = ps.getGeneratedKeys(); + if (resultSet.next()) { + Field field = getGenerateIdField(model.getClass()); + if (null != field) { + ReflectUtil.setFieldValue(field, model, resultSet.getInt(1)); + } + } + } + return res; + } + @Override + public Object[] getParams() { + return params; + } + @Override + public String getSql() { + return sql; + } + @Override + public Connection getConnection(){ + return conn; + } + }); + } + /** * 执行单条查询操作 * @param conn @@ -225,4 +266,28 @@ public class JdbcOperations { }); } + + /** + * 获取自动生成ID字段 + * @param clazz + * @return + */ + private static Field getGenerateIdField(Class clazz) { + if (null == clazz) { + return null; + } + if (generateIdFieldMap.containsKey(clazz)) { + return generateIdFieldMap.get(clazz); + } + Set fields = ReflectUtil.getDeclaredFields(clazz); + if (CollectionUtil.isEmpty(fields)) { + return null; + } + Field field = fields.stream().filter(f -> f.isAnnotationPresent(Id.class)).findFirst().orElse(null); + if (null != field) { + return field; + } + field = fields.stream().filter(f -> f.getName().equals("id")).findFirst().orElse(null); + return field; + } } diff --git a/neutrino-core/src/main/java/fun/asgc/neutrino/core/db/template/JdbcTemplate.java b/neutrino-core/src/main/java/fun/asgc/neutrino/core/db/template/JdbcTemplate.java index beddcf43..5bab3bf9 100644 --- a/neutrino-core/src/main/java/fun/asgc/neutrino/core/db/template/JdbcTemplate.java +++ b/neutrino-core/src/main/java/fun/asgc/neutrino/core/db/template/JdbcTemplate.java @@ -21,16 +21,12 @@ */ package fun.asgc.neutrino.core.db.template; -import fun.asgc.neutrino.core.util.ArrayUtil; - import javax.sql.DataSource; import java.sql.Connection; import java.sql.SQLException; -import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.stream.Collectors; /** * @@ -74,6 +70,31 @@ public class JdbcTemplate { return res; } + /** + * TODO 临时用来兼容insert之后需要返设主键的问题 + * @param sql + * @param params + * @return + * @throws SQLException + */ + public int updateByModel(String sql, Object model, Object ...params) throws SQLException { + int res = -1; + Connection conn = null; + + try { + conn = dataSourceHolder.getConnection(); + res = jdbcOperations.executeUpdateByModel(conn,sql, model, params); + } finally { + try { + dataSourceHolder.tryClose(conn); + } catch (SQLException e) { + e.printStackTrace(); + } + } + + return res; + } + public int update(SqlAndParams sqlAndParams) throws SQLException { return update(sqlAndParams.getSql(), sqlAndParams.getParamArray()); } @@ -83,7 +104,8 @@ public class JdbcTemplate { } public int updateByModel(String sql, Object model) throws SQLException { - return update(new SqlAndParams(sql, model)); + SqlAndParams sqlAndParams = new SqlAndParams(sql, model); + return updateByModel(sqlAndParams.getSql(), model, sqlAndParams.getParamArray()); } public T query(Class clazz, String sql, Object ...params) throws SQLException { 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 5316e227..b853610f 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 @@ -27,7 +27,6 @@ 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.ChannelUtil; -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.base.proxy.ProxyConfig; @@ -38,27 +37,14 @@ import fun.asgc.neutrino.proxy.server.constant.SuccessCodeEnum; import fun.asgc.neutrino.proxy.server.dal.LicenseMapper; import fun.asgc.neutrino.proxy.server.dal.entity.ClientConnectRecordDO; 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.*; import fun.asgc.neutrino.proxy.server.util.ProxyUtil; -import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInitializer; -import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.channel.socket.SocketChannel; -import io.netty.channel.socket.nio.NioServerSocketChannel; import lombok.extern.slf4j.Slf4j; -import java.net.BindException; import java.util.Date; -import java.util.List; -import java.util.stream.Collectors; /** * @@ -70,10 +56,6 @@ import java.util.stream.Collectors; @Match(type = Constants.ProxyDataTypeName.AUTH) @Component public class ProxyMessageAuthHandler implements ProxyMessageHandler { - @Autowired("serverBossGroup") - private NioEventLoopGroup serverBossGroup; - @Autowired("serverWorkerGroup") - private NioEventLoopGroup serverWorkerGroup; @Autowired private ProxyConfig proxyConfig; @Autowired @@ -90,6 +72,8 @@ public class ProxyMessageAuthHandler implements ProxyMessageHandler { private ClientConnectRecordService clientConnectRecordService; @Autowired private LicenseMapper licenseMapper; + @Autowired + private VisitorChannelService visitorChannelService; @Override public void handle(ChannelHandlerContext ctx, ProxyMessage proxyMessage) { @@ -173,46 +157,12 @@ public class ProxyMessageAuthHandler implements ProxyMessageHandler { // 更新license在线状态 licenseMapper.updateOnlineStatus(licenseDO.getId(), OnlineStatusEnum.ONLINE.getStatus(), now); - - List portMappingList = portMappingService.findEnableListByLicenseId(licenseDO.getId()); - // 没有端口映射仍然保持连接 - ProxyUtil.initProxyInfo(licenseDO.getId(), ProxyMapping.buildList(portMappingList)); - - ProxyUtil.addCmdChannel(licenseDO.getId(), ctx.channel(), portMappingList.stream().map(PortMappingDO::getServerPort).collect(Collectors.toSet())); - - startUserPortServer(ProxyUtil.getAttachInfo(ctx.channel()), portMappingList); + // 初始化VisitorChannel + visitorChannelService.initVisitorChannel(licenseDO.getId(), ctx.channel()); } @Override public String name() { return ProxyDataTypeEnum.AUTH.getDesc(); } - - private void startUserPortServer(CmdChannelAttachInfo cmdChannelAttachInfo, List portMappingList) { - if (CollectionUtil.isEmpty(portMappingList)) { - return; - } - ServerBootstrap bootstrap = new ServerBootstrap(); - bootstrap.group(serverBossGroup, serverWorkerGroup) - .channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer() { - @Override - public void initChannel(SocketChannel ch) throws Exception { - ch.pipeline().addFirst(new BytesMetricsHandler()); - ch.pipeline().addLast(new VisitorChannelHandler()); - } - }); - - for (PortMappingDO portMapping : portMappingList) { - try { - proxyMutualService.bindServerPort(cmdChannelAttachInfo, portMapping.getServerPort()); - bootstrap.bind(portMapping.getServerPort()).get(); - log.info("绑定用户端口： {}", portMapping.getServerPort()); - } catch (Exception ex) { - // BindException表示该端口已经绑定过 - if (!(ex.getCause() instanceof BindException)) { - throw new RuntimeException(ex); - } - } - } - } } diff --git a/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/VisitorChannelService.java b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/VisitorChannelService.java new file mode 100644 index 00000000..ba7b1a89 --- /dev/null +++ b/neutrino-proxy-server/src/main/java/fun/asgc/neutrino/proxy/server/service/VisitorChannelService.java @@ -0,0 +1,114 @@ +/** + * 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.Autowired; +import fun.asgc.neutrino.core.annotation.Component; +import fun.asgc.neutrino.core.annotation.NonIntercept; +import fun.asgc.neutrino.core.util.CollectionUtil; +import fun.asgc.neutrino.proxy.server.dal.entity.PortMappingDO; +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.util.ProxyUtil; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.SocketChannel; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import lombok.extern.slf4j.Slf4j; + +import java.net.BindException; +import java.util.List; +import java.util.stream.Collectors; + +/** + * 访问者通道服务 + * @author: aoshiguchen + * @date: 2023/2/5 + */ +@Slf4j +@NonIntercept +@Component +public class VisitorChannelService { + @Autowired("serverBossGroup") + private NioEventLoopGroup serverBossGroup; + @Autowired("serverWorkerGroup") + private NioEventLoopGroup serverWorkerGroup; + @Autowired + private PortMappingService portMappingService; + @Autowired + private ProxyMutualService proxyMutualService; + + /** + * 初始化 + * @param licenseId + */ + public void initVisitorChannel(Integer licenseId, Channel cmdChannel) { + List portMappingList = portMappingService.findEnableListByLicenseId(licenseId); + // 没有端口映射仍然保持连接 + ProxyUtil.initProxyInfo(licenseId, ProxyMapping.buildList(portMappingList)); + + ProxyUtil.addCmdChannel(licenseId, cmdChannel, portMappingList.stream().map(PortMappingDO::getServerPort).collect(Collectors.toSet())); + + startUserPortServer(ProxyUtil.getAttachInfo(cmdChannel), portMappingList); + } + + /** + * 更新 + * 触发时机：新增端口映射、修改端口映射、删除端口映射、禁用端口映射、启用端口映射、禁用license、启用license、禁用用户、启用用户 + * @param licenseId + */ + public void UpdateVisitorChannel(Integer licenseId) { + + } + + private void startUserPortServer(CmdChannelAttachInfo cmdChannelAttachInfo, List portMappingList) { + if (CollectionUtil.isEmpty(portMappingList)) { + return; + } + ServerBootstrap bootstrap = new ServerBootstrap(); + bootstrap.group(serverBossGroup, serverWorkerGroup) + .channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer() { + @Override + public void initChannel(SocketChannel ch) throws Exception { + ch.pipeline().addFirst(new BytesMetricsHandler()); + ch.pipeline().addLast(new VisitorChannelHandler()); + } + }); + + for (PortMappingDO portMapping : portMappingList) { + try { + proxyMutualService.bindServerPort(cmdChannelAttachInfo, portMapping.getServerPort()); + bootstrap.bind(portMapping.getServerPort()).get(); + log.info("绑定用户端口： {}", portMapping.getServerPort()); + } catch (Exception ex) { + // BindException表示该端口已经绑定过 + if (!(ex.getCause() instanceof BindException)) { + throw new RuntimeException(ex); + } + } + } + } +}