java - 如何操作来自 Netty 服务器/客户端的消息

Question

我正在对字符串的 Netty 客户端/服务器传输进行原型(prototype)设计，现在我想在这些字符串到达​​服务器端时将其传递到文件。

客户:

    private ClientBootstrap bootstrap;
    private Channel connector;
    private MyHandler handler=new MyHandler();

    public boolean start() {
        // Standard netty bootstrapping stuff.
        Executor bossPool = Executors.newCachedThreadPool();
        Executor workerPool = Executors.newCachedThreadPool();
        ChannelFactory factory =
                new NioClientSocketChannelFactory(bossPool, workerPool);
        this.bootstrap = new ClientBootstrap(factory);

        // Declared outside to fit under 80 char limit
        final DelimiterBasedFrameDecoder frameDecoder =
                new DelimiterBasedFrameDecoder(Integer.MAX_VALUE,
                        Delimiters.lineDelimiter());
        this.bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            public ChannelPipeline getPipeline() throws Exception {
                return Channels.pipeline(
                        handler,
                        frameDecoder,
                        new StringDecoder(),
                        new StringEncoder());
            }
        });

        ChannelFuture future = this.bootstrap
                .connect(new InetSocketAddress("localhost", 12345));
        if (!future.awaitUninterruptibly().isSuccess()) {
            System.out.println("--- CLIENT - Failed to connect to server at " +
                    "localhost:12345.");
            this.bootstrap.releaseExternalResources();
            return false;
        }

        this.connector = future.getChannel();
        return this.connector.isConnected();
    }

    public void stop() {
        if (this.connector != null) {
            this.connector.close().awaitUninterruptibly();
        }
        this.bootstrap.releaseExternalResources();
        System.out.println("--- CLIENT - Stopped.");
    }

    public boolean sendMessage(String message) {
        if (this.connector.isConnected()) {
            // Append \n if it's not present, because of the frame delimiter
            if (!message.endsWith("\n")) {
                this.connector.write(message + '\n');
            } else {
                this.connector.write(message);
            }
            System.out.print(message);
            return true;
        }

        return false;
    }

服务器:

    private final String id;
    private ServerBootstrap bootstrap;
    private ChannelGroup channelGroup;
    private MyHandler handler= new MyHandler();


    public Server(String id) {
        this.id = id;
    }

    // public methods ---------------------------------------------------------

    public boolean start() {
        // Pretty standard Netty startup stuff...
        // boss/worker executors, channel factory, channel group, pipeline, ...
        Executor bossPool = Executors.newCachedThreadPool();
        Executor workerPool = Executors.newCachedThreadPool();
        ChannelFactory factory =
                new NioServerSocketChannelFactory(bossPool, workerPool);
        this.bootstrap = new ServerBootstrap(factory);

        this.channelGroup = new DefaultChannelGroup(this.id + "-all-channels");


        // declared here to fit under the 80 char limit
        final ChannelHandler delimiter =
                new DelimiterBasedFrameDecoder(Integer.MAX_VALUE,
                        Delimiters.lineDelimiter());
        this.bootstrap.setPipelineFactory(new ChannelPipelineFactory() {

            @Override
            public ChannelPipeline getPipeline() throws Exception {
                SimpleChannelHandler handshakeHandler =
                        new SimpleChannelHandler();
                return Channels.pipeline(
                        handler,
                        delimiter,
                        new StringDecoder(),
                        new StringEncoder(),
                        handshakeHandler);
            }
        });

        Channel acceptor = this.bootstrap.bind(new InetSocketAddress(12345));
        if (acceptor.isBound()) {
            System.out.println("+++ SERVER - bound to *:12345");
            this.channelGroup.add(acceptor);
            return true;
        } else {
            System.err.println("+++ SERVER - Failed to bind to *:12345");
            this.bootstrap.releaseExternalResources();
            return false;
        }
    }

    public void stop() {
        this.channelGroup.close().awaitUninterruptibly();
        this.bootstrap.releaseExternalResources();
        System.err.println("+++ SERVER - Stopped.");
    }

使用的处理程序:客户端处理程序:

public class MyHandler extends SimpleChannelUpstreamHandler{
    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
            throws Exception {
        if(e.getMessage() instanceof String){
            System.out.println((String)e.getMessage());
        }
        System.out.println(e.getMessage().toString());
    }
}

服务器处理程序:

@Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
            throws Exception {
        Channel channel= ctx.getChannel();
        channel.write(e.getMessage());
        if(e.getMessage() instanceof String){
            System.out.println((String)e.getMessage());
        }
        System.out.println(e.getMessage().toString());
    }

客户端运行者:

public static void main(String[] args) throws InterruptedException {

        final int nMessages = 5;

        try {
            Client c = new Client();

            if (!c.start()) {
                return;
            }

            for (int i = 0; i < nMessages; i++) {

                Thread.sleep(1L);
                c.sendMessage((i + 1) + "\n");
            }
            c.stop();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

服务器运行器:

public static void main(String[] args) {
        final Server s = new Server("server1");

        if (!s.start()) {
            return;
        }

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                s.stop();
            }
        });
    }

现在我真正需要的是在客户端和服务器端打印我在 channel 上写的消息，我对此感到非常困惑。

Answer 1

乍一看，您的管道创建似乎是错误的。在服务器端解码时，需要首先是Delimiter，然后是StringDecoder，最后是业务处理程序。您可以通过在这些解码器和编码器中放置断点来解决这个问题。另请看看这个link有关其工作原理的非常好的文档。