无法使用udp套接字在netty中回复

Question

以前的代码是用c语言编写的，使用的是这样的代码：

sockHndl = socket(AF_INET,SOCK_DGRAM,0);
sockConf.sin_family = AF_INET;
sockConf.sin_addr.s_addr = INADDR_ANY;
sockConf.sin_port = htons(33333);

bind(sockHndl, (struct sockaddr*)&sockConf, sizeof(sockConf));

// in a thread
recvfrom(sockHndl, buffer, 1515, 0, (struct sockaddr*)&sockConf, &s);

// process any recevied data
....

// reply
sendto(sockHndl, buffer, strlen(buffer), 0, (struct sockaddr*)&sockConf, sizeof(sockConf));

这段代码可以工作，但我必须使用Netty在java中重新实现它。

在tcp中使用netty不是问题，但是用netty回复客户机对我来说是个问题，如果我想在通道上写，它失败了，出现了java.nio.channels.notyetConnectedException。

也许这与netty上的这个问题有关，但我不知道如何在代码中再现c行为。

final Bootstrap b = new Bootstrap();

b.group(comIOEventLoop).channel(NioDatagramChannel.class).handler(new ChannelInitializer<NioDatagramChannel>()
{
    protected void initChannel(NioDatagramChannel ch) throws Exception
    {
        final ChannelPipeline p = ch.pipeline();
        p.addLast("SITCDecoder", new SITCDecoder());
        p.addLast("SITCEncoder", new SITCEncoder());
        p.addLast("SITCHandler", new SimpleChannelInboundHandler<ITCUdpRequest>()
        {
            protected void channelRead0(ChannelHandlerContext ctx, ITCUdpRequest msg) throws Exception
            {
                // here ctx.channel().remoteAddress() == null !
                switch(msg.getType())
                {
                    case GET_PARAMS_0:
                        send(ctx.channel(), new ITCUdpResponse(-80, -20));
                        break;
                }
            }

            public void send(final Channel channel, final ITCUdpResponse data)
            {
                if(data == null)
                {
                    LOGGER.error("data == null !!!");
                }

                channel.writeAndFlush(data).addListener(new GenericFutureListener<Future<Void>>()
                {
                    public void operationComplete(final Future<Void> future) throws Exception
                    {
                        if(future.isDone() && future.isSuccess())
                        {
                            LOGGER.debug("OK");
                        }
                        else
                        {
                            LOGGER.error("error " + future.isDone() + " - " + future.isSuccess());
                            if(!future.isSuccess())
                            {
                                future.cause().printStackTrace();
                            }
                        }
                    }
                });
            }
        });
    }
});

channel = b.bind(port).sync().channel();
channel.closeFuture().await();

如何回复客户？

多谢了。

    public class SITCDecoder extends SimpleChannelInboundHandler<DatagramPacket>
{
    private static final Logger LOGGER              = LoggerFactory.getLogger(SITCDecoder.class);

    private static final String MSG_DD_HEADER       = "SITCDDVAL:";
    private static final int    MSG_DD_HEADER_SIZE  = MSG_DD_HEADER.length();

    protected void channelRead0(final ChannelHandlerContext ctx, final DatagramPacket msg) throws Exception
    {
        final ByteBuf data = (ByteBuf) msg.content();

        if(data != null)
        {
            // we must be able to read at last <code>MSG_HEADER_SIZE</code> Bytes
            if(data.readableBytes() < MSG_DD_HEADER_SIZE)
            {
                LOGGER.error("Not enought data");
                return;
            }

            if(!data.readBytes(MSG_DD_HEADER_SIZE).toString(StandardCharsets.ISO_8859_1).equals(MSG_DD_HEADER))
            {
                LOGGER.error("Header not found");
            }

            final String payload = data.readBytes(data.readableBytes()).toString(StandardCharsets.ISO_8859_1);

            final RecyclableArrayList out = RecyclableArrayList.newInstance();
            out.add(new ITCUdpRequest(payload));

            final int size = out.size();

            for(int i = 0; i < size; i++)
            {
                ctx.fireChannelRead(out.get(i));
            }
            out.recycle();
        }
        else
        {
            ctx.fireChannelRead(msg);
        }
    }
}

编码器代码：

public class SITCEncoder extends MessageToByteEncoder<ITCUdpResponse>
{
    private static final String MSG_RP_HEADER   = "SITCRPVAL:";
    private static final char   MSG_RP_FOOTER   = '*';
    private static final char   VAL_PREFIX      = '[';
    private static final char   VAL_SUFFIX      = ']';

    protected void encode(final ChannelHandlerContext ctx, final ITCUdpResponse msg, final ByteBuf out) throws Exception
    {
        final StringBuilder str = new StringBuilder();
        str.append(MSG_RP_HEADER);

        for(final Object val : msg.getParams())
        {
            str.append(VAL_PREFIX).append(val).append(VAL_SUFFIX);
        }

        str.append(MSG_RP_FOOTER);

        out.ensureWritable(str.length());

        out.writeBytes(str.toString().getBytes());
    }
}

Answer 1

好的，在阅读了stackoverflow针对我的问题给出的建议之后，我已经从NIO*切换到OIO*（就像这里说的那样），并且一切都按照预期工作，没有做任何改变。

桑斯克。

PS：为什么Oio在Nio不工作的地方工作？