公众号：bugstack虫洞栈 | SpringBoot+Netty+Elasticsearch收集日志信息数据存储

src/itstack-demo-netty/itstack-demo-netty-2-06/pom.xml

@@ -0,0 +1,81 @@
+<?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>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.0.1.RELEASE</version>
+        <relativePath/> <!-- lookup parent from repository -->
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>itstack-demo-netty-2-06</artifactId>
+
+    <properties>
+        <protostuff.version>1.0.10</protostuff.version>
+        <objenesis.version>2.4</objenesis.version>
+    </properties>
+
+    <dependencies>
+        <!-- Netty4.1 -->
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+            <version>4.1.36.Final</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework</groupId>
+                    <artifactId>spring-test</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <!-- fastjson -->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.58</version>
+        </dependency>
+        <!-- Protostuff -->
+        <dependency>
+            <groupId>com.dyuproject.protostuff</groupId>
+            <artifactId>protostuff-core</artifactId>
+            <version>${protostuff.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.dyuproject.protostuff</groupId>
+            <artifactId>protostuff-runtime</artifactId>
+            <version>${protostuff.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.objenesis</groupId>
+            <artifactId>objenesis</artifactId>
+            <version>${objenesis.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+
+</project>

src/itstack-demo-netty/itstack-demo-netty-2-06/src/main/java/org/itstack/demo/netty/Application.java

@@ -0,0 +1,45 @@
+package org.itstack.demo.netty;
+
+import io.netty.channel.ChannelFuture;
+import org.itstack.demo.netty.server.NettyServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+import javax.annotation.Resource;
+import java.net.InetSocketAddress;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛获取学习源码｝
+ * Create by fuzhengwei on 2019
+ */
+@SpringBootApplication
+public class Application implements CommandLineRunner {
+
+    private Logger logger = LoggerFactory.getLogger(Application.class);
+
+    @Value("${netty.host}")
+    private String host;
+    @Value("${netty.port}")
+    private int port;
+    @Resource
+    private NettyServer nettyServer;
+
+    public static void main(String[] args) {
+        System.setProperty("es.set.netty.runtime.available.processors", "false");
+        SpringApplication.run(Application.class, args);
+    }
+
+    @Override
+    public void run(String... args) throws Exception {
+        InetSocketAddress address = new InetSocketAddress(host, port);
+        ChannelFuture channelFuture = nettyServer.bing(address);
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> nettyServer.destroy()));
+        channelFuture.channel().closeFuture().syncUninterruptibly();
+    }
+
+}

src/itstack-demo-netty/itstack-demo-netty-2-06/src/main/java/org/itstack/demo/netty/codec/ObjDecoder.java

@@ -0,0 +1,40 @@
+package org.itstack.demo.netty.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import org.itstack.demo.netty.util.SerializationUtil;
+
+import java.util.List;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛关注获取学习源码｝
+ * 虫洞群：①群5398358 ②群5360692
+ * Create by fuzhengwei on 2019
+ */
+public class ObjDecoder extends ByteToMessageDecoder {
+
+    private Class<?> genericClass;
+
+    public ObjDecoder(Class<?> genericClass) {
+        this.genericClass = genericClass;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
+        if (in.readableBytes() < 4) {
+            return;
+        }
+        in.markReaderIndex();
+        int dataLength = in.readInt();
+        if (in.readableBytes() < dataLength) {
+            in.resetReaderIndex();
+            return;
+        }
+        byte[] data = new byte[dataLength];
+        in.readBytes(data);
+        out.add(SerializationUtil.deserialize(data, genericClass));
+    }
+
+}

src/itstack-demo-netty/itstack-demo-netty-2-06/src/main/java/org/itstack/demo/netty/codec/ObjEncoder.java

@@ -0,0 +1,31 @@
+package org.itstack.demo.netty.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+import org.itstack.demo.netty.util.SerializationUtil;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛关注获取学习源码｝
+ * 虫洞群：①群5398358 ②群5360692
+ * Create by fuzhengwei on 2019
+ */
+public class ObjEncoder extends MessageToByteEncoder {
+
+    private Class<?> genericClass;
+
+    public ObjEncoder(Class<?> genericClass) {
+        this.genericClass = genericClass;
+    }
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, Object in, ByteBuf out)  {
+        if (genericClass.isInstance(in)) {
+            byte[] data = SerializationUtil.serialize(in);
+            out.writeInt(data.length);
+            out.writeBytes(data);
+        }
+    }
+
+}

