如何在netty中执行长任务？

Question

我是netty.io的新手。我试图编写一个应用程序，在客户端第一次连接时插入超过200行。问题是一次只插入10行，看起来像netty只是打断了我的代码。我无法弄清楚如何为我的任务设置最大执行时间。

这是怎么了，我开始服务器：

public void start() throws InterruptedException{ 
     EventLoopGroup group = new NioEventLoopGroup(); 
     EventExecutorGroup sqlExecutorGroup = new DefaultEventExecutorGroup(10); 
     ServerBootstrap server = new ServerBootstrap(); 
     server.group(group); 
     server.channel(NioServerSocketChannel.class); 
     server.localAddress(new InetSocketAddress(port)); 
     server.childHandler(new ChannelInitializer<SocketChannel>(){ 
      @Override 
      protected void initChannel(SocketChannel ch) throws Exception { 
        ch.pipeline().addLast(new HttpServerCodec()); 
        ch.pipeline().addLast(new ChunkedWriteHandler()); 
        ch.pipeline().addLast(new HttpObjectAggregator(65536)); 
        ch.pipeline().addLast(new WebSocketServerProtocolHandler("/")); 
        ch.pipeline().addLast(new WebSocketFrameToByteBuf()); 
        ch.pipeline().addLast(new ProtobufEncoder()); 
        ch.pipeline().addLast(new ProtobufDecoder(Chat.MsgWrapper.getDefaultInstance())); 
        ch.pipeline().addLast(new MsgUserLogin(prop.getProperty("sessions_prefix"))); 
      } 
     }); 
     ChannelFuture f = server.bind().sync(); 
     f.channel().closeFuture().sync(); 
    } 

这是IM执行任务的处理程序中的功能：

private void register(ChannelHandlerContext ctx, Chat.User u){ 
     ctx.executor().execute(new Runnable() { 
      @Override 
      public void run() { 
       ctx.channel().attr(Main.KeyUser).set(u); 
       ConversationUserMember member = ConversationPool.getOrCacheConversationUserMember(u); 
       ConversationPool.UpdateConversations(member); 
       Main.loggedUsers.add(ctx.channel()); 
       member.newChannel(ctx.channel()); 
       returnConversations(member,ctx); //this function is the one getting cut by netty. 
      // fireNewUser(u); 
      } 
     }); 
    } 

Answer 1

多一点研究，我发现所有的后我正在做的错误：

1. 我没有添加EventExecutorGroup到我的处理程序。
2. 执行插入的代码没有正确释放数据库连接，这是主要问题。
3. 在我使用的处理程序ctx.executor（）。执行执行我的代码后，将EventExecutorGroup添加到处理程序，这不是必需的。

工作的代码如下所示：

public void start() throws InterruptedException{ 
     EventLoopGroup group = new NioEventLoopGroup(); 
     EventExecutorGroup sqlExecutorGroup = new DefaultEventExecutorGroup(10); 
     ServerBootstrap server = new ServerBootstrap(); 
     server.group(group); 
     server.channel(NioServerSocketChannel.class); 
     server.localAddress(new InetSocketAddress(port)); 
     server.childHandler(new ChannelInitializer<SocketChannel>(){ 
      @Override 
      protected void initChannel(SocketChannel ch) throws Exception { 
        ch.pipeline().addLast(new HttpServerCodec()); 
        ch.pipeline().addLast(new ChunkedWriteHandler()); 
        ch.pipeline().addLast(new HttpObjectAggregator(65536)); 
        ch.pipeline().addLast(new WebSocketServerProtocolHandler("/")); 
        ch.pipeline().addLast(new WebSocketFrameToByteBuf()); 
        ch.pipeline().addLast(new ProtobufEncoder()); 
        ch.pipeline().addLast(new ProtobufDecoder(Chat.MsgWrapper.getDefaultInstance())); 
        ch.pipeline().addLast(sqlExecutorGroup,new MsgUserLogin(prop.getProperty("sessions_prefix"))); 
      } 
     }); 
     ChannelFuture f = server.bind().sync(); 
     f.channel().closeFuture().sync(); 
    } 

和：

private void register(ChannelHandlerContext ctx, Chat.User u){ 
       ctx.channel().attr(Main.KeyUser).set(u); 
       ConversationUserMember member =       ConversationPool.getOrCacheConversationUserMember(u); 
       ConversationPool.UpdateConversations(member); 
       Main.loggedUsers.add(ctx.channel()); 
       member.newChannel(ctx.channel()); 
       returnConversations(member,ctx); //this function is the one getting cut by netty. 
      // fireNewUser(u); 
    }