devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

netty+springboot+dubbo+zk整合tcp项目_springboot 整合dubbo+netty

作者：我家小花儿 | 2024-07-20 03:01:39

赞

踩

springboot 整合dubbo+netty

1.描述

     为了能快速开发netty集成springboot整合tcp项目，本篇博客搭建勒一个空的架子，只需要修改对应的业务逻辑代码就行，很好的提高勒开发速度，降低勒netty开发的难度。开发人员，只需要关注于业务逻辑就行。

      建议：有一定的netty和springboot，dubbbo基础。

     下载socket调试工具

给一个测试的demo,十六进制字符串

2454585858007A0E320460047A3F00000330473336303020646D67642000000BE7130C040F0A00000000002D00390021004D0024004B0026005200300090060026005309000000000000000000000000000004AA04AA0A04A80700484A05470700278227830427820A000057B800559E0050C50500840D0A0000

2.netty

     2.1  服务器端   

              2.1.1 pom.xml 

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3. 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  4. 	<modelVersion>4.0.0</modelVersion>
  5.  
  6. 	<groupId>com.cloudtech</groupId>
  7. 	<artifactId>dipper</artifactId>
  8. 	<version>0.0.1-SNAPSHOT</version>
  9. 	<packaging>jar</packaging>
  10.  
  11. 	<name>dipper</name>
  12. 	<description>Demo project for Spring Boot</description>
  13.  
  14. 	<parent>
  15. 		<groupId>org.springframework.boot</groupId>
  16. 		<artifactId>spring-boot-starter-parent</artifactId>
  17. 		<version>2.0.5.RELEASE</version>
  18. 		<relativePath /> <!-- lookup parent from repository -->
  19. 	</parent>
  20.  
  21. 	<properties>
  22. 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  23. 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  24. 		<java.version>1.8</java.version>
  25. 	</properties>
  26.  
  27. 	<dependencies>
  28. 		<dependency>
  29. 			<groupId>org.springframework.boot</groupId>
  30. 			<artifactId>spring-boot-starter-web</artifactId>
  31. 		</dependency>
  32.  
  33. 		<dependency>
  34. 			<groupId>org.springframework.boot</groupId>
  35. 			<artifactId>spring-boot-starter-test</artifactId>
  36. 			<scope>test</scope>
  37. 		</dependency>
  38.  
  39. 		<!-- https://mvnrepository.com/artifact/com.alibaba.boot/dubbo-spring-boot-starter -->
  40. 		<dependency>
  41. 			<groupId>com.alibaba.boot</groupId>
  42. 			<artifactId>dubbo-spring-boot-starter</artifactId>
  43. 			<version>0.2.0</version>
  44. 		</dependency>
  45.  
  46.  
  47. 		<!-- netty -->
  48. 		<dependency>
  49. 			<groupId>io.netty</groupId>
  50. 			<artifactId>netty-all</artifactId>
  51. 			<version>5.0.0.Alpha2</version>
  52. 		</dependency>
  53.  
  54. 		<!-- 配置提示 -->
  55. 		<dependency>
  56. 			<groupId>org.springframework.boot</groupId>
  57. 			<artifactId>spring-boot-configuration-processor</artifactId>
  58. 			<optional>true</optional>
  59. 		</dependency>
  60. 	</dependencies>
  61.  
  62. 	<build>
  63. 	   <!-- 打成jar包的名称 -->
  64.   		<finalName>dipper</finalName>
  65. 		<plugins>
  66. 			<!-- 解决web.xml报错的问题 -->
  67. 			<plugin>
  68. 				<groupId>org.apache.maven.plugins</groupId>
  69. 				<artifactId>maven-war-plugin</artifactId>
  70. 				<configuration>
  71. 					<failOnMissingWebXml>false</failOnMissingWebXml>
  72. 				</configuration>
  73. 			</plugin>
  74.  
  75. 			<plugin>
  76. 				<groupId>org.springframework.boot</groupId>
  77. 				<artifactId>spring-boot-maven-plugin </artifactId>
  78. 			</plugin>
  79. 			<plugin>
  80. 				<groupId>org.springframework.boot</groupId>
  81. 				<artifactId>spring-boot-maven-plugin</artifactId>
  82. 				<executions>
  83. 					<execution>
  84. 						<goals>
  85. 							<goal>repackage</goal>
  86. 						</goals>
  87. 					</execution>
  88. 				</executions>
  89. 			</plugin>
  90. 		</plugins>
  91. 	</build>
  92. </project>

