使用netty框架接收物联网数据

1.使用到的pom.xml

<dependency>

    <groupId>io.netty</groupId>

    <artifactId>netty-all</artifactId>

</dependency>

2.客户端发送数据

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

import java.io.PrintWriter;

import java.net.Socket;

import java.net.UnknownHostException;

public class Sockt {

    public static void main(String[] args) {

        try {

            //创建一个Socket，跟服务器的8080端口链接

            Socket socket = new Socket("127.0.0.1",监听端口);

            //使用PrintWriter和BufferedReader进行读写数据

            PrintWriter pw = new PrintWriter(socket.getOutputStream());

            StringBuffer sb = new StringBuffer("上位机数据");

            BufferedReader is = new BufferedReader(new InputStreamReader(socket.getInputStream()));

            //发送数据

            pw.println(sb);

            pw.flush();

            //接收数据

            String line = is.readLine();

            System.out.println("数据==》" + line);

            //关闭资源

            pw.close();

            is.close();

            socket.close();

        } catch (UnknownHostException e) {

            // TODO Auto-generated catch block

            e.printStackTrace();

        } catch (IOException e) {

            // TODO Auto-generated catch block

            e.printStackTrace();

        }

    }

}

3.服务器端使用netty接收数据

import io.netty.bootstrap.ServerBootstrap;

import io.netty.buffer.ByteBuf;

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.LengthFieldBasedFrameDecoder;

import io.netty.handler.codec.LengthFieldPrepender;

import io.netty.handler.codec.MessageToByteEncoder;

import org.springframework.stereotype.Component;

/**

 * 服务器

 *

 * @author

 */

@Component

public class TCPServer {

    public void run() {

        // 8080 是你要监听的端口

        int port = 8080;

        EventLoopGroup boosGroup = new NioEventLoopGroup();

        EventLoopGroup workGroup = new NioEventLoopGroup();

        try {

            ServerBootstrap bootstrap = new ServerBootstrap();

            bootstrap.group(boosGroup, workGroup)

                    .channel(NioServerSocketChannel.class)

                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override

                        protected void initChannel(SocketChannel ch) throws Exception {

                            ChannelPipeline pipeline = ch.pipeline();

                            pipeline.addLast(new TCPHandler());

                            pipeline.addLast("encoder", new MessageToByteEncoder<byte[]>() {

                                @Override

                                protected void encode(ChannelHandlerContext ctx, byte[] msg, ByteBuf out) throws Exception {

                                    out.writeBytes(msg);

                                }

                            });

                            pipeline.addLast(new LengthFieldBasedFrameDecoder(65535

                                    , 0, 2, 0, 2));

                            pipeline.addLast(new LengthFieldPrepender(2));

                        }

                    })

                    .option(ChannelOption.SO_BACKLOG, 128)

                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            System.out.println("服务启动...");

            ChannelFuture channelFuture = bootstrap.bind(port).sync();

            channelFuture.channel().closeFuture().sync();

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            workGroup.shutdownGracefully();

            boosGroup.shutdownGracefully();

            System.out.println("服务关闭...");

        }

    }

}

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelFutureListener;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.SimpleChannelInboundHandler;

import io.netty.util.CharsetUtil;

import org.springframework.stereotype.Component;

import java.net.SocketAddress;

import java.text.SimpleDateFormat;

import java.util.Date;

/**

 * @author

 */

@Component

public class TCPHandler extends SimpleChannelInboundHandler<ByteBuf> {

    /**

     * 读取消息

     *

     * @param ctx

     * @param m

     * @throws Exception

     */

    @Override

    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf m) throws Exception {

        byte[] data = new byte[m.readableBytes()];

        m.readBytes(data);

        String msg = new String(data, CharsetUtil.UTF_8);

        SocketAddress address = ctx.channel().remoteAddress();

        System.out.println(address.toString().substring(1) + "---" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));

        System.out.println(msg);

    }

    /**

     * 发生异常

     *

     * @param ctx

     * @param cause

     * @throws Exception

     */

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        if (ctx.channel().isActive()) {

            ctx.channel().writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);

        }

    }

}

import com.example.socket.utils.TCPServer;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication

public class SocketApplication {

    public static void main(String[] args) {

        SpringApplication.run(SocketApplication.class, args);

        new Thread() {

            @Override

            public void run() {

                TCPServer server = new TCPServer();

                server.run();

            }

        }.start();

    }

}

yml文件

#springboot web访问端口

server:

  port: 9055

# netty配置

netty:

  # 端口号

  port: 6666

  # 最大线程数

  maxThreads: 1024

  # 数据包的最大长度

  max_frame_length: 65535

使用客户端发送数据后，服务器端就能接收到数据了。