netty之SpringBoot+Netty+Elasticsearch网络日志信息数据存储

天津储鑫盛钢材现货供应商  金牌会员 | 2024-10-11 17:53:29 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 807|帖子 807|积分 2421

#开发环境

环境准备
windows安装包 下载
注意 es是以来java环境 以是需要安装jdk 支持1.7以上
es-hander下载可视化操作插件
  1. @Document(indexName = "stack", type = "group_user")
  2. public class User {
  4.     @Id
  5.     private String id;
  6.     private String name;   //姓名
  7.     private Integer age;   //年龄
  8.     private String level;  //级别
  9.     private Date entryDate;//时间
  10.     private String mobile; //电话
  11.     private String email;  //邮箱
  12.     private String address;//地址
  15.     public User(String id, String name, Integer age, String level, Date entryDate, String mobile, String email, String address) {
  16.         this.id = id;
  17.         this.name = name;
  18.         this.age = age;
  19.         this.level = level;
  20.         this.entryDate = entryDate;
  21.         this.mobile = mobile;
  22.         this.email = email;
  23.         this.address = address;
  25.     }
  27.     public String getId() {
  28.         return id;
  29.     }
  31.     public void setId(String id) {
  32.         this.id = id;
  33.     }
  35.     public String getName() {
  36.         return name;
  37.     }
  39.     public void setName(String name) {
  40.         this.name = name;
  41.     }
  43.     public Integer getAge() {
  44.         return age;
  45.     }
  47.     public void setAge(Integer age) {
  48.         this.age = age;
  49.     }
  51.     public String getLevel() {
  52.         return level;
  53.     }
  55.     public void setLevel(String level) {
  56.         this.level = level;
  57.     }
  59.     public Date getEntryDate() {
  60.         return entryDate;
  61.     }
  63.     public void setEntryDate(Date entryDate) {
  64.         this.entryDate = entryDate;
  65.     }
  67.     public String getMobile() {
  68.         return mobile;
  69.     }
  71.     public void setMobile(String mobile) {
  72.         this.mobile = mobile;
  73.     }
  75.     public String getEmail() {
  76.         return email;
  77.     }
  79.     public void setEmail(String email) {
  80.         this.email = email;
  81.     }
  83.     public String getAddress() {
  84.         return address;
  85.     }
  87.     public void setAddress(String address) {
  88.         this.address = address;
  89.     }
  90. }
复制代码
  1. @Service("myServerHandler")
  2. public class MyServerHandler extends ChannelInboundHandlerAdapter {
  4.     private Logger logger = LoggerFactory.getLogger(MyServerHandler.class);
  6.     @Autowired
  7.     private UserService userService;
  9.     /**
  10.      * 当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据
  11.      */
  12.     @Override
  13.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  14.         SocketChannel channel = (SocketChannel) ctx.channel();
  15.         logger.info("链接报告开始");
  16.         logger.info("链接报告信息：有一客户端链接到本服务端");
  17.         logger.info("链接报告IP:{}", channel.localAddress().getHostString());
  18.         logger.info("链接报告Port:{}", channel.localAddress().getPort());
  19.         logger.info("链接报告完毕");
  20.     }
  22.     /**
  23.      * 当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端的关闭了通信通道并且不可以传输数据
  24.      */
  25.     @Override
  26.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  27.         logger.info("客户端断开链接{}", ctx.channel().localAddress().toString());
  28.     }
  30.     @Override
  31.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  32.         //接收msg消息{与上一章节相比，此处已经不需要自己进行解码}
  33.         logger.info(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 服务端接收到消息：" + JSON.toJSONString(msg));
  34.         //接收数据写入到Elasticsearch
  35.         TransportProtocol transportProtocol = (TransportProtocol) msg;
  36.         userService.save((User) transportProtocol.getObj());
  37.     }
  39.     /**
  40.      * 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接
  41.      */
  42.     @Override
  43.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  44.         ctx.close();
  45.         logger.info("异常信息：\r\n" + cause.getMessage());
  46.     }
  48. }
