udpServer 数据接收

钟日健

2021-02-24 6a65cca452ce1ec3d6132698c91b12eccdabca57

udpServer 数据接收

 blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/JfptApplication.java

@@ -18,12 +18,10 @@

import org.springblade.core.cloud.feign.EnableBladeFeign;
import org.springblade.core.launch.BladeApplication;
import org.springblade.core.launch.constant.AppConstant;
import org.springblade.jfpt.nettyServer.Server;
import org.springblade.jfpt.nettyUdpServer.UdpServer;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.cloud.client.SpringCloudApplication;

/**
 * Desk启动器
@@ -41,7 +39,9 @@

    @Override
    public void run(String... args) throws Exception {
        Server server=new Server(8088);
        UdpServer udpServer=new UdpServer(8088);
        //Server server=new Server(8088);
    }

}


 blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyUdpServer/UdpServer.java

New file
@@ -0,0 +1,123 @@
package org.springblade.jfpt.nettyUdpServer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.handler.codec.json.JsonObjectDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

/**
 * Created by wangshikai on 2016/7/18.
 */
public class UdpServer {
    private static Logger LOG = LoggerFactory.getLogger(UdpServer.class);

    public int port;

    private Channel channel;

    public UdpServer(int port) {
        this.port = port;
        run();
    }

    /**
     * 不用
     */
    public void init() {
        final NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(new ChannelInitializer<NioDatagramChannel>() {
                    @Override
                    public void initChannel(NioDatagramChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast("decoder", new JsonObjectDecoder());
                        p.addLast(new UdpServerHandler());
                    }
                });
            InetAddress address = InetAddress.getLocalHost();
            channel = b.bind(address, port).sync().channel();
            Executors.newSingleThreadExecutor().execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        channel.closeFuture().sync();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        group.shutdownGracefully();
                    }
                }
            });
            LOG.info("UDP服务器启动, host:{},port:{}", address.getHostAddress(), port);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    //给管道抽象出接口，给Channel更多的能力和配置，例如Channel的状态，参数，IO操作
    //使用ChannelPipeline实现自定义IO
    //Channel channel;
    public void run() {
        //InetSocketAddress socketAddress = new InetSocketAddress("s16s652780.51mypc.cn", 21403);
        //启动服务
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        //优化使用的线程
        final EventExecutorGroup group = new DefaultEventExecutorGroup(16);

        try {
            Bootstrap b = new Bootstrap();//udp不能使用ServerBootstrap
            b.group(workerGroup).channel(NioDatagramChannel.class)//设置UDP通道
                //设置udp的管道工厂
                .handler(new ChannelInitializer<NioDatagramChannel>() {
                    //NioDatagramChannel标志着是UDP格式的
                    @Override
                    protected void initChannel(NioDatagramChannel ch)
                        throws Exception {
                        // TODO Auto-generated method stub
                        //创建一个执行Handler的容器
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(new StringEncoder());
                        //执行具体的处理器
                        pipeline.addLast(group, "handler", new UdpServerHandler());//消息处理器
                    }

                })//初始化处理器
                //true / false 多播模式(UDP适用),可以向多个主机发送消息
                .option(ChannelOption.SO_BROADCAST, true)// 支持广播
                .option(ChannelOption.SO_RCVBUF, 2048 * 1024)// 设置UDP读缓冲区为2M
                .option(ChannelOption.SO_SNDBUF, 1024 * 1024);// 设置UDP写缓冲区为1M

            // 绑定端口，开始接收进来的连接
            ChannelFuture f = b.bind(port).sync();
            //获取channel通道
            //channel=f.channel();
            System.out.println("UDP Server 启动！");
            // 等待服务器 socket 关闭 。
            // 这不会发生，可以优雅地关闭服务器。
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 优雅退出 释放线程池资源
            group.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

}

 blade-service/blade-jfpts/src/main/java/org/springblade/jfpt/nettyUdpServer/UdpServerHandler.java

New file
@@ -0,0 +1,51 @@
package org.springblade.jfpt.nettyUdpServer;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * Created by wangshikai on 2016/7/18.
 */
@Component
public class UdpServerHandler extends SimpleChannelInboundHandler<DatagramPacket> {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket) throws Exception {
        // 读取收到的数据
        ByteBuf buf = (ByteBuf) datagramPacket.copy().content();
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, CharsetUtil.UTF_8);
        System.out.println("【NOTE】>>>>>> 收到客户端的数据："+body);

        // 回复一条信息给客户端
//        channelHandlerContext.writeAndFlush(new DatagramPacket(
//            Unpooled.copiedBuffer("Hello，我是Server，我的时间戳是"+System.currentTimeMillis()
//                , CharsetUtil.UTF_8)
//            , datagramPacket.sender())).sync();
    }

    //捕获异常
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {
        //logger.log(Level.INFO, "AuthServerInitHandler exceptionCaught");
        logger.error("UdpServerHandler exceptionCaught"+cause.getMessage());
        System.out.println("UdpServerHandler exceptionCaught"+cause.getMessage());
        cause.printStackTrace();
        ctx.close();
    }

    //消息没有结束的时候触发
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
}