利用Java搭建个简单的Netty通信实例教程

 更新时间:2020年05月17日 15:17:52   作者:严阵以待  
这篇文章主要给大家介绍了关于如何利用Java搭建个简单的Netty通信,文中通过示例代码介绍的非常详细,对大家学习或者使用Java具有一定的参考学习价值,需要的朋友们下面来一起学习学习吧

前言

看过dubbo源码的同学应该都清楚,使用dubbo协议的底层通信是使用的netty进行交互,而最近看了dubbo的Netty部分后,自己写了个简单的Netty通信例子。

准备

工程截图

模块详解

  • rpc-common

rpc-common作为各个模块都需使用的模块,工程中出现的是一些通信时请求的参数以及返回的参数,还有一些序列化的工具。

  • rpc-client

rpc-client中目前只是单单的一个NettyClient启动类。

  • rpc-server

rpc-client中目前也只是单单的一个NettyServer服务启动类。

需要的依赖

目前所有的依赖项都出现在 rpc-common 下的 pom.xml中。

<dependencies>
  <!-- Netty -->
  <dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.10.Final</version>
  </dependency>

  <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.25</version>
  </dependency>

  <!-- Protostuff -->
  <dependency>
    <groupId>com.dyuproject.protostuff</groupId>
    <artifactId>protostuff-core</artifactId>
    <version>1.0.9</version>
  </dependency>

  <dependency>
    <groupId>com.dyuproject.protostuff</groupId>
    <artifactId>protostuff-runtime</artifactId>
    <version>1.0.9</version>
  </dependency>

  <!-- Objenesis -->
  <dependency>
    <groupId>org.objenesis</groupId>
    <artifactId>objenesis</artifactId>
    <version>2.1</version>
  </dependency>

  <!-- fastjson -->
  <dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.38</version>
  </dependency>
</dependencies>

实现

首先我们在common中先定义本次的Request和Response的基类对象。

public class Request {

  private String requestId;

  private Object parameter;

  public String getRequestId() {
    return requestId;
  }

  public void setRequestId(String requestId) {
    this.requestId = requestId;
  }

  public Object getParameter() {
    return parameter;
  }

  public void setParameter(Object parameter) {
    this.parameter = parameter;
  }
}

public class Response {

  private String requestId;

  private Object result;

  public String getRequestId() {
    return requestId;
  }

  public void setRequestId(String requestId) {
    this.requestId = requestId;
  }

  public Object getResult() {
    return result;
  }

  public void setResult(Object result) {
    this.result = result;
  }
}

使用fastJson进行本次序列化

Netty对象的序列化转换很好懂, ByteToMessageDecoder 和 MessageToByteEncoder 分别只要继承它们,重写方法后,获取到Object和Byte,各自转换就OK。

不过如果是有要用到生产上的同学,建议不要使用 fastJson,因为它的漏洞补丁真的是太多了,可以使用google的 protostuff。

public class RpcDecoder extends ByteToMessageDecoder {

  // 目标对象类型进行解码
  private Class<?> target;

  public RpcDecoder(Class target) {
    this.target = target;
  }

  @Override
  protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    if (in.readableBytes() < 4) {  // 不够长度丢弃
      return;
    }
    in.markReaderIndex();  // 标记一下当前的readIndex的位置
    int dataLength = in.readInt(); // 读取传送过来的消息的长度。ByteBuf 的readInt()方法会让他的readIndex增加4

    if (in.readableBytes() < dataLength) { // 读到的消息体长度如果小于我们传送过来的消息长度,则resetReaderIndex. 这个配合markReaderIndex使用的。把readIndex重置到mark的地方
      in.resetReaderIndex();
      return;
    }
    byte[] data = new byte[dataLength];
    in.readBytes(data);

    Object obj = JSON.parseObject(data, target);  // 将byte数据转化为我们需要的对象
    out.add(obj);
  }
}

public class RpcEncoder extends MessageToByteEncoder {

  //目标对象类型进行编码
  private Class<?> target;

  public RpcEncoder(Class target) {
    this.target = target;
  }