2.1.2  application.yml

  1. ####自定义配置           
  2. netty:
  3.   port: 4002    ###分钟监听端口
  4.   module-type: 0    ####0标识主通讯模块   1标识从通讯模块
  5.   
  6.  
  7. stationTypeId: 17
  8. ###最大缓存长度
  9. maxBuffLen: 10240  
  10.   
  11. logging:
  12.   config: classpath:logback-boot.xml
  13. ####dubbot版本
  14. demo:
  15.   service:
  16.    version:  1.0.0
  17.  
  18. dubbo:
  19.   reference:
  20.     check: false
  21.     com:
  22.      cloudtech:
  23.        web:
  24.          dubbo:
  25.            check: false
  26.   consumers:
  27.     check: false
  28.   scan:
  29.    basePackages: com.cloudtech.web.dubbo     
  30.   application:
  31.     id: dipper-provider-demo 
  32.     name: dipper-provider-demo
  33.   provider:
  34.     retries: -1
  35.     timeout: 30000000   
  36.   protocol:
  37.     id: dubbo 
  38.     name: dubbo
  39.     port: -1
  40.   registry:
  41.     protocol: zookeeper
  42.     check:  false
  43.     address: zookeeper://127.0.0.1:2181

2.1.3 启动类

  1. package com.cloudtech.demo;
  2.  
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.beans.factory.annotation.Value;
  7. import org.springframework.boot.CommandLineRunner;
  8. import org.springframework.boot.WebApplicationType;
  9. import org.springframework.boot.autoconfigure.SpringBootApplication;
  10. import org.springframework.boot.builder.SpringApplicationBuilder;
  11. import org.springframework.context.annotation.ComponentScan;
  12.  
  13. import com.cloudtech.config.NettyConfig;
  14. import com.cloudtech.server.Server;
  15. import com.cloudtech.util.Consts;
  16.  
  17. @SpringBootApplication
  18. @ComponentScan(value = {"com.cloudtech"})
  19. public class DipperApplication implements CommandLineRunner {
  20. 	protected static final Logger LOGGER = LoggerFactory.getLogger(DipperApplication.class);
  21. 	
  22. 	@Autowired
  23. 	NettyConfig nettyConfig;
  24. 	
  25. 	@Autowired
  26. 	Server server;
  27. 	
  28. 	@Value("${stationTypeId}")
  29. 	private Integer stationTypeId;
  30. 	@Value("${maxBuffLen}")
  31. 	private Integer maxBuffLen;
  32. 	
  33. 	public static void main(String[] args) {
  34. 		new SpringApplicationBuilder(DipperApplication.class)
  35. 	    .web(WebApplicationType.NONE)
  36. 	    .run(args);
  37. 		//SpringApplication.run(Aws310Application.class, args);
  38. 	}
  39.     
  40. 	/**
  41. 	 * spring boot启动后，会进入该方法
  42. 	 */
  43. 	@Override
  44. 	public void run(String... args) throws Exception {
  45. 		init();
  46. 		//加载分钟监听线程
  47. 		new Thread("Dipper-recevice-thread") {
  48. 			@Override
  49. 			public void run() {
  50. 				super.run();
  51. 				server.start(Consts.SERVER_PORT);
  52. 			}
  53. 		}.start();
  54. 	}
  55.  
  56. 	/**
  57. 	 * 初始化信息
  58. 	 */
  59. 	@SuppressWarnings("unused")
  60. 	private void init() {
  61. 		Consts.SERVER_PORT = nettyConfig.getPort();
  62. 		Consts.MODULE_TYPE = nettyConfig.getModuleType();
  63. 		Consts.STATION_TYPE_ID = stationTypeId;
  64. 		Consts.MAX_BUFF_LEN = maxBuffLen;
  65. 	}
  66.  
  67. 	
  68. }
  1. package com.cloudtech.config;
  2. import org.springframework.beans.factory.annotation.Value;
  3. import org.springframework.context.annotation.PropertySource;
  4. import org.springframework.stereotype.Component;
  5.  
  6. /**
  7.  * 读取yml配置文件中的信息
  8. * @ClassName: NettyConfig  
  9. * @Description:   
  10. * @author wude  
  11. * @date 2018年9月19日  
  12. *
  13.  */
  14. @Component
  15. @PropertySource(value={"classpath:application.yml"})
  16. public class NettyConfig {
  17.     @Value("${netty.port}")
  18.     private int port;
  19.     @Value("${netty.module-type}")
  20.     private int moduleType;
  21.  
  22.     public int getPort() {
  23.         return port;
  24.     }
  25.  
  26.     public void setPort(int port) {
  27.         this.port = port;
  28.     }
  29.  
  30. 	public int getModuleType() {
  31. 		return moduleType;
  32. 	}
  33.  
  34. 	public void setModuleType(int moduleType) {
  35. 		this.moduleType = moduleType;
  36. 	}
  37. }
  1. package com.cloudtech.util;
  2.  
  3. import com.cloudtech.enums.ModuleType;
  4.  
  5. public class Consts {
  6. 	/** 新版本服务器端口号 */
  7. 	public static Integer SERVER_PORT = 4000;
  8. 	/** 保存session的的属性值 */
  9. 	public static final String PILE_ID = "pileId";
  10. 	/** 请求成功 **/
  11. 	public static Integer SUCCESS = 1;
  12. 	/** 请求失败 **/
  13. 	public static Integer FAILED = 0;
  14. 	/** 主模块  */
  15. 	public static Integer MODULE_TYPE = ModuleType.MASTER.getCode();
  16. 	/** 模块类型id  */
  17. 	public static Integer STATION_TYPE_ID = 7;
  18. 	/** 设置最大缓存长度  */
  19. 	public static Integer MAX_BUFF_LEN=10240;
  20. }
  1. package com.cloudtech.server;
  2.  
  3. import javax.annotation.Resource;
  4.  
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  9.  
  10. import io.netty.bootstrap.ServerBootstrap;
  11. import io.netty.channel.ChannelFuture;
  12. import io.netty.channel.ChannelInitializer;
  13. import io.netty.channel.ChannelOption;
  14. import io.netty.channel.EventLoopGroup;
  15. import io.netty.channel.nio.NioEventLoopGroup;
  16. import io.netty.channel.socket.SocketChannel;
  17. import io.netty.channel.socket.nio.NioServerSocketChannel;
  18. import io.netty.handler.codec.string.StringDecoder;
  19. import io.netty.handler.timeout.IdleStateHandler;
  20.  
  21. /**
  22.  * 
  23. * @ClassName: Server  
  24. * @Description:netty服务端   
  25. * @author wude  
  26. * @date 2018年9月7日  
  27. *
  28.  */
  29. @Component
  30. public class Server {
  31. 	protected static final Logger logger = LoggerFactory.getLogger(Server.class);
  32. 	@Autowired
  33. 	private ServerHandler serverHandler;
  34.  
  35. 	
  36. 	/**
  37. 	 * 启动
  38. 	 */
  39. 	public void start(int port) {
  40.  
  41. 		// 服务类
  42. 		ServerBootstrap b = new ServerBootstrap();
  43.  
  44. 		// 创建boss和worker
  45. 		EventLoopGroup bossGroup = new NioEventLoopGroup();
  46.  		EventLoopGroup workerGroup = new NioEventLoopGroup();
  47. 		//业务线程池，实现消息串行化
  48. 		EventLoopGroup busyGroup = new NioEventLoopGroup();
  49.  
  50. 		try {
  51. 			// 设置循环线程组事例
  52. 			b.group(bossGroup, workerGroup);
  53.  
  54. 			// 设置channel工厂
  55. 			b.channel(NioServerSocketChannel.class);
  56.  
  57. 			// 设置管道
  58. 			b.childHandler(new ChannelInitializer<SocketChannel>() {
  59. 				@Override
  60. 				public void initChannel(SocketChannel ch) throws Exception {
  61. 					//第一个参数 读超时
  62. 					//第二个参数 写超时
  63. 					//第三个参数读写超时
  64. 					ch.pipeline().addLast(new IdleStateHandler(5, 5, 60));
  65. 					ch.pipeline().addLast(new RequestDecoder());
  66. 					//ch.pipeline().addLast(new DelimiterBasedFrameDecoder(Consts.MAX_BUFF_LEN,delimiter));
  67. 					ch.pipeline().addLast(new StringDecoder());
  68. 					ch.pipeline().addLast(busyGroup,serverHandler);
  69. 				}
  70. 			});
  71.  
  72. 			b.option(ChannelOption.SO_BACKLOG, 2048);//serverSocketchannel的设置，链接缓冲池的大小
  73. 			b.childOption(ChannelOption.SO_KEEPALIVE, true);//socketchannel的设置,维持链接的活跃，清除死链接
  74. 			b.childOption(ChannelOption.TCP_NODELAY, true);//socketchannel的设置,关闭延迟发送
  75.  
  76. 			logger.info("启动Dipper采集成功!port={}",port);
  77. 			//绑定端口
  78. 			ChannelFuture future = b.bind(port);
  79. 			
  80. 			//等待服务端关闭
  81. 			future.channel().closeFuture().sync();
  82. 			
  83. 		} catch (Exception e) {
  84. 			e.printStackTrace();
  85. 		} finally{
  86. 			//释放资源
  87. 			bossGroup.shutdownGracefully();
  88. 			workerGroup.shutdownGracefully();
  89. 			busyGroup.shutdownGracefully();
  90. 		}
  91. 	}
  92.  
  93. }

