Preamble Introduction
In a real development scenario, we want to store a large amount of business and user behavior data for analysis and processing, but because of the large amount of data and the need for analytic capabilities, it is more reasonable to store data in the file system.Especially for some Internet applications with high concurrency level, database design is usually based on database and table, so it is necessary to summarize these scattered data into a unified file system through binlog.
development environment
1. jdk1.8 [below jdk1.7 can only partially support netty]
2. Netty4.1.36.Final [netty3.x 4.x 5 changes a lot each time, so does the interface class name]
3,elasticsearch6.2.2
3.1,Installing elasticsearch 6.2.2 in windows
3.2,elasticsearch-head plug-in installation
Code Samples
itstack-demo-netty-2-06 └── src ├── main │ ├── java │ │ └── org.itstack.demo.netty │ │ ├── codec │ │ │ ├── ObjDecoder.java │ │ │ └── ObjEncoder.java │ │ ├── domain │ │ │ ├── TransportProtocol.java │ │ │ └── User.java │ │ ├── server │ │ │ ├── MyChannelInitializer.java │ │ │ ├── MyServerHandler.java │ │ │ └── NettyServer.java │ │ ├── service │ │ │ ├── impl │ │ │ │ └── UserServiceImpl.java │ │ │ ├── UserRepository.java │ │ │ └── UserService.java │ │ ├── util │ │ │ └── SerializationUtil.java │ │ ├── web │ │ │ └── NettyController.java │ │ └── Application.java │ └── resources │ └── application.yml │ └── test └── java └── org.itstack.demo.test └── ApiTest.java
Demonstrate some key code blocks, complete code download pays attention to public numbers; bugstack bug stack
domain/User.java
/** * Hole stack: https://bugstack.cn * Public Number: bugstack wormhole stack Get learning source * Create by fuzhengwei on 2019 */ @Document(indexName = "stack", type = "group_user") public class User { @Id private String id; private String name; //Full name private Integer age; //Age private String level; //level private Date entryDate;//time private String mobile; //Telephone private String email; //mailbox private String address;//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; } ... get/set }
server/MyServerHandler.java
/** * Hole stack: https://bugstack.cn * Public Number: bugstack wormhole stack Get learning source * Create by fuzhengwei on 2019 */ @Service("myServerHandler") public class MyServerHandler extends ChannelInboundHandlerAdapter { private Logger logger = LoggerFactory.getLogger(MyServerHandler.class); @Autowired private UserService userService; /** * This channel is active when the client actively links the links on the server side.That is, the client establishes a communication channel with the server and can transfer data. */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { SocketChannel channel = (SocketChannel) ctx.channel(); logger.info("Link Report Start"); logger.info("Link Report Information: A client is linked to this server"); logger.info("Link Report IP:{}", channel.localAddress().getHostString()); logger.info("Link Report Port:{}", channel.localAddress().getPort()); logger.info("Link report complete"); } /** * This channel is inactive when the client actively disconnects the server.That is, the client and the server have closed the communication channel and cannot transfer data. */ @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { logger.info("Client disconnects{}", ctx.channel().localAddress().toString()); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //Receive msg messages {You no longer need to decode yourself here compared to the previous chapter} logger.info(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " The server receives a message:" + JSON.toJSONString(msg)); //Receive data written to Elasticsearch TransportProtocol transportProtocol = (TransportProtocol) msg; userService.save((User) transportProtocol.getObj()); } /** * Catch exceptions, and when they occur, you can do something about them, such as printing logs, closing links */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); logger.info("Exception information:\r\n" + cause.getMessage()); } }
service/UserService.java * Provides CRUD methods
/** * Hole stack: https://bugstack.cn * Public Number: bugstack wormhole stack Get learning source * Create by fuzhengwei on 2019 */ public interface UserService { void save(User user); void deleteById(String id); User queryUserById(String id); Iterable<User> queryAll(); Page<User> findByName(String name, PageRequest request); }
service/UserRepository.java * can extend the methods needed, User is table, String is ID
/** * Hole stack: https://bugstack.cn * Public Number: bugstack wormhole stack Get learning source * Create by fuzhengwei on 2019 */ public interface UserRepository extends ElasticsearchRepository<User, String> { Page<User> findByName(String name, Pageable pageable); }
service/impl/UserServiceImpl.java *CRUD Implementation Class
/** * Hole stack: https://bugstack.cn * Public Number: bugstack wormhole stack Get learning source * Create by fuzhengwei on 2019 */ @Service("userService") public class UserServiceImpl implements UserService { private UserRepository dataRepository; @Autowired public void setDataRepository(UserRepository dataRepository) { this.dataRepository = dataRepository; } @Override public void save(User user) { dataRepository.save(user); } @Override public void deleteById(String id) { dataRepository.deleteById(id); } @Override public User queryUserById(String id) { Optional<User> optionalUser = dataRepository.findById(id); return optionalUser.get(); } @Override public Iterable<User> queryAll() { return dataRepository.findAll(); } @Override public Page<User> findByName(String name, PageRequest request) { return dataRepository.findByName(name, request); } }
Application.java *springboot starts Netty service at the same time
/** * Hole stack: https://bugstack.cn * Public Number: bugstack wormhole stack Get learning source * 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(); } }
application.properties * configuration file service port, netty, Elasticsearch
## Service Port server.port = 8080 ## Netty Server Side Configuration netty.host = 127.0.0.1 netty.port = 7397 ## Elasticsearch Configuration Change to your own cluster-name,cluster-nodes｝ spring.data.elasticsearch.cluster-name=es-itstack spring.data.elasticsearch.cluster-nodes=127.0.0.1:9300 spring.data.elasticsearch.repositories.enabled=true
ApiTest.java *Netty client for sending data to the server
/** * Hole stack: https://bugstack.cn * Public Number: bugstack wormhole stack Get learning source * Create by fuzhengwei on 2019 */ public class ApiTest { public static void main(String[] args) { EventLoopGroup workerGroup = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(workerGroup); b.channel(NioSocketChannel.class); b.option(ChannelOption.AUTO_READ, true); b.handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel channel) throws Exception { //Object Transfer Processing channel.pipeline().addLast(new ObjDecoder(TransportProtocol.class)); channel.pipeline().addLast(new ObjEncoder(TransportProtocol.class)); // Add our own implementation of receiving data to the pipeline channel.pipeline().addLast(new ChannelInboundHandlerAdapter() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { } }); } }); ChannelFuture f = b.connect("127.0.0.1", 7397).sync(); System.out.println("itstack-demo-netty client start done. {Focus on Public Number: bugstack Bug hole stack, get source code}"); TransportProtocol tp1 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Li Xiaoming", 1, "T0-1", new Date(), "13566668888", "184172133@qq.com", "Beijing")); TransportProtocol tp2 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Zhang Daming", 2, "T0-2", new Date(), "13566660001", "huahua@qq.com", "Nanjing")); TransportProtocol tp3 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Li Shupeng", 2, "T1-1", new Date(), "13566660002", "xiaobai@qq.com", "Elm")); TransportProtocol tp4 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Han Xiaoxue", 2, "T2-1", new Date(), "13566660002", "xiaobai@qq.com", "Elm")); TransportProtocol tp5 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Dong Shufei", 2, "T4-1", new Date(), "13566660002", "xiaobai@qq.com", "HEBEI")); TransportProtocol tp6 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Waiting for Ming Phase", 2, "T5-1", new Date(), "13566660002", "xiaobai@qq.com", "Down the Garden")); TransportProtocol tp7 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Tian Mingming", 2, "T3-1", new Date(), "13566660002", "xiaobai@qq.com", "Dongping")); TransportProtocol tp8 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "David Wong", 2, "T4-1", new Date(), "13566660002", "xiaobai@qq.com", "West Lake")); TransportProtocol tp9 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Li Xueming", 2, "T1-1", new Date(), "13566660002", "xiaobai@qq.com", "Nanchang")); TransportProtocol tp10 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Zhu Xiaofei", 2, "T2-1", new Date(), "13566660002", "xiaobai@qq.com", "Jilin")); TransportProtocol tp11 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Niu Daming", 2, "T1-1", new Date(), "13566660002", "xiaobai@qq.com", "Changchun")); TransportProtocol tp12 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "Guan Xuer", 2, "T2-1", new Date(), "13566660002", "xiaobai@qq.com", "Shenzhen")); //Send information to the server f.channel().writeAndFlush(tp1); f.channel().writeAndFlush(tp2); f.channel().writeAndFlush(tp3); f.channel().writeAndFlush(tp4); f.channel().writeAndFlush(tp5); f.channel().writeAndFlush(tp6); f.channel().writeAndFlush(tp7); f.channel().writeAndFlush(tp8); f.channel().writeAndFlush(tp9); f.channel().writeAndFlush(tp10); f.channel().writeAndFlush(tp11); f.channel().writeAndFlush(tp12); f.channel().closeFuture().syncUninterruptibly(); } catch (InterruptedException e) { e.printStackTrace(); } finally { workerGroup.shutdownGracefully(); } } }
test result
Start Elasticsearch * or double-click directly...elasticsearch-6.2.2/bin/elasticsearch.bat
D:\Program Files\elasticsearch\node01\elasticsearch-6.2.2\bin>elasticsearch.bat [2019-08-10T14:09:26,562][INFO ][o.e.n.Node ] [node01] initializin g ... [2019-08-10T14:09:26,770][INFO ][o.e.e.NodeEnvironment ] [node01] using [1] d ata paths, mounts [[QiJoy (D:)]], net usable_space [301.3gb], net total_space [ 407.1gb], types [NTFS] [2019-08-10T14:09:26,771][INFO ][o.e.e.NodeEnvironment ] [node01] heap size [ 990.7mb], compressed ordinary object pointers [true] [2019-08-10T14:09:26,843][INFO ][o.e.n.Node ] [node01] node name [ node01], node ID [R5wRCDr0SSKsVsgkZwB-Hg] [2019-08-10T14:09:26,843][INFO ][o.e.n.Node ] [node01] version[6.2 .2], pid[22264], build[10b1edd/2018-02-16T19:01:30.685723Z], OS[Windows 7/6.1/am d64], JVM[Oracle Corporation/Java HotSpot(TM) 64-Bit Server VM/1.8.0_45/25.45-b0 2] [2019-08-10T14:09:26,843][INFO ][o.e.n.Node ] [node01] JVM argumen ts [-Xms1g, -Xmx1g, -XX:+UseConcMarkSweepGC, -XX:CMSInitiatingOccupancyFraction= 75, -XX:+UseCMSInitiatingOccupancyOnly, -XX:+AlwaysPreTouch, -Xss1m, -Djava.awt. headless=true, -Dfile.encoding=UTF-8, -Djna.nosys=true, -XX:-OmitStackTraceInFas tThrow, -Dio.netty.noUnsafe=true, -Dio.netty.noKeySetOptimization=true, -Dio.net
Start Elasticsearch-head
D:\Program Files\elasticsearch\head>npm run start > elasticsearch-head@0.0.0 start D:\Program Files\elasticsearch\head > grunt server Running "connect:server" (connect) task Waiting forever... Started connect web server on http://localhost:9100
Starting the StringBoot *Netty service will start with {Application.main}
. ____ _ __ _ _ /\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \ ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \ \\/ ___)| |_)| | | | | || (_| | ) ) ) ) ' |____| .__|_| |_|_| |_\__, | / / / / =========|_|==============|___/=/_/_/_/ :: Spring Boot :: (v2.0.1.RELEASE) 2019-08-10 14:14:49.619 INFO 5976 --- [ main] org.itstack.demo.netty.Application : Starting Application on JRA1W11T0247 with PID 5976 (E:\itstack\GIT\itstack.org\itstack-demo-netty\itstack-demo-netty-2-06\target\classes started by fuzhengwei1 in E:\itstack\GIT\itstack.org\itstack-demo-netty) 2019-08-10 14:14:49.622 INFO 5976 --- [ main] org.itstack.demo.netty.Application : No active profile set, falling back to default profiles: default 2019-08-10 14:14:49.704 INFO 5976 --- [ main] ConfigServletWebServerApplicationContext : Refreshing org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@7f010382: startup date [Sat Aug 10 14:14:49 CST 2019]; root of context hierarchy 2019-08-10 14:14:51.341 INFO 5976 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat initialized with port(s): 8080 (http) 2019-08-10 14:14:51.367 INFO 5976 --- [ main] o.apache.catalina.core.StandardService : Starting service [Tomcat] 2019-08-10 14:14:51.367 INFO 5976 --- [ main] org.apache.catalina.core.StandardEngine : Starting Servlet Engine: Apache Tomcat/8.5.29 2019-08-10 14:14:51.377 INFO 5976 --- [ost-startStop-1] o.a.catalina.core.AprLifecycleListener : The APR based Apache Tomcat Native library which allows optimal performance in production environments was not found on the java.library.path: [C:\Program Files\Java\jdk1.8.0_45\bin;C:\windows\Sun\Java\bin;C:\windows\system32;C:\windows;C:\ProgramData\Oracle\Java\javapath;C:\Program Files (x86)\Common Files\NetSarang;C:\Program Files (x86)\Intel\iCLS Client\;C:\Program Files\Intel\iCLS Client\;C:\windows\system32;C:\windows;C:\windows\System32\Wbem;C:\windows\System32\WindowsPowerShell\v1.0\;C:\Program Files\Intel\Intel(R) Management Engine Components\DAL;C:\Program Files\Intel\Intel(R) Management Engine Components\IPT;C:\Program Files (x86)\Intel\Intel(R) Management Engine Components\DAL;C:\Program Files (x86)\Intel\Intel(R) Management Engine Components\IPT;C:\Program Files\Java\jdk1.8.0_45/bin;C:\Program Files\Java\jdk1.8.0_45/jre/bin;D:\Program Files\SlikSvn\bin;D:\Program Files\TortoiseSVN\bin;D:\Program Files (x86)\apache-maven-2.2.1\bin;D:\Program Files\TortoiseGit\bin;D:\Program Files\nodejs\;D:\Program Files (x86)\SSH Communications Security\SSH Secure Shell;C:\Users\fuzhengwei1\AppData\Roaming\npm;;.] 2019-08-10 14:14:51.523 INFO 5976 --- [ost-startStop-1] o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring embedded WebApplicationContext 2019-08-10 14:14:51.523 INFO 5976 --- [ost-startStop-1] o.s.web.context.ContextLoader : Root WebApplicationContext: initialization completed in 1819 ms 2019-08-10 14:14:51.659 INFO 5976 --- [ost-startStop-1] o.s.b.w.servlet.ServletRegistrationBean : Servlet dispatcherServlet mapped to [/] 2019-08-10 14:14:51.663 INFO 5976 --- [ost-startStop-1] o.s.b.w.servlet.FilterRegistrationBean : Mapping filter: 'characterEncodingFilter' to: [/*] 2019-08-10 14:14:51.664 INFO 5976 --- [ost-startStop-1] o.s.b.w.servlet.FilterRegistrationBean : Mapping filter: 'hiddenHttpMethodFilter' to: [/*] 2019-08-10 14:14:51.664 INFO 5976 --- [ost-startStop-1] o.s.b.w.servlet.FilterRegistrationBean : Mapping filter: 'httpPutFormContentFilter' to: [/*] 2019-08-10 14:14:51.664 INFO 5976 --- [ost-startStop-1] o.s.b.w.servlet.FilterRegistrationBean : Mapping filter: 'requestContextFilter' to: [/*] 2019-08-10 14:14:52.090 INFO 5976 --- [ main] o.elasticsearch.plugins.PluginsService : no modules loaded 2019-08-10 14:14:52.092 INFO 5976 --- [ main] o.elasticsearch.plugins.PluginsService : loaded plugin [org.elasticsearch.index.reindex.ReindexPlugin] 2019-08-10 14:14:52.092 INFO 5976 --- [ main] o.elasticsearch.plugins.PluginsService : loaded plugin [org.elasticsearch.join.ParentJoinPlugin] 2019-08-10 14:14:52.092 INFO 5976 --- [ main] o.elasticsearch.plugins.PluginsService : loaded plugin [org.elasticsearch.percolator.PercolatorPlugin] 2019-08-10 14:14:52.092 INFO 5976 --- [ main] o.elasticsearch.plugins.PluginsService : loaded plugin [org.elasticsearch.script.mustache.MustachePlugin] 2019-08-10 14:14:52.092 INFO 5976 --- [ main] o.elasticsearch.plugins.PluginsService : loaded plugin [org.elasticsearch.transport.Netty3Plugin] 2019-08-10 14:14:52.092 INFO 5976 --- [ main] o.elasticsearch.plugins.PluginsService : loaded plugin [org.elasticsearch.transport.Netty4Plugin] 2019-08-10 14:14:52.973 INFO 5976 --- [ main] o.s.d.e.c.TransportClientFactoryBean : adding transport node : 127.0.0.1:9300 2019-08-10 14:14:54.486 INFO 5976 --- [ main] o.s.w.s.handler.SimpleUrlHandlerMapping : Mapped URL path [/**/favicon.ico] onto handler of type [class org.springframework.web.servlet.resource.ResourceHttpRequestHandler] 2019-08-10 14:14:54.730 INFO 5976 --- [ main] s.w.s.m.m.a.RequestMappingHandlerAdapter : Looking for @ControllerAdvice: org.springframework.boot.web.servlet.context.AnnotationConfigServletWebServerApplicationContext@7f010382: startup date [Sat Aug 10 14:14:49 CST 2019]; root of context hierarchy 2019-08-10 14:14:54.800 INFO 5976 --- [ main] s.w.s.m.m.a.RequestMappingHandlerMapping : Mapped "{[/localAddress]}" onto public java.lang.String org.itstack.demo.netty.web.NettyController.localAddress() 2019-08-10 14:14:54.803 INFO 5976 --- [ main] s.w.s.m.m.a.RequestMappingHandlerMapping : Mapped "{[/error]}" onto public org.springframework.http.ResponseEntity<java.util.Map<java.lang.String, java.lang.Object>> org.springframework.boot.autoconfigure.web.servlet.error.BasicErrorController.error(javax.servlet.http.HttpServletRequest) 2019-08-10 14:14:54.804 INFO 5976 --- [ main] s.w.s.m.m.a.RequestMappingHandlerMapping : Mapped "{[/error],produces=[text/html]}" onto public org.springframework.web.servlet.ModelAndView org.springframework.boot.autoconfigure.web.servlet.error.BasicErrorController.errorHtml(javax.servlet.http.HttpServletRequest,javax.servlet.http.HttpServletResponse) 2019-08-10 14:14:54.822 INFO 5976 --- [ main] o.s.w.s.handler.SimpleUrlHandlerMapping : Mapped URL path [/webjars/**] onto handler of type [class org.springframework.web.servlet.resource.ResourceHttpRequestHandler] 2019-08-10 14:14:54.822 INFO 5976 --- [ main] o.s.w.s.handler.SimpleUrlHandlerMapping : Mapped URL path [/**] onto handler of type [class org.springframework.web.servlet.resource.ResourceHttpRequestHandler] 2019-08-10 14:14:54.985 INFO 5976 --- [ main] o.s.j.e.a.AnnotationMBeanExporter : Registering beans for JMX exposure on startup 2019-08-10 14:14:55.013 INFO 5976 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port(s): 8080 (http) with context path '' 2019-08-10 14:14:55.016 INFO 5976 --- [ main] org.itstack.demo.netty.Application : Started Application in 5.982 seconds (JVM running for 6.516) 2019-08-10 14:14:55.043 INFO 5976 --- [ main] o.itstack.demo.netty.server.NettyServer : itstack-demo-netty server start done. {Focus on Public Number: bugstack Bug hole stack, get source code}
Launch Netty client to send data ApiTest.main
itstack-demo-netty client start done. {Focus on Public Number: bugstack Bug hole stack, get source code} 14:16:33.543 [main] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.maxCapacityPerThread: 4096 14:16:33.543 [main] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.maxSharedCapacityFactor: 2 14:16:33.543 [main] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.linkCapacity: 16 14:16:33.543 [main] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.ratio: 8 14:16:33.555 [nioEventLoopGroup-2-1] DEBUG io.netty.buffer.AbstractByteBuf - -Dio.netty.buffer.checkAccessible: true 14:16:33.555 [nioEventLoopGroup-2-1] DEBUG io.netty.buffer.AbstractByteBuf - -Dio.netty.buffer.checkBounds: true 14:16:33.556 [nioEventLoopGroup-2-1] DEBUG io.netty.util.ResourceLeakDetectorFactory - Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@529fc511
Execution Result* Data has been written to Elasticsearch
WeChat Public Number: bugstack wormhole stack, welcome your attention & get the source code!
Netty Development Group (5360692):itstack | Netty the sniper