复制代码
  1. @Component("nettyServer")
  2. public class NettyServer {
  4.     private Logger logger = LoggerFactory.getLogger(NettyServer.class);
  6.     @Resource
  7.     private MyChannelInitializer myChannelInitializer;
  9.     //配置服务端NIO线程组
  10.     private final EventLoopGroup parentGroup = new NioEventLoopGroup(); //NioEventLoopGroup extends MultithreadEventLoopGroup Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
  11.     private final EventLoopGroup childGroup = new NioEventLoopGroup();
  12.     private Channel channel;
  14.     public ChannelFuture bing(InetSocketAddress address) {
  15.         ChannelFuture channelFuture = null;
  16.         try {
  17.             ServerBootstrap b = new ServerBootstrap();
  18.             b.group(parentGroup, childGroup)
  19.                     .channel(NioServerSocketChannel.class)    //非阻塞模式
  20.                     .option(ChannelOption.SO_BACKLOG, 128)
  21.                     .childHandler(myChannelInitializer);
  23.             channelFuture = b.bind(address).syncUninterruptibly();
  24.             channel = channelFuture.channel();
  25.         } catch (Exception e) {
  26.             logger.error(e.getMessage());
  27.         } finally {
  28.             if (null != channelFuture && channelFuture.isSuccess()) {
  29.                 logger.info("itstack-demo-netty server start done. {关注明哥，获取源码}");
  30.             } else {
  31.                 logger.error("itstack-demo-netty server start error. {关注明哥，获取源码}");
  32.             }
  33.         }
  34.         return channelFuture;
  35.     }
  37.     public void destroy() {
  38.         if (null == channel) return;
  39.         channel.close();
  40.         parentGroup.shutdownGracefully();
  41.         childGroup.shutdownGracefully();
  42.     }
  44.     public Channel getChannel() {
  45.         return channel;
  46.     }
  48. }
复制代码
  2. public interface UserService {
  4.     void save(User user);
  6.     void deleteById(String id);
  8.     User queryUserById(String id);
  10.     Iterable<User> queryAll();
  12.     Page<User> findByName(String name, PageRequest request);
  14. }
复制代码
提供一个可拓展的操作实体表的接口
  2. public interface UserRepository extends ElasticsearchRepository<User, String> {
  4.     Page<User> findByName(String name, Pageable pageable);
  6. }
复制代码
  1. @Service("userService")
  2. public class UserServiceImpl implements UserService {
  4.     private UserRepository dataRepository;
  6.     @Autowired
  7.     public void setDataRepository(UserRepository dataRepository) {
  8.         this.dataRepository = dataRepository;
  9.     }
  11.     @Override
  12.     public void save(User user) {
  13.         dataRepository.save(user);
  14.     }
  16.     @Override
  17.     public void deleteById(String id) {
  18.         dataRepository.deleteById(id);
  19.     }
  21.     @Override
  22.     public User queryUserById(String id) {
  23.         Optional<User> optionalUser = dataRepository.findById(id);
  24.         return optionalUser.get();
  25.     }
  27.     @Override
  28.     public Iterable<User> queryAll() {
  29.         return dataRepository.findAll();
  30.     }
  32.     @Override
  33.     public Page<User> findByName(String name, PageRequest request) {
  34.         return dataRepository.findByName(name, request);
  35.     }
  37. }
复制代码
  1. @RestController
  2. public class NettyController {
  4.     @Resource
  5.     private NettyServer nettyServer;
  7.     @RequestMapping("/localAddress")
  8.     public String localAddress() {
  9.         return "nettyServer localAddress " + nettyServer.getChannel().localAddress();
  10.     }
  12. }
复制代码
  1. @SpringBootApplication
  2. public class Application implements CommandLineRunner {
  4.     private Logger logger = LoggerFactory.getLogger(Application.class);
  6.     @Value("${netty.host}")
  7.     private String host;
  8.     @Value("${netty.port}")
  9.     private int port;
  10.     @Resource
  11.     private NettyServer nettyServer;
  13.     public static void main(String[] args) {
  14.         System.setProperty("es.set.netty.runtime.available.processors", "false");
  15.         SpringApplication.run(Application.class, args);
  16.     }
  18.     @Override
  19.     public void run(String... args) throws Exception {
  20.         InetSocketAddress address = new InetSocketAddress(host, port);
  21.         ChannelFuture channelFuture = nettyServer.bing(address);
  22.         Runtime.getRuntime().addShutdownHook(new Thread(() -> nettyServer.destroy()));
  23.         channelFuture.channel().closeFuture().syncUninterruptibly();
  24.     }
  26. }
复制代码
  1. ## 服务端口
  2. server.port = 8080
  4. ## Netty服务端配置
  5. netty.host = 127.0.0.1
  6. netty.port = 7397
  8. ## Elasticsearch配置｛更换为自己的cluster-name、cluster-nodes｝
  9. spring.data.elasticsearch.cluster-name=es-itstack
  10. spring.data.elasticsearch.cluster-nodes=127.0.0.1:9300
  11. spring.data.elasticsearch.repositories.enabled=true