  @Override
  protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
    if (target.isInstance(msg)) {
      byte[] data = JSON.toJSONBytes(msg);  // 使用fastJson将对象转换为byte
      out.writeInt(data.length); // 先将消息长度写入,也就是消息头
      out.writeBytes(data);  // 消息体中包含我们要发送的数据
    }
  }

}

NetyServer

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    Request request = (Request) msg;

    System.out.println("Client Data:" + JSON.toJSONString(request));

    Response response = new Response();
    response.setRequestId(request.getRequestId());
    response.setResult("Hello Client !");

    // client接收到信息后主动关闭掉连接
    ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
  }

  @Override
  public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    ctx.flush();
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    ctx.close();
  }
}

public class NettyServer {

  private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

  private String ip;
  private int port;

  public NettyServer(String ip, int port) {
    this.ip = ip;
    this.port = port;
  }

  public void server() throws Exception {

    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();

    try {

      final ServerBootstrap serverBootstrap = new ServerBootstrap();

      serverBootstrap.group(bossGroup, workerGroup)
          .channel(NioServerSocketChannel.class)
          .option(ChannelOption.SO_BACKLOG, 1024)
          .option(ChannelOption.SO_SNDBUF, 32 * 1024)
          .option(ChannelOption.SO_RCVBUF, 32 * 1024)
          .option(ChannelOption.SO_KEEPALIVE, true)
          .childHandler(new ChannelInitializer<SocketChannel>() {
            protected void initChannel(SocketChannel socketChannel) throws Exception {
              socketChannel.pipeline().addLast(new RpcDecoder(Request.class))
                  .addLast(new RpcEncoder(Response.class))
                  .addLast(new NettyServerHandler());
            }
          });

      serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true); // 开启长连接

      ChannelFuture future = serverBootstrap.bind(ip, port).sync();

//      if (future.isSuccess()) {
//
//        new Register().register("/yanzhenyidai/com.yanzhenyidai.server", ip + ":" + port);
//      }

      future.channel().closeFuture().sync();
    } finally {
      bossGroup.shutdownGracefully();
      workerGroup.shutdownGracefully();
    }
  }

  public static void main(String[] args) throws Exception {
    new NettyServer("127.0.0.1", 20000).server();
  }
}

关键名词:

  • EventLoopGroup
    • workerGroup
    • bossGroup
      Server端的EventLoopGroup分为两个,一般workerGroup作为处理请求,bossGroup作为接收请求。
  • ChannelOption
    • SO_BACKLOG
    • SO_SNDBUF
    • SO_RCVBUF
    • SO_KEEPALIVE
      以上四个常量作为TCP连接中的属性。
  • ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);

NettyServerHandler中出现的 ChannelFutureListener.CLOSE ,作为Server端主动关闭与Client端的通信,如果没有主动Close,那么NettyClient将会一直处于阻塞状态,得不到NettyServer的返回信息。

NettyClient

public class NettyClient extends SimpleChannelInboundHandler<Response> {

  private final String ip;
  private final int port;
  private Response response;

  public NettyClient(String ip, int port) {
    this.ip = ip;
    this.port = port;
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    ctx.close();
  }

  @Override
  protected void channelRead0(ChannelHandlerContext channelHandlerContext, Response response) throws Exception {
    this.response = response;
  }

  public Response client(Request request) throws Exception {
    EventLoopGroup group = new NioEventLoopGroup();

    try {

      // 创建并初始化 Netty 客户端 Bootstrap 对象
      Bootstrap bootstrap = new Bootstrap();
      bootstrap.group(group);
      bootstrap.channel(NioSocketChannel.class);
      bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel channel) throws Exception {
          ChannelPipeline pipeline = channel.pipeline();

          pipeline.addLast(new RpcDecoder(Response.class));
          pipeline.addLast(new RpcEncoder(Request.class));
          pipeline.addLast(NettyClient.this);
        }
      });
      bootstrap.option(ChannelOption.TCP_NODELAY, true);


