胖虎的工具箱-编程导航 胖虎的工具箱-编程导航

用netty实现tcp简单通信

2年前 (2022) 程序员胖胖胖虎阿
211 0 0

一般我们想要使用tcp通信，最原始的方法就是使用socket套接字，但是每次都要创建新的socket对象，并且在finally中关闭，好浪费资源，很麻烦。所以就是用netty工具，可以轻松地实现tcp通信，当然也可以实现http通信，因为我们需要用tcp通信，所以这里只研究tcp。

最好先了解一下netty是干嘛的，以及NIO是什么。

netty：

https://blog.csdn.net/qq_34730511/article/details/98472924?ops_request_misc=&request_id=&biz_id=102&utm_term=netty&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduweb~default-8-98472924.nonecase&spm=1018.2226.3001.4187

BIO  NIO：

https://blog.csdn.net/weixin_34301132/article/details/85786733?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522164499824216780264047131%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=164499824216780264047131&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduend~default-1-85786733.first_rank_v2_pc_rank_v29&utm_term=NIO+BIO+AIO&spm=1018.2226.3001.4187

接下来就建项目了。这里我用的是maven项目，jdk1.8

  1. 引入netty的依赖
<!--netty配置-->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.36.Final</version>
</dependency>

创建NettyServer类，在这个类中先创建一个方法，然后在这个方法中创建事件线程组，这个构造器里面可以不传参数。写一个run()方法，里边对需要的参数进行配置，

package com.yy.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LoggingHandler;

public class NettyServer {

    public void run() {
        //创建事件线程组，
        EventLoopGroup boosGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        //创建服务启动器
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        try {
            //给服务启动器配置参数
            serverBootstrap.group(boosGroup, workerGroup)//绑定线程组
                    .channel(NioServerSocketChannel.class)//Nio模式
                    .option(ChannelOption.SO_BACKLOG, 1024)//标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。主要是作用于boss线程，用于处理新连接
                    .childOption(ChannelOption.SO_KEEPALIVE, true)//启用心跳保活机制,主要作用与worker线程，也就是已创建的channel。
                    .handler(new LoggingHandler("DEBUG"))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline()
                                    .addLast(new StringDecoder())//加载编解码器
                                    .addLast(new StringEncoder())
                                    .addLast(new NettyServerHandler());//加载handler处理器
                        }
                    });

            //绑定端口,这里用到的是ChannelFuture类
            ChannelFuture future = serverBootstrap.bind(8080).sync();
            System.out.println("服务端启动成功....");
            //关闭通道
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //在这里关闭事件线程组,释放资源
            boosGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        new NettyServer().run();
    }

}

再创建另外一个处理类，用于接收客户端发送的消息，并且给客户端发送应答，这个类继承于ChannelInboundHandlerAdapter类，重写其中的方法即可：

package com.yy.netty;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.log4j.Logger;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.InetAddress;

/**
 * 接收客户端发送的消息并返回应答
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    private static Logger logger = Logger.getLogger(NettyServerHandler.class);

    /**
     * 读取客户端的数据，也可以在这个方法中给ctx赋值，也就是给客户端返回的应答
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//        ByteBuf in = (ByteBuf)msg;
//        int readableBytes = in.readableBytes();
//        byte[] bytes =new byte[readableBytes];
//        in.readBytes(bytes);
//        System.out.println("服务端接收的倒的消息："+new String(bytes));
//        System.out.println("远程连接到的地址为："+ctx.channel().remoteAddress());
        //System.out.print(in.toString(CharsetUtil.UTF_8));
        System.out.println("服务端接收到一条来自客户端的消息："+msg);
        ctx.write("我是服务端，我收到客户端的消息了，我现在给你返回");
       // logger.error("服务端接受的消息 : " + msg);
    }

    /**
     * 发生异常时关闭
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    /**
     * 读取数据完毕,就刷新ctx对象，把应答的信息推送给客户端
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
        //也可以使用下面的方法，一次性将应答信息写入并且推送出去，但是一般不这么用，一般是在read方法中返回应答信息，处理复杂的业务逻辑，在complete方法中推送
        //ctx.writeAndFlush(Unpooled.copiedBuffer("Hello World,I am Server.", CharsetUtil.UTF_8));
    }

    /**
     * 将object类型转为字节数组的方法
     * @param object
     * @return
     * @throws IOException
     */
    public static byte[] objectToBytes(final Serializable object) throws IOException {

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream oos  =  null;
        try {
            oos = new ObjectOutputStream(baos);
            oos.writeObject(object);
            oos.flush();
            return baos.toByteArray();
        } finally {
            if (oos != null)  {
                oos.close();
            }
            if (baos != null) {
                baos.close();
            }
        }
    }
}

