java - Netty:发送消息时出现问题

Question

所以我最近接触了一点 netty (5.0.0.Alpha2)，我非常喜欢它!但不幸的是我无法发送/接收消息。奇怪的是，连接和断开连接就像一种魅力。服务器接收客户端已连接/断开连接以及添加/删除 channel 的消息。只是消息无法正常工作。

我确实以很多其他方式尝试过(例如没有编码器)，但它从来没有工作过..也许这里有人有任何想法？我真的很感激!

提前致谢!您可以在下面找到所有使用的源代码:

CoreClient.java

package me.creepsterlgc.coreclient;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class CoreClient extends Thread {

    private String host;
    private int port;

    private Channel channel;

    public CoreClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void run() {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap()
            .group(group)
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {

                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
                        pipeline.addLast("encoder", new StringEncoder());
                        pipeline.addLast("decoder", new StringDecoder());
                        pipeline.addLast("handler", new CoreClientHandler());

                    }

            });

            ChannelFuture f = bootstrap.connect(host, port);
            channel = f.sync().channel();
            ChannelFuture cf = null;
            try {
                cf = channel.writeAndFlush("Testing..").sync();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            if (!cf.isSuccess()) {
                System.out.println("Send failed: " + cf.cause());
            }

        }
        catch (InterruptedException e) {
            e.printStackTrace();    
        }
        finally {

        }

    }

    public void send(String message) {
        ChannelFuture cf = null;
        try {
            cf = channel.writeAndFlush(message).sync();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        channel.flush();
        if (!cf.isSuccess()) {
            System.out.println("Send failed: " + cf.cause());
        }
    }

    public void shutdown() {

    }

}

CoreClientHandler.java

package me.creepsterlgc.coreclient;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

public class CoreClientHandler extends ChannelHandlerAdapter {

    ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    public void channelRead(ChannelHandlerContext context, Object message) throws Exception {
        context.write(message);
        Channel channel = context.channel();
        Log.message(channel.remoteAddress().toString(), message.toString());
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext context) {
        context.flush();
    }

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

}

CoreServer.java

package me.creepsterlgc.coreserver;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class CoreServer extends Thread {

    private int port;

    public CoreServer(int port) {
        this.port = port;
    }

    public void run() {

        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap()
            .group(boss, worker)
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {

                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
                        pipeline.addLast("encoder", new StringEncoder());
                        pipeline.addLast("decoder", new StringDecoder());
                        pipeline.addLast("handler", new CoreServerHandler());

                    }

            });

            ChannelFuture f = bootstrap.bind(port).sync();
            f.channel().closeFuture().sync();

        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }
        finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }

    }

}

CoreServerHandler.java

package me.creepsterlgc.coreserver;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.GlobalEventExecutor;

public class CoreServerHandler extends ChannelHandlerAdapter {

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

    @Override
    public void handlerAdded(ChannelHandlerContext context) {
        Channel channel = context.channel();
        channels.add(channel);
        Log.connect(channel.remoteAddress().toString());
        System.out.println("There are currently " + channels.size() + " clients connected.");
        ChannelFuture cf = null;
        cf = channel.write("Successfully connected to: master");
        channel.flush();
        if (!cf.isSuccess()) {
            System.out.println("Send failed: " + cf.cause());
        }
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext context) {
        Channel channel = context.channel();
        channels.remove(channel);
        Log.disconnect(channel.remoteAddress().toString());
    }

    @Override
    public void channelRead(ChannelHandlerContext context, Object message) throws Exception {
        context.write(message);
        System.out.println("Received: ");
        ByteBuf in = (ByteBuf) message;
        try {
            while (in.isReadable()) {
                System.out.print((char) in.readByte());
                System.out.flush();
            }
        } finally {
            ReferenceCountUtil.release(message);
        }
        Channel channel = context.channel();
        Log.message(channel.remoteAddress().toString(), message.toString());
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext context) {
        context.flush();
    }

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

    public static void read() {
        for(Channel channel : channels) channel.read();
    }

}

Answer 1

你应该更换

 try {
    cf = channel.write("Testing..").sync();
 } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
 }
 channel.flush();

与:

 try {
    cf = channel.writeAndFlush("Testing..").sync();
 } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
 }