//      String[] discover = new Discover().discover("/yanzhenyidai/com.yanzhenyidai.server").split(":");

      // 连接 RPC 服务器
      ChannelFuture future = bootstrap.connect(ip, port).sync();

      // 写入 RPC 请求数据并关闭连接
      Channel channel = future.channel();

      channel.writeAndFlush(request).sync();
      channel.closeFuture().sync();

      return response;
    } finally {
      group.shutdownGracefully();
    }
  }

  public static void main(String[] args) throws Exception {
    Request request = new Request();
    request.setRequestId(UUID.randomUUID().toString());
    request.setParameter("Hello Server !");
    System.out.println(JSON.toJSONString(new NettyClient("127.0.0.1", 30000).client(request)));
  }
}

测试

如果以上所有内容都准备就绪,那么就可以进行调试了。

启动顺序,先启动NettyServer,再启动NettyClient。

总结

记得刚出来工作时,有工作很多年的同事问我了不了解Netty,当时工作太短,直说听过Putty,现在回想起来真的挺丢人的,哈哈。😋

Netty作为通信框架,如果你了解TCP,而且项目中有类似传输信息的需求,又不想集成HTTP或者Socket,那么Netty真的挺实用的。

参考资料:

本项目Github地址:Netty-RPC

到此这篇关于利用Java搭建个简单的Netty通信的文章就介绍到这了,更多相关Java搭建Netty通信内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • Jlabel实现内容自动换行简单实例

    Jlabel实现内容自动换行简单实例

    这篇文章主要介绍了Jlabel实现内容自动换行简单实例,具有一定借鉴价值,需要的朋友可以参考下
    2018-01-01
  • SpringBoot集成IJPay实现微信v3支付的示例代码

    SpringBoot集成IJPay实现微信v3支付的示例代码

    本文主要介绍了SpringBoot集成IJPay实现微信v3支付的示例代码,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2023-07-07
  • 关于MyBatis中Mapper XML热加载优化

    关于MyBatis中Mapper XML热加载优化

    大家好,本篇文章主要讲的是关于MyBatis中Mapper XML热加载优化,感兴趣的同学赶快来看一看吧,对你有帮助的话记得收藏一下
    2022-01-01
  • Spring实战之使用@Resource配置依赖操作示例

    Spring实战之使用@Resource配置依赖操作示例

    这篇文章主要介绍了Spring实战之使用@Resource配置依赖操作,结合实例形式分析了Spring使用@Resource配置依赖具体步骤、实现及测试案例,需要的朋友可以参考下
    2019-12-12
  • Mybatis中关于自定义mapper.xml时,参数传递的方式及写法

    Mybatis中关于自定义mapper.xml时,参数传递的方式及写法

    这篇文章主要介绍了Mybatis中关于自定义mapper.xml时,参数传递的方式及写法,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2023-12-12
  • 减小Maven项目生成的JAR包体积实现提升运维效率

    减小Maven项目生成的JAR包体积实现提升运维效率

    在Maven构建Java项目过程中,减小JAR包体积可通过排除不必要的依赖和使依赖jar包独立于应用jar包来实现,在pom.xml文件中使用<exclusions>标签排除不需要的依赖,有助于显著降低JAR包大小,此外,将依赖打包到应用外,可减少应用包的体积
    2024-10-10
  • Java emoji持久化mysql过程详解

    Java emoji持久化mysql过程详解

    这篇文章主要介绍了Java emoji持久化mysql过程详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2019-07-07
  • Java 数据结构与算法系列精讲之队列

    Java 数据结构与算法系列精讲之队列

    这篇文章主要介绍了Java队列数据结构的实现,队列是一种特殊的线性表,只允许在表的队头进行删除操作,在表的后端进行插入操作,队列是一个有序表先进先出,想了解更多相关资料的小伙伴可以参考下面文章的详细内容
    2022-02-02
  • Java实战之小米交易商城系统的实现

    Java实战之小米交易商城系统的实现

    这篇文章将利用Java实现小米交易商城系统,文中采用的技术有:JSP 、Spring、SpringMVC、MyBatis等,感兴趣的小伙伴可以跟随小编一起学习一下
    2022-04-04
  • MyBatis中有关int和Integer的使用方式

    MyBatis中有关int和Integer的使用方式

    这篇文章主要介绍了MyBatis中有关int和Integer的使用方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2023-03-03

最新评论