java Netty和BoneCP/Basic Socket Server的更多用户
免责声明——我不是Java程序员。我很可能需要根据任何建议做家庭作业，但我很乐意这么做：）
这就是说，我编写了一个完整的数据库支持的socket服务器，对于我的小测试来说，它运行得很好，现在我正在为初始版本做准备。因为我对Java/Netty/BoneCP不太了解，所以我不知道我是否在某个地方犯了一个根本性的错误，在服务器出问题之前就已经伤害了它
例如，我不知道执行者小组到底做了什么，我应该使用什么数字。将BoneCP作为一个单例实现是否合适，是否真的有必要为每个数据库查询提供所有这些try/catch？等等
我试图将我的整个服务器简化为一个基本的示例，它的运行方式与真实的服务器相同（我在文本中剥离了所有内容，没有用java本身进行测试，因此请原谅由此产生的任何语法错误）
其基本思想是，客户端可以连接，与服务器交换消息，断开其他客户端的连接，并无限期地保持连接，直到他们选择或被迫断开连接。（客户端将每分钟发送ping消息，以保持连接处于活动状态）
除了不测试这个示例之外，唯一的主要区别是clientID的设置方式（安全地假设每个连接的客户机的clientID确实是唯一的），以及在检查值等方面有更多的业务逻辑
底线-可以做些什么来改善这一点，以便它能够处理尽可能多的并发用户
//MAIN
public class MainServer {
public static void main(String[] args) {
EdgeController edgeController = new EdgeController();
edgeController.connect();
}
}
//EdgeController
public class EdgeController {
public void connect() throws Exception {
ServerBootstrap b = new ServerBootstrap();
ChannelFuture f;
try {
b.group(new NioEventLoopGroup(), new NioEventLoopGroup())
.channel(NioServerSocketChannel.class)
.localAddress(9100)
.childOption(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new EdgeInitializer(new DefaultEventExecutorGroup(10)));
// Start the server.
f = b.bind().sync();
// Wait until the server socket is closed.
f.channel().closeFuture().sync();
} finally { //Not quite sure how to get here yet... but no matter
// Shut down all event loops to terminate all threads.
b.shutdown();
}
}
}
//EdgeInitializer
public class EdgeInitializer extends ChannelInitializer<SocketChannel> {
private EventExecutorGroup executorGroup;
public EdgeInitializer(EventExecutorGroup _executorGroup) {
executorGroup = _executorGroup;
}
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("idleStateHandler", new IdleStateHandler(200,0,0));
pipeline.addLast("idleStateEventHandler", new EdgeIdleHandler());
pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.nulDelimiter()));
pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
pipeline.addLast(this.executorGroup, "handler", new EdgeHandler());
}
}
//EdgeIdleHandler
public class EdgeIdleHandler extends ChannelHandlerAdapter {
private static final Logger logger = Logger.getLogger( EdgeIdleHandler.class.getName());
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception{
if(evt instanceof IdleStateEvent) {
ctx.close();
}
}
private void trace(String msg) {
logger.log(Level.INFO, msg);
}
}
//DBController
public enum DBController {
INSTANCE;
private BoneCP connectionPool = null;
private BoneCPConfig connectionPoolConfig = null;
public boolean setupPool() {
boolean ret = true;
try {
Class.forName("com.mysql.jdbc.Driver");
connectionPoolConfig = new BoneCPConfig();
connectionPoolConfig.setJdbcUrl("jdbc:mysql://" + DB_HOST + ":" + DB_PORT + "/" + DB_NAME);
connectionPoolConfig.setUsername(DB_USER);
connectionPoolConfig.setPassword(DB_PASS);
try {
connectionPool = new BoneCP(connectionPoolConfig);
} catch(SQLException ex) {
ret = false;
}
} catch(ClassNotFoundException ex) {
ret = false;
}
return(ret);
}
public Connection getConnection() {
Connection ret;
try {
ret = connectionPool.getConnection();
} catch(SQLException ex) {
ret = null;
}
return(ret);
}
}
//EdgeHandler
public class EdgeHandler extends ChannelInboundMessageHandlerAdapter<String> {
private final Charset CHARSET_UTF8 = Charset.forName("UTF-8");
private long clientID;
static final ChannelGroup channels = new DefaultChannelGroup();
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Connection dbConnection = null;
Statement statement = null;
ResultSet resultSet = null;
String query;
Boolean okToPlay = false;
//Check if status for ID #1 is true
try {
query = "SELECT `Status` FROM `ServerTable` WHERE `ID` = 1";
dbConnection = DBController.INSTANCE.getConnection();
statement = dbConnection.createStatement();
resultSet = statement.executeQuery(query);
if (resultSet.first()) {
if (resultSet.getInt("Status") > 0) {
okToPlay = true;
}
}
} catch (SQLException ex) {
okToPlay = false;
} finally {
if (resultSet != null) {
try {
resultSet.close();
} catch (SQLException logOrIgnore) {
}
}
if (statement != null) {
try {
statement.close();
} catch (SQLException logOrIgnore) {
}
}
if (dbConnection != null) {
try {
dbConnection.close();
} catch (SQLException logOrIgnore) {
}
}
}
if (okToPlay) {
//clientID = setClientID();
sendCommand(ctx, "HELLO", "WORLD");
} else {
sendErrorAndClose(ctx, "CLOSED");
}
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
channels.remove(ctx.channel());
}
@Override
public void messageReceived(ChannelHandlerContext ctx, String request) throws Exception {
// Generate and write a response.
String[] segments_whitespace;
String command, command_args;
if (request.length() > 0) {
segments_whitespace = request.split("\\s+");
if (segments_whitespace.length > 1) {
command = segments_whitespace[0];
command_args = segments_whitespace[1];
if (command.length() > 0 && command_args.length() > 0) {
switch (command) {
case "HOWDY": processHowdy(ctx, command_args); break;
default: break;
}
}
}
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
TraceUtils.severe("Unexpected exception from downstream - " + cause.toString());
ctx.close();
}
/* */
/* STATES - / CLIENT SETUP */
/* */
private void processHowdy(ChannelHandlerContext ctx, String howdyTo) {
Connection dbConnection = null;
Statement statement = null;
ResultSet resultSet = null;
String replyBack = null;
try {
dbConnection = DBController.INSTANCE.getConnection();
statement = dbConnection.createStatement();
resultSet = statement.executeQuery("SELECT `to` FROM `ServerTable` WHERE `To`='" + howdyTo + "'");
if (resultSet.first()) {
replyBack = "you!";
}
} catch (SQLException ex) {
} finally {
if (resultSet != null) {
try {
resultSet.close();
} catch (SQLException logOrIgnore) {
}
}
if (statement != null) {
try {
statement.close();
} catch (SQLException logOrIgnore) {
}
}
if (dbConnection != null) {
try {
dbConnection.close();
} catch (SQLException logOrIgnore) {
}
}
}
if (replyBack != null) {
sendCommand(ctx, "HOWDY", replyBack);
} else {
sendErrorAndClose(ctx, "ERROR");
}
}
private boolean closePeer(ChannelHandlerContext ctx, long peerClientID) {
boolean success = false;
ChannelFuture future;
for (Channel c : channels) {
if (c != ctx.channel()) {
if (c.pipeline().get(EdgeHandler.class).receiveClose(c, peerClientID)) {
success = true;
break;
}
}
}
return (success);
}
public boolean receiveClose(Channel thisChannel, long remoteClientID) {
ChannelFuture future;
boolean didclose = false;
long thisClientID = (clientID == null ? 0 : clientID);
if (remoteClientID == thisClientID) {
future = thisChannel.write("CLOSED BY PEER" + '\n');
future.addListener(ChannelFutureListener.CLOSE);
didclose = true;
}
return (didclose);
}
private ChannelFuture sendCommand(ChannelHandlerContext ctx, String cmd, String outgoingCommandArgs) {
return (ctx.write(cmd + " " + outgoingCommandArgs + '\n'));
}
private ChannelFuture sendErrorAndClose(ChannelHandlerContext ctx, String error_args) {
ChannelFuture future = sendCommand(ctx, "ERROR", error_args);
future.addListener(ChannelFutureListener.CLOSE);
return (future);
}
}
# 1 楼答案
当网络消息到达服务器时，它将被解码并释放messageReceived事件
如果你看一下你的管道，最后添加到管道中的是executor。因此，执行器将接收已解码的内容，并释放messageReceived事件
执行者是事件的处理器，服务器将通过它们来判断发生了哪些事件。因此，如何使用遗嘱执行人是一个重要的课题。如果只有一个执行器，并且由于这个原因，所有客户端都使用同一个执行器，那么将有一个队列用于使用同一个执行器
当有许多执行者时，事件的处理时间将减少，因为不会有任何空闲执行者等待
在你的代码中
这意味着该服务器引导在其整个生命周期内只使用10个执行器
初始化新通道时，使用相同的执行器组：
因此，每个新的客户端通道将使用相同的执行器组（10个执行器线程）
如果10个线程能够正确处理传入的事件，那么这就足够了。但是，如果我们能看到消息正在被解码/编码，但不能快速地作为事件处理，那就意味着需要增加它们的数量
我们可以将遗嘱执行人的数量从10人增加到100人，如下所示：
因此，如果有足够的CPU功率，那么处理事件队列的速度会更快
不应该做的是为每个新通道创建新的执行器：
上面这一行是为每个新通道创建一个新的执行器组，这将大大降低速度，例如，如果有3000个客户端，将有3000个执行器组（线程）。这就消除了NIO的主要优势，即能够以较低的线程量使用
我们可以在启动时创建3000个执行器，而不是为每个通道创建一个执行器，至少不会在每次客户端连接/断开连接时删除和创建它们
上面这一行比为每个客户机创建一个执行器更容易接受，因为所有客户机都连接到同一个执行器组，并且当一个客户机断开连接时，即使删除了客户机数据，执行器仍然存在
如果我们必须谈论数据库请求，一些数据库查询可能需要很长时间才能完成，因此，如果有10个执行者，并且有10个作业正在处理，那么第11个作业将不得不等待其他作业中的一个完成。如果服务器同时接收10个以上非常耗时的数据库作业，这将是一个瓶颈。增加执行者的数量将在一定程度上解决瓶颈问题