2.2.3  编码器 

  1. package com.cloudtech.server;
  2.  
  3. import java.util.List;
  4.  
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7.  
  8. import com.cloudtech.util.BufferUtil;
  9. import com.cloudtech.web.vo.RequestVo;
  10.  
  11. import io.netty.buffer.ByteBuf;
  12. import io.netty.channel.ChannelHandlerContext;
  13. import io.netty.handler.codec.ByteToMessageDecoder;
  14. /**
  15.  * 数据包解码器
  16.  * <pre>
  17.  * 数据包格式
  18.  * +——----——--------+——-----——+——----———+——----——-+-----------+---------+--------+---------|--------|-------
  19.  * |内容(5)|整个包长度(n)|用户地址 (3)|信息类别(1)|发信方地址(3)| 发信时间(2)|电文长度(n-20)|电文内容|CRC校验(1)|校验和(1)
  20.  * +——----——--------+——-----——+——----———+——----——-+-----------+---------+--------+---------|--------|-------
  21.  * 
  22.  * 
  23.  * </pre>
  24.  *  一个完整包的长度至少为20
  25.  *  
  26. * @author wude  
  27. * @date 2018年9月7日  
  28.  * 
  29.  */
  30. public class RequestDecoder extends ByteToMessageDecoder {
  31. 	protected static Logger LOGGER = LoggerFactory.getLogger(RequestDecoder.class);
  32. 	
  33. 	/**
  34. 	 * 数据包基本长度
  35. 	 */
  36. 	public static int BASE_LENTH = 20;
  37.  
  38. 	@Override
  39. 	protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
  40. 		//防止socket攻击,也就是说可能会存在有人恶意攻击程序，定义Integer最大值，向服务器发请求,造成服务器cpu,内存的开销增多
  41. 		try {
  42. 			if(buffer.readableBytes() > 2048){
  43. 				buffer.skipBytes(buffer.readableBytes());
  44. 			}
  45. 		/*	while(true){*/
  46. 				if(buffer.readableBytes() >= BASE_LENTH){   //包最小长度应为20
  47. 					//第一个可读数据包的起始位置
  48. 					while(true) {
  49. 						//标记初始读游标位置
  50. 						buffer.markReaderIndex();
  51. 						
  52. 						String str = BufferUtil.readFixLenAscStr(buffer, 5);
  53. 						//判断头部
  54. 						if(str.equals("$TXXX")){   //通讯信息
  55. 							buffer.resetReaderIndex();
  56. 							break;
  57. 						}
  58. 						//未读到包头标识略过一个字节
  59. 						buffer.resetReaderIndex();
  60. 						buffer.readByte();
  61. 						//不满足
  62. 						if(buffer.readableBytes() < BASE_LENTH - 5){
  63. 							return;
  64. 						}
  65. 					}
  66. 					//标记初始读游标位置
  67. 					buffer.markReaderIndex();
  68. 					
  69. 					//内容
  70. 					String content = BufferUtil.readFixLenAscStr(buffer, 5);
  71. 					//长度
  72. 					Short len = buffer.readShort();
  73. 					
  74. 					if(buffer.readableBytes() <len-7){
  75. 						return;
  76. 					}
  77. 					RequestVo requestVo = new RequestVo();
  78. 					//地址
  79. 					String userAddress =BufferUtil.readFixHexLenStr(buffer, 3);
  80. 					//信息类别
  81. 					byte infoType = buffer.readByte();
  82. 					//发信方地
  83. 					String sendStr = BufferUtil.readFixHexLenStr(buffer, 3);
  84. 					//发信时间
  85. 					byte hh = buffer.readByte();
  86. 					byte mm = buffer.readByte();
  87. 					//电文长度
  88. 					int contenLen = buffer.readShort() / 8;
  89. 					if(len !=(contenLen + 20) || contenLen <= 78){  
  90. 						//说明报文长度不对
  91. 						buffer.resetReaderIndex();
  92. 						BufferUtil.readFixLen(buffer, len);
  93. 						return;
  94. 					}
  95. 					//数据
  96. 					byte[] datas = new byte[contenLen];
  97. 					buffer.readBytes(datas);
  98. 					//CRC校验
  99. 					byte crc = buffer.readByte();
  100. 					//校验和
  101. 					byte checkSum = buffer.readByte();
  102. 					requestVo.setContent(content);
  103. 					requestVo.setLen(len.intValue());
  104. 					requestVo.setUserAddress(userAddress);
  105. 					requestVo.setInfoType(infoType);
  106. 					requestVo.setSendAddress(sendStr);
  107. 					requestVo.setHh(hh);
  108. 					requestVo.setMm(mm);
  109. 					requestVo.setContenLen(contenLen);
  110. 					
  111. 					requestVo.setData(datas);
  112. 					requestVo.setCrc(crc);
  113. 					requestVo.setCheckSum(checkSum);
  114. 					
  115. 					buffer.resetReaderIndex();
  116. 					//原始报文
  117. 					String bytes2Hex = BufferUtil.readFixHexLen(buffer, len);
  118. 					requestVo.setReportStr(bytes2Hex);
  119. 					
  120. 					out.add(requestVo);
  121. 				}else{
  122. 					//数据不完整，等待完整的数据包
  123. 					return;
  124. 				}
  125. 		/*	}*/
  126. 			//数据不完整，等待完整的数据包
  127. 			return;
  128. 		} catch (Exception e) {
  129. 			return;
  130. 		}
  131. 	}
  132. }

 2.2.4 业务处理类

  1. package com.cloudtech.server;
  2.  
  3. import java.net.InetSocketAddress;
  4.  
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.stereotype.Component;
  8.  
  9. import com.cloudtech.server.comm.ModuleId;
  10. import com.cloudtech.server.scanner.Invoker;
  11. import com.cloudtech.server.scanner.InvokerHoler;
  12. import com.cloudtech.server.serial.BufferFactory;
  13. import com.cloudtech.server.session.Session;
  14. import com.cloudtech.server.session.SessionImpl;
  15. import com.cloudtech.util.BufferUtil;
  16. import com.cloudtech.web.dubbo.BaseDataResult;
  17. import com.cloudtech.web.vo.RequestVo;
  18. import com.cloudtech.web.vo.StationVo;
  19.  
  20. import io.netty.buffer.ByteBuf;
  21. import io.netty.channel.ChannelHandlerContext;
  22. import io.netty.channel.SimpleChannelInboundHandler;
  23.  
  24. /**
  25.  * 
  26.  * @ClassName: SessionManager
  27.  * @Description: 消息接受处理类
  28.  * @author wude
  29.  * @date 2018年9月7日
  30.  */
  31. @Component
  32. public class ServerHandler extends SimpleChannelInboundHandler<RequestVo> {
  33. 	private static Logger LOGGER = LoggerFactory.getLogger(ServerHandler.class);
  34.    	@Reference(version = "${demo.service.version}",timeout=30000,check=false)
  35. 	DubboService dubboService;
  36.  
  37. 	@Override
  38. 	protected void messageReceived(ChannelHandlerContext ctx, RequestVo requestVo) throws Exception {
  39. 		System.out.println(requestVo.toString());
  40. 		byte[] data = requestVo.getData();
  41. 		ByteBuf buffer = BufferFactory.getBuffer(data);
  42. 		String stationNum = BufferUtil.readFixLenStr(buffer, 5);
  43. 		System.out.println(stationNum);
  44. 	}
  45. }

@Component标识加入容器中

可再接收内中，调用dubboserver接口

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/854840
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号