付政委 bugstack虫洞栈 更新1-09 1-10

Author: fuzhengwei

itstack-demo-netty-1-09/pom.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>itstack-demo-netty</artifactId>
+        <groupId>org.itatack.demo</groupId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>itstack-demo-netty-1-09</artifactId>
+
+    <dependencies>
+        <!-- fastjson -->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.58</version>
+        </dependency>
+    </dependencies>
+
+</project>

itstack-demo-netty-1-09/src/main/java/org/itstack/demo/netty/codec/MyDecoder.java

@@ -0,0 +1,84 @@
+package org.itstack.demo.netty.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+
+import java.nio.charset.Charset;
+import java.util.List;
+
+/**
+ * 自定义解码器
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛关注获取学习源码｝
+ * 虫洞群：①群5398358 ②群5360692
+ * Create by fuzhengwei on 2019
+ */
+public class MyDecoder extends ByteToMessageDecoder {
+
+    //数据包基础长度
+    private final int BASE_LENGTH = 4;
+
+    @Override
+    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
+
+        //基础长度不足，我们设定基础长度为4
+        if (in.readableBytes() < BASE_LENGTH) {
+            return;
+        }
+
+        int beginIdx; //记录包头位置
+
+        while (true) {
+            // 获取包头开始的index
+            beginIdx = in.readerIndex();
+            // 标记包头开始的index
+            in.markReaderIndex();
+            // 读到了协议的开始标志，结束while循环
+            if (in.readByte() == 0x02) {
+                break;
+            }
+            // 未读到包头，略过一个字节
+            // 每次略过，一个字节，去读取，包头信息的开始标记
+            in.resetReaderIndex();
+            in.readByte();
+            // 当略过，一个字节之后，
+            // 数据包的长度，又变得不满足
+            // 此时，应该结束。等待后面的数据到达
+            if (in.readableBytes() < BASE_LENGTH) {
+                return;
+            }
+
+        }
+
+        //剩余长度不足可读取数量[没有内容长度位]
+        int readableCount = in.readableBytes();
+        if (readableCount <= 1) {
+            in.readerIndex(beginIdx);
+            return;
+        }
+
+        //长度域占4字节，读取int
+        ByteBuf byteBuf = in.readBytes(1);
+        String msgLengthStr = byteBuf.toString(Charset.forName("GBK"));
+        int msgLength = Integer.parseInt(msgLengthStr);
+
+        //剩余长度不足可读取数量[没有结尾标识]
+        readableCount = in.readableBytes();
+        if (readableCount < msgLength + 1) {
+            in.readerIndex(beginIdx);
+            return;
+        }
+
+        ByteBuf msgContent = in.readBytes(msgLength);
+
+        //如果没有结尾标识，还原指针位置[其他标识结尾]
+        byte end = in.readByte();
+        if (end != 0x03) {
+            in.readerIndex(beginIdx);
+            return;
+        }
+
+        out.add(msgContent.toString(Charset.forName("GBK")));
+    }
+}

itstack-demo-netty-1-09/src/main/java/org/itstack/demo/netty/codec/MyEncoder.java

@@ -0,0 +1,32 @@
+package org.itstack.demo.netty.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+
+/**
+ * 自定义编码器
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛关注获取学习源码｝
+ * 虫洞群：①群5398358 ②群5360692
+ * Create by fuzhengwei on 2019
+ */
+public class MyEncoder extends MessageToByteEncoder {
+
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext, Object in, ByteBuf out) throws Exception {
+
+        String msg = in.toString();
+        byte[] bytes = msg.getBytes();
+
+        byte[] send = new byte[bytes.length + 2];
+        System.arraycopy(bytes, 0, send, 1, bytes.length);
+        send[0] = 0x02;
+        send[send.length - 1] = 0x03;
+        
+        out.writeInt(send.length);
+        out.writeBytes(send);
+
+    }
+
+}

itstack-demo-netty-1-09/src/main/java/org/itstack/demo/netty/server/MyChannelInitializer.java

@@ -0,0 +1,28 @@
+package org.itstack.demo.netty.server;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import org.itstack.demo.netty.codec.MyDecoder;
+import org.itstack.demo.netty.codec.MyEncoder;
+
+import java.nio.charset.Charset;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛关注获取学习源码｝
+ * 虫洞群：①群5398358 ②群5360692
+ * Create by fuzhengwei on 2019
+ */
+public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
+
+    @Override
+    protected void initChannel(SocketChannel channel) {
+        //自定义解码器
+        channel.pipeline().addLast(new MyDecoder());
+        //自定义编码器
+        channel.pipeline().addLast(new MyEncoder());
+        //在管道中添加我们自己的接收数据实现方法
+        channel.pipeline().addLast(new MyServerHandler());
+    }
+
+}

itstack-demo-netty-1-09/src/main/java/org/itstack/demo/netty/server/MyServerHandler.java

@@ -0,0 +1,36 @@
+package org.itstack.demo.netty.server;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.socket.SocketChannel;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛关注获取学习源码｝
+ * 虫洞群：①群5398358 ②群5360692
+ * Create by fuzhengwei on 2019
+ */
+public class MyServerHandler extends ChannelInboundHandlerAdapter {
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        SocketChannel channel = (SocketChannel) ctx.channel();
+        System.out.println("链接报告开始 {公众号：bugstack虫洞栈 >获取学习源码}");
+        System.out.println("链接报告信息：有一客户端链接到本服务端");
+        System.out.println("链接报告IP:" + channel.localAddress().getHostString());
+        System.out.println("链接报告Port:" + channel.localAddress().getPort());
+        System.out.println("链接报告完毕");
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        //接收msg消息{与上一章节相比，此处已经不需要自己进行解码}
+        System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 接收到消息：" + msg);
+
+        ctx.writeAndFlush("hi I'm ok");
+    }
+
+}

