java - 在Netty中发送和接收ArrayList对象或Array对象

Question

大家好，我是 Netty 的新手。我想要的是这样的东西。在我的客户端上，我想输入一个字符串查询示例SELECT * FROM drivers。我正在使用 mysql xampp 服务器。然后我的服务器将查询它并将其添加到 arraylist

 private List<Drivers> getRecords(ResultSet rs) throws SQLException {

        List<Drivers> records = new ArrayList<Drivers>();
        while(rs.next()){
            records.add(new Drivers(rs.getInt("first_name"), rs.getString("last_name")));
        }
        return records;
    }

之后它将把 ArrayList 对象发送回客户端。

我现在的问题是如何将其填充到客户端

这是我的服务器管道

public class ServerInitializer extends ChannelInitializer<SocketChannel> {

    private final SslContext sslCtx;

    public ServerInitializer(SslContext sslCtx) {
        this.sslCtx = sslCtx;
    }

    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        // Add SSL handler first to encrypt and decrypt everything.
        // In this example, we use a bogus certificate in the server side
        // and accept any invalid certificates in the client side.
        // You will need something more complicated to identify both
        // and server in the real world.
        pipeline.addLast(sslCtx.newHandler(ch.alloc()));

        // On top of the SSL handler, add the text line codec.
        /*
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder());
        pipeline.addLast(new StringEncoder());
          */
        //pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        /*

        */
        //pipeline.addLast("frameDecoder",new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
        //pipeline.addLast("bytesDecoder",new ByteArrayDecoder());
        pipeline.addLast(new ObjectEncoder());
        pipeline.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));

        // and then business logic.
        pipeline.addLast(new ServerHandler());
    }
}

我的客户管道

 public class ClientInitializer extends ChannelInitializer<SocketChannel> {

        private final SslContext sslCtx;

        public ClientInitializer(SslContext sslCtx) {
            this.sslCtx = sslCtx;
        }

        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();

            // Add SSL handler first to encrypt and decrypt everything.
            // In this example, we use a bogus certificate in the server side
            // and accept any invalid certificates in the client side.
            // You will need something more complicated to identify both
            // and server in the real world.
            pipeline.addLast(sslCtx.newHandler(ch.alloc(), Client.HOST, Client.PORT));

            // On top of the SSL handler, add the text line codec.

            /*
            pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
            pipeline.addLast(new StringDecoder());
            pipeline.addLast(new StringEncoder());
              */
            //pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
            /*
            */
            pipeline.addLast("frameDecoder",new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
            pipeline.addLast("bytesDecoder",new ByteArrayDecoder());
            pipeline.addLast(new ObjectEncoder());
            pipeline.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
            // and then business logic.
            pipeline.addLast(new ClientHandler());
        }
    }

我的服务器 channel 处理程序

public class ServerHandler extends ChannelInboundHandlerAdapter {

    static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    public void channelActive(final ChannelHandlerContext ctx) {
        // Once session is secured, send a greeting and register the channel to the global channel
        // list so the channel received the messages from others.
        ctx.pipeline().get(SslHandler.class).handshakeFuture().addListener(
                new GenericFutureListener<Future<Channel>>() {
                    @Override
                    public void operationComplete(Future<Channel> future) throws Exception {

                        channels.add(ctx.channel());
                    }
        });
    }

    private List<Drivers> getRecords(ResultSet rs) throws SQLException {

        List<Drivers> records=new ArrayList<Drivers>();
        while(rs.next()){
            records.add(new Drivers(rs.getInt(0), rs.getString(1), rs.getString(2), rs.getString(3), rs.getString(4), 
                    rs.getString(5), rs.getString(6), rs.getInt(7), rs.getString(8), rs.getString(9), rs.getString(10), 
                    rs.getString(11), rs.getString(12), rs.getString(13), rs.getString(14), 
                    rs.getString(15), rs.getString(16), rs.getString(17), rs.getString(18), rs.getString(19)));
        }
        return records;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
    {
         // Send the received message to all channels but the current one.
        for (Channel c: channels) {
            if (c != ctx.channel()) {
                c.writeAndFlush("[" + ctx.channel().remoteAddress() + "] " + msg + '\n');

            } else {

                try {
                    new DataManipulator();
                    System.out.print(msg.toString());
                    ResultSet rs = DataManipulator.generalQuery(msg.toString());
                    c.writeAndFlush(getRecords(rs));

                } catch (ClassNotFoundException | InstantiationException
                        | IllegalAccessException | SQLException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }

        // Close the connection if the client has sent 'bye'.
        if ("bye".equals(msg.toString().toLowerCase())) {
            ctx.close();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

我的客户端 channel handler 程序

public class ClientHandler extends SimpleChannelInboundHandler<Object[]> {

     static int count = 1;
     @Override
     public void channelActive(final ChannelHandlerContext ctx) {
         System.out.println(ctx.read());
     }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object[] msg)
            throws Exception {
        // Still don't know how to receive the data.

    }
}

请帮助如何在 netty 上完美完成此操作。

Answer 1

您的管道中不需要 ByteArrayDecoder 和 LengthFieldBasedFrameDecoder。您需要的是 ObjectDecoder 和 ObjectEncoder，这应该就是全部(除了最后一个处理程序。)