3)创建NettyClient类

package com.yy.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Scanner;
import java.util.concurrent.TimeUnit;

public class NettyClient {
    private SocketChannel socketChannel;
    public  void run(Object msg){
        //配置线程组
        EventLoopGroup group = new NioEventLoopGroup();
        //创建服务启动器
        Bootstrap bootstrap = new Bootstrap();

        //配置参数
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline()
                                    .addLast(new StringEncoder())
                                    .addLast(new StringDecoder())
                                    .addLast(new NettyClientHandler());
                    }
                })
                .remoteAddress("127.0.0.1",8080);

        //连接
             ChannelFuture future = bootstrap.connect();
             System.out.println("客户端正在连接服务端...");
        //客户端断线重连逻辑
        future.addListener((ChannelFutureListener) future1 -> {
            if (future1.isSuccess()) {
                System.out.println("连接Netty服务端成功");
                future.channel().writeAndFlush(msg);
            } else {
                System.out.println("连接失败，进行断线重连");
                future1.channel().eventLoop().schedule(() -> run(msg), 20, TimeUnit.SECONDS);
            }
        });
        socketChannel = (SocketChannel) future.channel();

    }

    public static void main(String[] args) {
        NettyClient nettyClient = new NettyClient();
        nettyClient.run("hi");
    }
}

创建Client的处理类

package com.yy.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import org.apache.log4j.Logger;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.Date;

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

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

    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    public static byte[] responseByte;
    /**
     * 管道就绪时触发该方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("云上建立连接时：" + sdf.format(new Date()));
        //ctx.writeAndFlush(Unpooled.copiedBuffer("Hello World,I am Client.", CharsetUtil.UTF_8));
    }

    /**
     * 管道读取事件时会触发该方法,
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("客户端收到服务端返回的消息："+msg);
        byte[] bytes = objectToBytes((Serializable) msg);
        //byte []scaleMsg=(byte[]) msg;
        StringBuffer originalData=new StringBuffer();
        for (byte b : bytes) {
            originalData.append(b+" ");
        }
        System.out.println(originalData.toString());
        if(bytes!=null){
            responseByte = bytes;
        }


    }

    /**
     * 发生异常时触发
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

    public static byte[] objectToBytes(final Serializable object) throws IOException {

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream oos  =  null;
        try {
                oos = new ObjectOutputStream(baos);
            oos.writeObject(object);
            oos.flush();
            return baos.toByteArray();
        } finally {
            if (oos != null)  {
                oos.close();
            }
            if (baos != null) {
                baos.close();
            }
        }
    }
}

单机测试是否能连通，先启动服务端代码，再启动客户端代码，就可以直接发数据了，效果如下图所示：

服务端：

用netty实现tcp简单通信 

客户端：

用netty实现tcp简单通信

现在暂时只能发送一次数据，但是服务端始终处于监听的状态，后边再研究吧！

版权声明：程序员胖胖胖虎阿 发表于 2022年10月5日 上午3:00。
转载请注明：用netty实现tcp简单通信 | 胖虎的工具箱-编程导航

相关文章

IDEA永久 激活码(IDEA永久破解版永久有效)
程序员胖胖胖虎阿
428
【Java】static关键字，内部类
程序员胖胖胖虎阿
107
IDEA全新永久激活码(idea的安装及激活码激活教程,激活破解教程)
程序员胖胖胖虎阿
306
IntelliJ IDEA2020注册码(Intellij IDEA2022.2 破解教程,绝对有效)
程序员胖胖胖虎阿
113
IDEA2022.2 破解教程(IDEA专业版破解插件亲测有效!!)
程序员胖胖胖虎阿
234
IntelliJ IDEA专业版 破解教程(IDEA安装及激活码激活教程,全年更新)
程序员胖胖胖虎阿
116

暂无评论

暂无评论...
致力于称为最好的程序员导航网站

友链申请 免责声明 关于我们

Copyright © 2022  胖虎的工具箱-编程导航   陇ICP备2022001249号