NIO框架-Netty（13）
Handler如何使用在前面的例子中已经有了示范，那么同样是扩展自ChannelHandler的Encoder和Decoder，与Handler混合后又是如何使用的？本文将通过一个实际的小例子来展示它们的用法。
该例子模拟一个Server和Client，两者之间通过http协议进行通讯，在Server内部通过一个自定义的StringDecoder把httprequest转换成String。Server端处理完成后，通过StringEncoder把String转换成httpresponse，发送给客户端。具体的处理流程如图所示：
其中红色框中的Decoder、Encoder及request都是Netty框架自带的，灰色框中的三个类是我自己实现的。
Server端的类有：Server&StringDecoder&BusinessHandler&StringEncoder四个类。
1、Server 启动netty服务，并注册handler、coder，注意注册的顺序：
2、StringDecoder 把httpRequest转换成String，其中ByteBufToBytes是一个工具类，负责对ByteBuf中的数据进行读取
3、BusinessHandler 具体处理业务的类，把客户端的请求打印出来，并向客户端发送信息
4、StringEncoder 把字符串转换成HttpResponse
Client端有两个类：Client&ClientInitHandler
1、Client 与Server端建立连接，并向Server端发送HttpRequest请求。
2、ClientInitHandler 从Server端读取响应信息
工具类：ByteBufToBytes 对ByteBuf的数据进行读取，支持流式读取（reading 和&readFull方法结合使用）
ByteBufToBytes&
运行结果：
23:50:48 StringDecoder : msg's type is class io.netty.handler.codec.http.DefaultHttpRequest
23:50:48 StringDecoder : msg's type is class io.netty.handler.codec.http.DefaultLastHttpContent
23:50:48 StringDecoder : change httpcontent to string&
23:50:48 BusinessHandler read msg from client :client said : Are you ok?
23:50:48 StringEncoder response to client.
可以看到执行顺序为：StringDecoder&BusinessHandler&StringEncoder ，其它的都是Netty自身的，没有打印。
通过该实例证明，Encoder、Decoder的本质也是Handler，它们的执行顺序、使用方法与Handler保持一致。
执行顺序是：Encoder 先注册的后执行，与OutboundHandler一致；Decoder是先注册的先执行，与InboundHandler一致。
原文：http://blog.csdn.net/u/article/details/
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：277430次
积分：4650
积分：4650
排名：第4561名
原创：21篇
转载：1141篇
评论：18条
(1)(56)(43)(43)(47)(63)(29)(51)(124)(23)(25)(10)(27)(20)(17)(14)(1)(42)(41)(2)(20)(38)(232)(4)(3)(28)(4)(43)(20)(22)(51)(1)(21)(1)(1)(6)(1)(1)(1)Copyright © 2008&#x . All rights reserved.io.netty.handler.codec.DecoderException: com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero) - Stack Overflow
to customize your list.
Join the Stack Overflow Community
Stack Overflow is a community of 6.3 million programmers, just like you, helping each other.
J it only takes a minute:
My problem is that I get this kind of exception sometimes:
[nioEventLoopGroup-2-6] WARN io.netty.channel.DefaultChannelPipeline - An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
io.netty.handler.codec.DecoderException: com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:99)
at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:368)
at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:353)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:173)
at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:368)
at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:353)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:173)
at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:368)
at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:353)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:780)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:100)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:497)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:465)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:359)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:101)
at java.lang.Thread.run(Thread.java:680)
Caused by: com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
at com.google.protobuf.InvalidProtocolBufferException.invalidTag(InvalidProtocolBufferException.java:89)
at com.google.protobuf.CodedInputStream.readTag(CodedInputStream.java:108)
at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:460)
at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:579)
at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:280)
at com.google.protobuf.CodedInputStream.readGroup(CodedInputStream.java:240)
at com.google.protobuf.UnknownFieldSet$Builder.mergeFieldFrom(UnknownFieldSet.java:488)
at com.google.protobuf.AbstractMessage$Builder.mergeFieldFrom(AbstractMessage.java:508)
at com.google.protobuf.GeneratedMessage$ExtendableMessage.parseUnknownField(GeneratedMessage.java:661)
at eu.domain.netty.protobuf.ProtocolMessages$WirePayload.&init&(ProtocolMessages.java:5299)
at eu.domain.netty.protobuf.ProtocolMessages$WirePayload.&init&(ProtocolMessages.java:5282)
at eu.domain.netty.protobuf.ProtocolMessages$WirePayload$1.parsePartialFrom(ProtocolMessages.java:5452)
at eu.domain.netty.protobuf.ProtocolMessages$WirePayload$1.parsePartialFrom(ProtocolMessages.java:1)
at com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:141)
at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:176)
at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:182)
at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:49)
at io.netty.handler.codec.protobuf.ProtobufDecoder.decode(ProtobufDecoder.java:114)
at io.netty.handler.codec.protobuf.ProtobufDecoder.decode(ProtobufDecoder.java:62)
at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:89)
... 15 more
[nioEventLoopGroup-2-6] WARN eu.domain.netty.handler.ClientConnectResponseHandler - Exception caught during RPC connection handshake.
io.netty.handler.codec.DecoderException: com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:99)
at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:368)
at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:353)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:173)
at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:368)
at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:353)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:173)
at io.netty.channel.DefaultChannelHandlerContext.invokeChannelRead(DefaultChannelHandlerContext.java:368)
at io.netty.channel.DefaultChannelHandlerContext.fireChannelRead(DefaultChannelHandlerContext.java:353)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:780)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:100)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:497)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:465)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:359)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:101)
at java.lang.Thread.run(Thread.java:680)
Caused by: com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
at com.google.protobuf.InvalidProtocolBufferException.invalidTag(InvalidProtocolBufferException.java:89)
at com.google.protobuf.CodedInputStream.readTag(CodedInputStream.java:108)
at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:460)
at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:579)
at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:280)
at com.google.protobuf.CodedInputStream.readGroup(CodedInputStream.java:240)
at com.google.protobuf.UnknownFieldSet$Builder.mergeFieldFrom(UnknownFieldSet.java:488)
at com.google.protobuf.AbstractMessage$Builder.mergeFieldFrom(AbstractMessage.java:508)
at com.google.protobuf.GeneratedMessage$ExtendableMessage.parseUnknownField(GeneratedMessage.java:661)
at eu.domain.netty.protobuf.ProtocolMessages$WirePayload.&init&(ProtocolMessages.java:5299)
at eu.domain.netty.protobuf.ProtocolMessages$WirePayload.&init&(ProtocolMessages.java:5282)
at eu.domain.netty.protobuf.ProtocolMessages$WirePayload$1.parsePartialFrom(ProtocolMessages.java:5452)
at eu.domain.netty.protobuf.ProtocolMessages$WirePayload$1.parsePartialFrom(ProtocolMessages.java:1)
at com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:141)
at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:176)
at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:182)
at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:49)
at io.netty.handler.codec.protobuf.ProtobufDecoder.decode(ProtobufDecoder.java:114)
at io.netty.handler.codec.protobuf.ProtobufDecoder.decode(ProtobufDecoder.java:62)
at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:89)
... 15 more
Sounds like partial parsing of protobuf messages fails. I guess, this is something on Netty encoding/decoding level.
Can I get some comments on that?
UPD: my problem was in decoder:
out.add(MyMessage.parseFrom(buf));
where buf was zero length array
UPD2: there is chance that message was corrupted, so decoder was unable to decode, or maybe it is bug in coder/decoder realisation (?!)
Your Answer
Sign up or
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Post as a guest
By posting your answer, you agree to the
Not the answer you're looking for?
Browse other questions tagged
Stack Overflow works best with JavaScript enabledJava Code Example io.netty.handler.codec.http.HttpResponse
Java Code Examples for io.netty.handler.codec.http.HttpResponse
The following are top voted examples for showing how to use
io.netty.handler.codec.http.HttpResponse. These examples are extracted from open source projects.
You can vote up the examples you like and your votes will be used in our system to product
more good examples.
+ Save this class to your library
@SuppressWarnings(&deprecation&)
private SpdyHeadersFrame createHeadersFrame(HttpResponse httpResponse) throws Exception {
// Get the Stream-ID from the headers
final HttpHeaders httpHeaders = httpResponse.headers();
int streamId = SpdyHttpHeaders.getStreamId(httpResponse);
httpHeaders.remove(SpdyHttpHeaders.Names.STREAM_ID);
// The Connection, Keep-Alive, Proxy-Connection, and Transfer-Encoding
// headers are not valid and MUST not be sent.
httpHeaders.remove(HttpHeaders.Names.CONNECTION);
httpHeaders.remove(&Keep-Alive&);
httpHeaders.remove(&Proxy-Connection&);
httpHeaders.remove(HttpHeaders.Names.TRANSFER_ENCODING);
SpdyHeadersFrame spdyHeadersF
if (SpdyCodecUtil.isServerId(streamId)) {
spdyHeadersFrame = new DefaultSpdyHeadersFrame(streamId);
spdyHeadersFrame = new DefaultSpdySynReplyFrame(streamId);
SpdyHeaders frameHeaders = spdyHeadersFrame.headers();
// Unfold the first line of the response into name/value pairs
frameHeaders.set(SpdyHeaders.HttpNames.STATUS, httpResponse.getStatus().code());
frameHeaders.set(SpdyHeaders.HttpNames.VERSION, httpResponse.getProtocolVersion());
// Transfer the remaining HTTP headers
for (Map.Entry&String, String& entry: httpHeaders) {
spdyHeadersFrame.headers().add(entry.getKey(), entry.getValue());
currentStreamId = streamId;
spdyHeadersFrame.setLast(isLast(httpResponse));
return spdyHeadersF
private boolean handleHttpResponse(final HttpResponse response, final Channel channel, final NettyResponseFuture&?& future, AsyncHandler&?& handler) throws Exception {
HttpRequest httpRequest = future.getNettyRequest().getHttpRequest();
ProxyServer proxyServer = future.getProxyServer();
logger.debug(&\n\nRequest {}\n\nResponse {}\n&, httpRequest, response);
// store the original headers so we can re-send all them to
// the handler in case of trailing headers
future.setHttpHeaders(response.headers());
future.setKeepAlive(config.getKeepAliveStrategy().keepAlive(future.getTargetRequest(), httpRequest, response));
NettyResponseStatus status = new NettyResponseStatus(future.getUri(), config, response, channel);
int statusCode = response.getStatus().code();
Request request = future.getCurrentRequest();
Realm realm = request.getRealm() != null ? request.getRealm() : config.getRealm();
NettyResponseHeaders responseHeaders = new NettyResponseHeaders(response.headers());
return exitAfterProcessingFilters(channel, future, handler, status, responseHeaders) || //
exitAfterHandling401(channel, future, response, request, statusCode, realm, proxyServer) || //
exitAfterHandling407(channel, future, response, request, statusCode, proxyServer) || //
exitAfterHandling100(channel, future, statusCode) || //
exitAfterHandlingRedirect(channel, future, response, request, statusCode, realm) || //
exitAfterHandlingConnect(channel, future, request, proxyServer, statusCode, httpRequest) || //
exitAfterHandlingStatus(channel, future, response, handler, status) || //
exitAfterHandlingHeaders(channel, future, response, handler, responseHeaders) || exitAfterHandlingReactiveStreams(channel, future, response, handler);
public HttpResponse handleRequest(HttpRequest request, Matcher uriMatcher, ByteBuf body) throws RequestException {
final String responseBody = handle(request, uriMatcher, body);
final ByteBuf buffer = Unpooled.copiedBuffer(responseBody, CharsetUtil.UTF_8);
final HttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buffer);
final HttpHeaders headers = response.headers();
headers.add(HttpHeaders.Names.CONTENT_TYPE, &application/ charset=UTF-8&);
headers.add(HttpHeaders.Names.CONTENT_LENGTH, buffer.readableBytes());
private boolean setResponseKeepAlive(HttpResponse response) {
HttpHeaders headers = response.headers();
boolean closeConn = HttpHeaders.Values.CLOSE.equalsIgnoreCase(headers.get(HttpHeaders.Names.CONNECTION));
boolean responseKeepAlive = this.keepAlive && !closeC
if (responseKeepAlive) {
headers.set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
headers.set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);
return responseKeepA
public HttpResponse toResponse(Event&?& evt, Charset charset) {
if (!canHaveBody(getResponseCode()) && (message != null || listener != null)) {
if (listener != ChannelFutureListener.CLOSE) {
System.err.println(evt
+ & attempts to attach a body to & + getResponseCode()
+ & which cannot have one: & + message
+ & - & + listener);
ByteBuf buf = writeMessage(evt, charset);
if (buf != null) {
long size = buf.readableBytes();
add(Headers.CONTENT_LENGTH, size);
DefaultFullHttpResponse r = new DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, getResponseCode(), buf);
resp = new HackHttpResponse(getResponseCode(), this.status == NOT_MODIFIED ? false : chunked);
for (Entry&?& e : headers) {
// Remove things which cause problems for non-modified responses -
// browsers will hold the connection open regardless
if (this.status == NOT_MODIFIED) {
if (e.decorator == Headers.CONTENT_LENGTH) {
} else if (HttpHeaders.Names.CONTENT_ENCODING.equals(e.decorator.name())) {
} else if (Headers.TRANSFER_ENCODING.name().equals(e.decorator.name())) {
e.write(resp);
// Ensure a 0 content length is present for items with no content
if (message == null && listener == null && resp.headers() instanceof HackHttpHeaders) {
((HackHttpHeaders) resp.headers()).orig.set(Headers.CONTENT_LENGTH.name(), 0);
((HackHttpHeaders) resp.headers()).remove(Headers.TRANSFER_ENCODING.name());
((HackHttpHeaders) resp.headers()).remove(Headers.CONTENT_ENCODING.name());
private SpdySynReplyFrame createSynReplyFrame(HttpResponse httpResponse)
throws Exception {
// Get the Stream-ID from the headers
int streamID = SpdyHttpHeaders.getStreamId(httpResponse);
SpdyHttpHeaders.removeStreamId(httpResponse);
// The Connection, Keep-Alive, Proxy-Connection, and Transfer-Encoding
// headers are not valid and MUST not be sent.
httpResponse.headers().remove(HttpHeaders.Names.CONNECTION);
httpResponse.headers().remove(&Keep-Alive&);
httpResponse.headers().remove(&Proxy-Connection&);
httpResponse.headers().remove(HttpHeaders.Names.TRANSFER_ENCODING);
SpdySynReplyFrame spdySynReplyFrame = new DefaultSpdySynReplyFrame(streamID);
// Unfold the first line of the response into name/value pairs
SpdyHeaders.setStatus(spdyVersion, spdySynReplyFrame, httpResponse.getStatus());
SpdyHeaders.setVersion(spdyVersion, spdySynReplyFrame, httpResponse.getProtocolVersion());
// Transfer the remaining HTTP headers
for (Map.Entry&String, String& entry: httpResponse.headers()) {
spdySynReplyFrame.headers().add(entry.getKey(), entry.getValue());
currentStreamId = streamID;
spdySynReplyFrame.setLast(isLast(httpResponse));
return spdySynReplyF
* If and only if our proxy is not running in transparent mode, modify the
* response headers to reflect that it was proxied.
* @param httpResponse
private void modifyResponseHeadersToReflectProxying(
HttpResponse httpResponse) {
if (!proxyServer.isTransparent()) {
HttpHeaders headers = httpResponse.headers();
stripConnectionTokens(headers);
stripHopByHopHeaders(headers);
ProxyUtils.addVia(httpResponse);
* RFC2616 Section 14.18
* A received message that does not have a Date header field MUST be
* assigned one by the recipient if the message will be cached by
* that recipient or gatewayed via a protocol which requires a Date.
if (!headers.contains(&Date&)) {
headers.set(&Date&, ProxyUtils.httpDate());
private HttpHeadersFrame createHttpHeadersFrame(HttpResponse httpResponse)
throws Exception {
// Get the Stream-ID from the headers
int streamId = HttpHeaders.getIntHeader(httpResponse, &X-SPDY-Stream-ID&);
httpResponse.headers().remove(&X-SPDY-Stream-ID&);
// The Connection, Keep-Alive, Proxy-Connection, and Transfer-Encoding
// headers are not valid and MUST not be sent.
httpResponse.headers().remove(HttpHeaders.Names.CONNECTION);
httpResponse.headers().remove(&Keep-Alive&);
httpResponse.headers().remove(&Proxy-Connection&);
httpResponse.headers().remove(HttpHeaders.Names.TRANSFER_ENCODING);
HttpHeadersFrame httpHeadersFrame = new DefaultHttpHeadersFrame(streamId);
// Unfold the first line of the response into name/value pairs
httpHeadersFrame.headers().set(&:status&, httpResponse.getStatus().code());
// Transfer the remaining HTTP headers
for (Map.Entry&String, String& entry : httpResponse.headers()) {
httpHeadersFrame.headers().add(entry.getKey(), entry.getValue());
return httpHeadersF
private static void assertCORSPreflightResponseHeaders(final HttpResponse response, HttpMethod... methods) {
final HttpHeaders headers = response.headers();
assertThat(headers.get(CONTENT_TYPE), is(&text/ charset=UTF-8&));
assertThat(headers.get(CACHE_CONTROL), containsString(&public&));
assertThat(headers.get(CACHE_CONTROL), containsString(&max-age=&));
assertThat(headers.get(ACCESS_CONTROL_ALLOW_CREDENTIALS), is(&true&));
assertThat(headers.get(ACCESS_CONTROL_MAX_AGE), is(&&));
for (HttpMethod method : methods) {
assertThat(headers.get(ACCESS_CONTROL_ALLOW_METHODS), containsString(method.toString()));
assertThat(headers.get(ACCESS_CONTROL_ALLOW_HEADERS), is(&Content-Type&));
assertThat(headers.get(ACCESS_CONTROL_ALLOW_CREDENTIALS), is(&true&));
assertThat(headers.get(EXPIRES), is(notNullValue()));
assertThat(headers.get(SET_COOKIE), is(&JSESSIONID=path=/&));
Example 10
public static void copyMetadata(@NotNull Multimap&String, String& headers, @NotNull HttpResponse response) {
HttpHeaders httpHeaders = response.headers();
for (Map.Entry&String, String& elt : headers.entries()) {
if (elt.getKey().equals(AUTH_TOKEN_HEADER)) {
if (elt.getKey().equals(CONTENT_TYPE_HEADER)) {
httpHeaders.add(CONTENT_TYPE, elt.getValue());
httpHeaders.add(elt.getKey(), elt.getValue());}