itstack-demo-netty-1-09/src/main/java/org/itstack/demo/netty/server/NettyServer.java

@@ -0,0 +1,44 @@
+package org.itstack.demo.netty.server;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛关注获取学习源码｝
+ * 虫洞群：①群5398358 ②群5360692
+ * Create by fuzhengwei on 2019
+ */
+public class NettyServer {
+
+    public static void main(String[] args) {
+        new NettyServer().bing(7397);
+    }
+
+    private void bing(int port) {
+        //配置服务端NIO线程组
+        EventLoopGroup parentGroup = new NioEventLoopGroup(); //NioEventLoopGroup extends MultithreadEventLoopGroup Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
+        EventLoopGroup childGroup = new NioEventLoopGroup();
+        try {
+            ServerBootstrap b = new ServerBootstrap();
+            b.group(parentGroup, childGroup)
+                    .channel(NioServerSocketChannel.class)    //非阻塞模式
+                    .option(ChannelOption.SO_BACKLOG, 128)
+                    .childHandler(new MyChannelInitializer());
+            ChannelFuture f = b.bind(port).sync();
+            System.out.println("itstack-demo-netty server start done. {关注公众号：bugstack虫洞栈，获取源码}");
+            f.channel().closeFuture().sync();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        } finally {
+            childGroup.shutdownGracefully();
+            parentGroup.shutdownGracefully();
+        }
+
+    }
+
+}

itstack-demo-netty-1-09/src/test/java/org/itstack/demo/test/ApiTest.java

@@ -0,0 +1,132 @@
+package org.itstack.demo.test;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
+
+import java.nio.charset.Charset;
+import java.util.Arrays;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛关注获取学习源码｝
+ * 虫洞群：①群5398358 ②群5360692
+ * Create by fuzhengwei on 2019
+ */
+public class ApiTest {
+
+    public static void main(String[] args) {
+
+        ByteBuf in = Unpooled.buffer(5);
+        byte[] bytes = {2, 50, 104, 105, 3};
+        in.writeBytes(bytes);
+        in.readBytes(1);
+        ByteBuf byteBuf = in.readBytes(1);
+
+        System.out.println(byteBuf.toString(Charset.forName("GBK")));
+    }
+
+    public void test_03() {
+        ByteBuf in = Unpooled.buffer(5);
+        byte[] bytes = {104, 105, 0x02, 104, 105, 0x03, 104};
+        in.writeBytes(bytes);
+
+        //可读数据大小
+        int size = in.readableBytes();
+
+        //读取到byte
+        byte[] data = new byte[size];
+        in.readBytes(data);
+
+        int beginIdx = 0, endIdx = 0;
+        for (int i = 0; i < size; i++) {
+            if (data[i] == 0x02) {
+                beginIdx = i;
+                continue;
+            }
+            if (data[i] == 0x03) {
+                endIdx = i;
+                break;
+            }
+        }
+
+        ByteBuf copy = in.copy(beginIdx + 1, endIdx - beginIdx - 1);
+        int i = copy.readableBytes();
+        byte[] data2 = new byte[i];
+        copy.readBytes(data2);
+        System.out.println(Arrays.toString(data2));
+
+        //置位
+        in.resetReaderIndex();
+        in.readBytes(endIdx + 1);
+
+        //余下数据
+        int i2 = in.readableBytes();
+        byte[] data3 = new byte[i2];
+        in.readBytes(data3);
+        System.out.println(Arrays.toString(data3));
+    }
+
+    public void test_01() {
+        ByteBuf in = Unpooled.buffer(5);
+        byte[] bytes = {104, 105, 0x02, 104, 105, 0x03, 104};
+        in.writeBytes(bytes);
+
+        while (true) {
+            int i = in.readerIndex();
+            in.markReaderIndex();
+            if (in.readByte() == 0x02) {
+                break;
+            }
+            //未读到包头，略过一个字节
+            in.resetReaderIndex();
+            in.readByte();
+        }
+
+        //假定我们的数据长度是2
+        ByteBuf byteBuf = in.readBytes(2);
+        int size = byteBuf.readableBytes();
+        byte[] data = new byte[size];
+        byteBuf.readBytes(data);
+        in.readByte();
+        System.out.println(Arrays.toString(data));
+
+        //余下数据
+        int i = in.readableBytes();
+        byte[] data2 = new byte[i];
+        in.readBytes(data2);
+        System.out.println(Arrays.toString(data2));
+    }
+
+    public void test_02() {
+         /* int size = in.readableBytes();
+
+        byte[] data = new byte[size];
+        in.readBytes(data);
+
+        byte begin = data[0];     //开始符02
+        byte end = data[size - 1];//结束符03
+
+        //无开始符，只有结束符，数据丢包
+        if (begin != 0x02 && end == 0x03){
+            System.out.println("公众号：bugstack虫洞栈 提示；byteBuf数据，无开始符，只有结束符，数据丢包。");
+            channelHandlerContext.writeAndFlush("error");
+            return; //直接返回，不置位指针
+        }
+        //有开始符，无结束符号，数据半包。置位指针，接收余下数据
+        if (begin != 0x02 || end != 0x03) {
+            in.resetReaderIndex();
+            System.out.println("公众号：bugstack虫洞栈 提示；byteBuf数据，有开始符，无结束符号，数据半包。置位指针，接收余下数据。");
+            return;
+        }
+        //数据完整，解析处理
+        System.out.println(JSON.toJSONString(data));
+        //越过02 03位
+        ByteBuf copy = in.copy(1, size - 1);
+        //转换
+        String msg = copy.toString(Charset.forName("GBK"));
+        //填充
+        out.add(msg);*/
+    }
+
+}