Netty入门实例-使用POJO代替ByteBuf

2019-04-20 15:44|来源: 网路

到目前为止,我们上面几篇教程中的所有例子都使用ByteBuf作为协议消息的主要数据结构。 在本节中,我们将改进TIME协议的客户端和服务器示例,让它们使用POJO来代替原来的ByteBuf。

ChannelHandler中使用POJO的优点是显而易见的; 处理程序将从ByteBuf中提取信息的代码,将从处理程序中分离出来,变得更易维护和可重用。 在TIME客户端和服务器示例中,我们只读取一个32位整数,它不是直接使用ByteBuf来解码转换的。在实现真实世界协议时,这是必须要进行分离的。

首先,我们定义一个名为 UnixTime 的新类型(一个简单的Java类)。

package cn.netty.timepojo;

import java.util.Date;public class UnixTime {

    private final long value;

    public UnixTime() {
        this(System.currentTimeMillis() / 1000L + 2208988800L);
    }

    public UnixTime(long value) {
        this.value = value;
    }

    public long value() {
        return value;
    }

    @Override
    public String toString() {
        Date date = new Date((value() - 2208988800L) * 1000L);
        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String dateString = formatter.format(date);
        return dateString;
    }
}


我们现在来修改时间解码器(TimeDecoder)来生成UnixTime,而不是ByteBuf

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    if (in.readableBytes() < 4) {
        return;
    }

    out.add(new UnixTime(in.readUnsignedInt()));
}


使用更新的解码器,TimeClientHandler不再使用ByteBuf:

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    UnixTime m = (UnixTime) msg;
    System.out.println(m);
    ctx.close();
}


怎么样?看起更简单和优雅,对吧? 相同地也可以应用在服务器端。现在我们首先更新TimeServerHandler中的代码:

@Override
public void channelActive(ChannelHandlerContext ctx) {
    ChannelFuture f = ctx.writeAndFlush(new UnixTime());
    f.addListener(ChannelFutureListener.CLOSE);
}


现在,唯一缺少的是一个编码器,它是一个ChannelOutboundHandler的实现,是将UnixTime转换回ByteBuf。 它比编写解码器简单得多,因为在编码消息时不需要处理数据包分段和组合。

package cn.sxt.netty.timepojo;

public class TimeEncoder extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        UnixTime m = (UnixTime) msg;
        ByteBuf encoded = ctx.alloc().buffer(4);
        encoded.writeInt((int)m.value());
        ctx.write(encoded, promise); // (1)
    }
}


  1. 在这一行有很多重要的东西。
    首先,我们按原样传递原始的ChannelPromise,以便Netty将编码数据实际写入时将其标记为成功或失败。
    第二步,我们没有调用ctx.flush()。 有一个单独的处理程序方法void flush(ChannelHandlerContext ctx),它用于覆盖flush()操作。

要进一步简化,可以使用MessageToByteEncoder

public class TimeEncoder extends MessageToByteEncoder<UnixTime> {
    @Override
    protected void encode(ChannelHandlerContext ctx, UnixTime msg, ByteBuf out) {
        out.writeInt((int)msg.value());
    }
}


剩下的最后一个任务是在TimeServerHandler之前将TimeEncoder插入到服务器端的ChannelPipeline中,这里将留作一个简单的练习吧。


相关问答

更多

netty bytebuf怎么存入redis

于是决定通过NIO来实现和Redis服务器的网络连接,现在业界最优秀的NIO框架非Netty莫属了

netty的UnPooled ByteBuf是否需要引用计数?(Do netty's UnPooled ByteBuf need Reference counted?)

是的,有参考计数,以便我们及时免费直接记忆。 Yes there are reference counted to allow us free direct memory in a timely manner.

在netty中重用bytebuf(Reusing of bytebuf in netty)

共享Netty缓冲区时,您需要遵循基本引用计数的规则,这与基本垃圾收集的简单规则不同。 这些规则基本归结为: 从类中发送ByteBuf ,请调用retain() 如果您使用了ByteBuf ,请调用release() 大多数情况下,当您在发送它时同时使用bytebuf时,您可以删除这两个调用。 在您的示例中,在将共享的ByteBuf写入套接字时,应调用retain()来增加引用计数,因为该对象现在在2个不同的位置使用。 调用.retain()之后仍然需要做一个技巧,即调用.duplicate() ...

如何有效地将入站netty.io ByteBuf消息分发给ChannelGroup?(How to efficiently distribute inbound netty.io ByteBuf messages to a ChannelGroup?)

嗯...其实你的问题并不那么简单。 我会尽量简短回答。 如果您的应用中不需要,则无需将ByteBuf转换为byte[] 。 所以我假设你有下一个结构: public class Update { private final String id; private final UpdateType updateType; private final Map<String, ByteBuf> keyValueMapping; } 这里的问题是你部分解析ByteBuf 。 ...

为什么netty客户端返回Observable >而不是Observable ?(why does netty client return Observable> instead of Observable?)

是的,它本质上是一个Single<HttpClientResponse> 。 由于这些API是在RxJava中引入Single之前创建的,因此它们不使用Single Yes it is essentially a Single<HttpClientResponse>. Since these APIs were created before Single was introduced in RxJava, they do not use Single

Netty 4 - 将ByteBuf直接写入通道,绕过管道(Netty 4 - write ByteBuf directly to channel, bypassing pipeline)

得到它,获取管道末端的上下文并从那里写入: channel.pipeline().firstContext().write(myByteBuf) Got it, get the context at the end of the pipeline and write from there: channel.pipeline().firstContext().write(myByteBuf)

Netty:MessageToMessageEncoder并转换为ByteBuf(Netty: MessageToMessageEncoder and conversion to ByteBuf)

是。 最终,有效载荷必须采用ByteBuf的形式。 但是,按照预期,但未在您引用的示例中显示,MessageToMessage编码器将在下游管道中由另一个编码器跟踪,该编码器将消息转换为ByteBuf。 在您的情况下, http://netty.io/4.1/api/io/netty/handler/codec/string/StringEncoder.html可以解决这个问题,如该链接中的代码所示: ChannelPipeline pipeline = ...; // Decoders p ...

Netty PooledDirect ByteBuf和ArrayList意外行为(Netty PooledDirect ByteBuf and ArrayList unexpected behaviour)

那是因为你使用: for(int i=0;i<arr.size();i++){ ByteBuf b = arr.remove(i); b.release(); } 这将无法正常工作,因为您将索引增加0,但也删除缓冲区。 这样你就会跳过缓冲区。 更好地使用Queue并使用poll() ; Queue<ByteBuf> arr = new ArrayDeque<>(); for(;;){ ByteBuf b = arr.poll(); if (b == null ...

netty 4.1 ByteBuf endian访问字节(netty 4.1 ByteBuf endian accessing bytes)

即使使用ByteBuffer.order(...)直接访问字节也没有什么不同。 所以只需使用getBytes(...) 。 There is no different when accessing the bytes directly even when using ByteBuffer.order(...). So just use the getBytes(...).

如何知道Bytebuf是从Netty4中的哪个频道读取的?(How to know a Bytebuf was read from which channel in Netty4?)

您可以通过附加一些属性 channel.attr(key).set(...) You can attach some attribute via channel.attr(key).set(...)