复制代码
ApiTest.java *Netty客户端，用于向服务端发送数据
  1. public class ApiTest {
  3.     public static void main(String[] args) {
  4.         System.out.println("hi 微信公众号：关注明哥");
  5.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  6.         try {
  7.             Bootstrap b = new Bootstrap();
  8.             b.group(workerGroup);
  9.             b.channel(NioSocketChannel.class);
  10.             b.option(ChannelOption.AUTO_READ, true);
  11.             b.handler(new ChannelInitializer<SocketChannel>() {
  12.                 @Override
  13.                 protected void initChannel(SocketChannel channel) throws Exception {
  14.                     //对象传输处理
  15.                     channel.pipeline().addLast(new ObjDecoder(TransportProtocol.class));
  16.                     channel.pipeline().addLast(new ObjEncoder(TransportProtocol.class));
  17.                     // 在管道中添加我们自己的接收数据实现方法
  18.                     channel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  19.                         @Override
  20.                         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  22.                         }
  23.                     });
  24.                 }
  25.             });
  26.             ChannelFuture f = b.connect("127.0.0.1", 7397).sync();
  27.             System.out.println("itstack-demo-netty client start done. {关注明哥，获取源码}");
  29.             TransportProtocol tp1 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "李小明", 1, "T0-1", new Date(), "13566668888", "184172133@qq.com", "北京"));
  30.             TransportProtocol tp2 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "张大明", 2, "T0-2", new Date(), "13566660001", "huahua@qq.com", "南京"));
  31.             TransportProtocol tp3 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "李书鹏", 2, "T1-1", new Date(), "13566660002", "xiaobai@qq.com", "榆树"));
  32.             TransportProtocol tp4 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "韩小雪", 2, "T2-1", new Date(), "13566660002", "xiaobai@qq.com", "榆树"));
  33.             TransportProtocol tp5 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "董叔飞", 2, "T4-1", new Date(), "13566660002", "xiaobai@qq.com", "河北"));
  34.             TransportProtocol tp6 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "候明相", 2, "T5-1", new Date(), "13566660002", "xiaobai@qq.com", "下花园"));
  35.             TransportProtocol tp7 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "田明明", 2, "T3-1", new Date(), "13566660002", "xiaobai@qq.com", "东平"));
  36.             TransportProtocol tp8 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "王大伟", 2, "T4-1", new Date(), "13566660002", "xiaobai@qq.com", "西湖"));
  37.             TransportProtocol tp9 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "李雪明", 2, "T1-1", new Date(), "13566660002", "xiaobai@qq.com", "南昌"));
  38.             TransportProtocol tp10 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "朱小飞", 2, "T2-1", new Date(), "13566660002", "xiaobai@qq.com", "吉林"));
  39.             TransportProtocol tp11 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "牛大明", 2, "T1-1", new Date(), "13566660002", "xiaobai@qq.com", "长春"));
  40.             TransportProtocol tp12 = new TransportProtocol(1, new User(UUID.randomUUID().toString(), "关雪儿", 2, "T2-1", new Date(), "13566660002", "xiaobai@qq.com", "深圳"));
  42.             //向服务端发送信息
  43.             f.channel().writeAndFlush(tp1);
  44.             f.channel().writeAndFlush(tp2);
  45.             f.channel().writeAndFlush(tp3);
  46.             f.channel().writeAndFlush(tp4);
  47.             f.channel().writeAndFlush(tp5);
  48.             f.channel().writeAndFlush(tp6);
  49.             f.channel().writeAndFlush(tp7);
  50.             f.channel().writeAndFlush(tp8);
  51.             f.channel().writeAndFlush(tp9);
  52.             f.channel().writeAndFlush(tp10);
  53.             f.channel().writeAndFlush(tp11);
  54.             f.channel().writeAndFlush(tp12);
  56.             f.channel().closeFuture().syncUninterruptibly();
  57.         } catch (InterruptedException e) {
  58.             e.printStackTrace();
  59.         } finally {
  60.             workerGroup.shutdownGracefully();
  61.         }
  62.     }
  64. }
复制代码
好了到这里就结束了netty之SpringBoot+Netty+Elasticsearch网络日志信息数据存储的学习，各人一定要跟着动手操作起来。需要的源码的 可si我获取；

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

正序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

天津储鑫盛钢材现货供应商

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表