java - Netty 4 解码器问题

Question

我正在开发一个基于 smpp 的应用程序。然而，我在解码 smpp PDU 时注意到了一些事情。当应用程序一个接一个地接收到几个 pdu 时，解码不会引起任何问题。然而，当它异步接收大量 pdu 时，我得到了一些空指针异常。这就是为什么我在 ntty 3.X 中没有问题，但在 netty 中。我怀疑解码速度比 pdu 的速率慢或者我遗漏了一些东西。

异常(exception):

.............
c.c.smpp.channel.SmppPduReader - 
io.netty.handler.codec.DecoderException: java.lang.NullPointerException: element
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:263) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:131) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:337) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:323) [netty-all-4.0.10.Final.jar:na]
    at com.cloudhopper.smpp.channel.SmppChannelLogger.channelRead(SmppChannelLogger.java:63) ~[bin/:na]
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:337) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:323) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:337) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:323) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:785) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:100) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:480) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:447) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:341) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:101) ~[netty-all-4.0.10.Final.jar:na]
    at java.lang.Thread.run(Thread.java:724) ~[na:1.7.0_25]
Caused by: java.lang.NullPointerException: element
    at io.netty.util.internal.RecyclableArrayList.add(RecyclableArrayList.java:104) ~[netty-all-4.0.10.Final.jar:na]
    at com.cloudhopper.smpp.channel.SmppPduDecoder.decode(SmppPduDecoder.java:42) ~[bin/:na]
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:232) ~[netty-all-4.0.10.Final.jar:na]
    ... 16 common frames omitted
c.c.s.i.GenericSmppSessionHandler - Default handling is to discard an unknown throwable:
io.netty.handler.codec.DecoderException: java.lang.NullPointerException: element
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:263) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:131) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:337) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:323) [netty-all-4.0.10.Final.jar:na]
    at com.cloudhopper.smpp.channel.SmppChannelLogger.channelRead(SmppChannelLogger.java:63) ~[bin/:na]
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:337) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:323) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:337) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:323) [netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:785) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:100) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:480) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:447) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:341) ~[netty-all-4.0.10.Final.jar:na]
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:101) ~[netty-all-4.0.10.Final.jar:na]
    at java.lang.Thread.run(Thread.java:724) ~[na:1.7.0_25]
Caused by: java.lang.NullPointerException: element
    at io.netty.util.internal.RecyclableArrayList.add(RecyclableArrayList.java:104) ~[netty-all-4.0.10.Final.jar:na]
    at com.cloudhopper.smpp.channel.SmppPduDecoder.decode(SmppPduDecoder.java:42) ~[bin/:na]
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:232) ~[netty-all-4.0.10.Final.jar:na]
    ... 16 common frames omitted 

公共(public)类 SmppPduDecoder 扩展了 ByteToMessageDecoder {

private final SmppPduTranscoder transcoder;
@SuppressWarnings("unused")
private static final Logger logger = LoggerFactory.getLogger(SmppPduDecoder.class);

/**
 * 
 */
public SmppPduDecoder(SmppPduTranscoder transcoder) {
    this.transcoder = transcoder;
}

/* (non-Javadoc)
 * @see io.netty.handler.codec.ByteToMessageDecoder#decode(io.netty.channel.ChannelHandlerContext, io.netty.buffer.ByteBuf, java.util.List)
 */
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in,
        List<Object> out) throws Exception {
    Pdu pdu = this.transcoder.decode(in);
    out.add(pdu);
}

}

这是进行解码的转码器代码片段。

@Override
public Pdu decode(ByteBuf buffer) throws UnrecoverablePduException, RecoverablePduException {
    // wait until the length prefix is available
    if (buffer.readableBytes() < SmppConstants.PDU_INT_LENGTH) {
        return null;
    }

    // parse the command length (first 4 bytes)
    int commandLength = buffer.getInt(buffer.readerIndex());
    //logger.trace("PDU commandLength [" + commandLength + "]");

    // valid command length is >= 16 bytes
    if (commandLength < SmppConstants.PDU_HEADER_LENGTH) {
        throw new UnrecoverablePduException("Invalid PDU length [0x" + HexUtil.toHexString(commandLength) + "] parsed");
    }

    // wait until the whole pdu is available (entire pdu)
    if (buffer.readableBytes() < commandLength) {

        return null;
    }

    // at this point, we have the entire PDU and length already in the buffer
    // we'll create a new "view" of this PDU and read the data from the actual buffer
    // NOTE: this should be super fast since the underlying byte array doesn't get copied
    ByteBuf buffer0 = buffer.readSlice(commandLength);

    return doDecode(commandLength, buffer0);
}