src/itstack-demo-netty/itstack-demo-netty-2-06/src/main/java/org/itstack/demo/netty/domain/TransportProtocol.java

@@ -0,0 +1,37 @@
+package org.itstack.demo.netty.domain;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛获取学习源码｝
+ * Create by fuzhengwei on 2019
+ */
+public class TransportProtocol {
+
+    private Integer type; //1用户信息
+    private Object obj;   //传输对象
+
+    public TransportProtocol() {
+    }
+
+    public TransportProtocol(Integer type, Object obj) {
+        this.type = type;
+        this.obj = obj;
+    }
+
+    public Integer getType() {
+        return type;
+    }
+
+    public void setType(Integer type) {
+        this.type = type;
+    }
+
+    public Object getObj() {
+        return obj;
+    }
+
+    public void setObj(Object obj) {
+        this.obj = obj;
+    }
+
+}

src/itstack-demo-netty/itstack-demo-netty-2-06/src/main/java/org/itstack/demo/netty/domain/User.java

@@ -0,0 +1,102 @@
+package org.itstack.demo.netty.domain;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.elasticsearch.annotations.Document;
+
+import java.util.Date;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛获取学习源码｝
+ * Create by fuzhengwei on 2019
+ */
+@Document(indexName = "stack", type = "group_user")
+public class User {
+
+    @Id
+    private String id;
+    private String name;   //姓名
+    private Integer age;   //年龄
+    private String level;  //级别
+    private Date entryDate;//时间
+    private String mobile; //电话
+    private String email;  //邮箱
+    private String address;//地址
+
+
+    public User(String id, String name, Integer age, String level, Date entryDate, String mobile, String email, String address) {
+        this.id = id;
+        this.name = name;
+        this.age = age;
+        this.level = level;
+        this.entryDate = entryDate;
+        this.mobile = mobile;
+        this.email = email;
+        this.address = address;
+
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public Integer getAge() {
+        return age;
+    }
+
+    public void setAge(Integer age) {
+        this.age = age;
+    }
+
+    public String getLevel() {
+        return level;
+    }
+
+    public void setLevel(String level) {
+        this.level = level;
+    }
+
+    public Date getEntryDate() {
+        return entryDate;
+    }
+
+    public void setEntryDate(Date entryDate) {
+        this.entryDate = entryDate;
+    }
+
+    public String getMobile() {
+        return mobile;
+    }
+
+    public void setMobile(String mobile) {
+        this.mobile = mobile;
+    }
+
+    public String getEmail() {
+        return email;
+    }
+
+    public void setEmail(String email) {
+        this.email = email;
+    }
+
+    public String getAddress() {
+        return address;
+    }
+
+    public void setAddress(String address) {
+        this.address = address;
+    }
+}

src/itstack-demo-netty/itstack-demo-netty-2-06/src/main/java/org/itstack/demo/netty/server/MyChannelInitializer.java

@@ -0,0 +1,36 @@
+package org.itstack.demo.netty.server;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.LineBasedFrameDecoder;
+import io.netty.handler.codec.string.StringDecoder;
+import io.netty.handler.codec.string.StringEncoder;
+import org.itstack.demo.netty.codec.ObjDecoder;
+import org.itstack.demo.netty.codec.ObjEncoder;
+import org.itstack.demo.netty.domain.TransportProtocol;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.nio.charset.Charset;
+
+/**
+ * 虫洞栈：https://bugstack.cn
+ * 公众号：bugstack虫洞栈  ｛获取学习源码｝
+ * Create by fuzhengwei on 2019
+ */
+@Service("myChannelInitializer")
+public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {
+
+    @Resource
+     private MyServerHandler myServerHandler;
+
+    @Override
+    protected void initChannel(SocketChannel channel) {
+        //对象传输处理
+        channel.pipeline().addLast(new ObjDecoder(TransportProtocol.class));
+        channel.pipeline().addLast(new ObjEncoder(TransportProtocol.class));
+        // 在管道中添加我们自己的接收数据实现方法
+        channel.pipeline().addLast(myServerHandler);
+    }
+
+}