protected Pdu doDecode(int commandLength, ByteBuf buffer) throws UnrecoverablePduException, RecoverablePduException {
    // skip the length field because we already parsed it
    buffer.skipBytes(SmppConstants.PDU_INT_LENGTH);

    // read the remaining portion of the PDU header
    int commandId = buffer.readInt();
    int commandStatus = buffer.readInt();
    int sequenceNumber = buffer.readInt();

    // this is a major issue if the sequence number is invalid
    SequenceNumber.assertValid(sequenceNumber);

    Pdu pdu = null;

    // any command id with its 31st bit set to true is a response
    if (PduUtil.isRequestCommandId(commandId)) {
        if (commandId == SmppConstants.CMD_ID_ENQUIRE_LINK) {
            pdu = new EnquireLink();
        } else if (commandId == SmppConstants.CMD_ID_DELIVER_SM) {
            pdu = new DeliverSm();
        } else if (commandId == SmppConstants.CMD_ID_SUBMIT_SM) {
            pdu = new SubmitSm();
        } else if (commandId == SmppConstants.CMD_ID_DATA_SM) {
            pdu = new DataSm();
        } else if (commandId == SmppConstants.CMD_ID_CANCEL_SM) {
            pdu = new CancelSm();
        } else if (commandId == SmppConstants.CMD_ID_QUERY_SM) {
            pdu = new QuerySm();
        } else if (commandId == SmppConstants.CMD_ID_BIND_TRANSCEIVER) {
            pdu = new BindTransceiver();
        } else if (commandId == SmppConstants.CMD_ID_BIND_TRANSMITTER) {
            pdu = new BindTransmitter();
        } else if (commandId == SmppConstants.CMD_ID_BIND_RECEIVER) {
            pdu = new BindReceiver();
        } else if (commandId == SmppConstants.CMD_ID_UNBIND) {
            pdu = new Unbind();
        } else {
            pdu = new PartialPdu(commandId);
        }
    } else {
        if (commandId == SmppConstants.CMD_ID_SUBMIT_SM_RESP) {
            pdu = new SubmitSmResp();
        } else if (commandId == SmppConstants.CMD_ID_DELIVER_SM_RESP) {
            pdu = new DeliverSmResp();
        } else if (commandId == SmppConstants.CMD_ID_DATA_SM_RESP) {
            pdu = new DataSmResp();
        } else if (commandId == SmppConstants.CMD_ID_CANCEL_SM_RESP) {
            pdu = new CancelSmResp();
        } else if (commandId == SmppConstants.CMD_ID_QUERY_SM_RESP) {
            pdu = new QuerySmResp();
        } else if (commandId == SmppConstants.CMD_ID_ENQUIRE_LINK_RESP) {
            pdu = new EnquireLinkResp();
        } else if (commandId == SmppConstants.CMD_ID_BIND_TRANSCEIVER_RESP) {
            pdu = new BindTransceiverResp();
        } else if (commandId == SmppConstants.CMD_ID_BIND_RECEIVER_RESP) {
            pdu = new BindReceiverResp();
        } else if (commandId == SmppConstants.CMD_ID_BIND_TRANSMITTER_RESP) {
            pdu = new BindTransmitterResp();
        } else if (commandId == SmppConstants.CMD_ID_UNBIND_RESP) {
            pdu = new UnbindResp();
        } else if (commandId == SmppConstants.CMD_ID_GENERIC_NACK) {
            pdu = new GenericNack();
        } else {
            pdu = new PartialPduResp(commandId);
        }
    }

    // set pdu header values
    pdu.setCommandLength(commandLength);
    pdu.setCommandStatus(commandStatus);
    pdu.setSequenceNumber(sequenceNumber);

    // check if we need to throw an exception
    if (pdu instanceof PartialPdu) {
        throw new UnknownCommandIdException(pdu, "Unsupported or unknown PDU request commandId [0x" + HexUtil.toHexString(commandId) + "]");
    } else if (pdu instanceof PartialPduResp) {
        throw new UnknownCommandIdException(pdu, "Unsupported or unknown PDU response commandId [0x" + HexUtil.toHexString(commandId) + "]");
    }

    // see if we can map the command status into a message
    if (pdu instanceof PduResponse) {
        PduResponse response = (PduResponse)pdu;
        response.setResultMessage(context.lookupResultMessage(commandStatus));
    }

    try {
        // parse pdu body parameters (may throw exception)
        pdu.readBody(buffer);
        // parse pdu optional parameters (may throw exception)
        pdu.readOptionalParameters(buffer, context);
    } catch (RecoverablePduException e) {
        // check if we should add the partial pdu to the exception
        if (e.getPartialPdu() == null) {
            e.setPartialPdu(pdu);
        }
        // rethrow it
        throw e;
    }

    return pdu;
}

请 Netty 极客帮助我。

Answer 1

看起来您正在将 null 对象添加到结果列表中:

protected 无效解码(ChannelHandlerContext ctx，ByteBuf中， 列出)抛出异常 { Pdu pdu = this.transcoder.decode(in); 输出.add(pdu);}

只需添加检查 pdu 是否不为 null:

if (pdu != null) out.add(pdu);