mysql udp服务器_netty学习：UDP服务器与Spring整合（2）
上一篇文章中，介紹了netty實現UDP服務器的栗子。
本文將會對UDP服務器與spring boot整合起來，并使用RedisTemplate的操作類訪問Redis和使用Spring DATA JPA鏈接MySQL數據庫，其中會使用多線程、異步等知識。
只公布了一個框架，需要的同學可以根據此來進行擴展，增加自己需要的功能模塊。如Controller部分。
本人使用的編輯器是IntelliJ IDEA 2017.1.exe版本(鏈接：http://pan.baidu.com/s/1pLODHm7 密碼：dlx7)；建議使用STS或者是idea編輯器來進行spring的學習。
1)項目目錄結構
整個項目的目錄結構如下：
2)jar包
其中pom.xml文件的內容如下：
只有netty-all和commons-lang3是手動加入的jar包，其余的都是創建spring boot項目時候選擇組件后自動導入的。
1 <?xml version="1.0" encoding="UTF-8"?>
2
3 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4 4.0.0
5
6 com.example
7 udplearning
8 0.0.1-SNAPSHOT
9 jar
10
11 udplearning
12 Demo project for Spring Boot
13
14
15 org.springframework.boot
16 spring-boot-starter-parent
17 1.5.6.RELEASE
18
19
20
21
22 UTF-8
23 UTF-8
24 3.4
25 1.8
26
27
28
29
30
31
32
33 io.netty
34 netty-all
35 4.0.49.Final
36
37
38
39
40 org.apache.commons
41 commons-lang3
42 ${commons-lang3.version}
43
44
45
46
47
48 org.springframework.boot
49 spring-boot-starter-data-jpa
50
51
52 org.springframework.boot
53 spring-boot-starter-data-redis
54
55
56 org.springframework.boot
57 spring-boot-starter-jdbc
58
59
60 org.springframework.boot
61 spring-boot-starter-web
62
63
64 org.springframework.boot
65 spring-boot-starter-web-services
66
67
68
69 mysql
70 mysql-connector-java
71 runtime
72
73
74 org.springframework.boot
75 spring-boot-starter-test
76 test
77
78
79
80
81
82
83 org.springframework.boot
84 spring-boot-maven-plugin
85
86
87
88
89
90
3)配置文件application.properties
application.properties的內容：
1 spring.profiles.active=test2
3 spring.messages.encoding=utf-8
4
5 logging.config=classpath:logback.xml
“spring.profiles.active”?針對多種啟動環境的spring boot配置方法，此時啟動的是test運行環境，即默認是啟動application-test.properties里面的配置信息；
“spring.messages.encoding=utf-8”是指編碼方式utf-8；
“logging.config=classpath:logback.xml”是指日志文件位置。
application-test.properties的內容如下：
1 context.listener.classes=com.example.demo.init.StartupEvent2
3 #mysql4 spring.jpa.show-sql=true
5 spring.jpa.database=mysql6 #spring.jpa.hibernate.ddl-auto=update7 spring.datasource.url=jdbc:mysql://127.0.0.1/test
8 spring.datasource.username=root9 spring.datasource.password=123456
10 spring.datasource.driver-class-name=com.mysql.jdbc.Driver11 spring.datasource.jdbc-interceptors=ConnectionState;SlowQueryReport(threshold=0)12
13 spring.session.store-type=none14
15 # (RedisProperties)16 spring.redis.database=3
17 spring.redis.host=127.0.0.1
18 spring.redis.port=6379
19 spring.redis.password=123456
20 spring.redis.pool.max-active=8
21 spring.redis.pool.max-wait=-1
22 spring.redis.pool.max-idle=8
23 spring.redis.pool.min-idle=0
24 spring.redis.timeout=0
25
26
27 #UDP消息接收打端口28 sysfig.udpReceivePort = 7686
29
30 #線程池31 spring.task.pool.corePoolSize = 5
32 spring.task.pool.maxPoolSize = 100
33 spring.task.pool.keepAliveSeconds = 100
34 spring.task.pool.queueCapacity = 100
其中配置了context.listener.classes=com.example.demo.init.StartupEvent，將StartupEvent類作為Spring boot啟動后執行文件。
其中還配置了一些mysql、redis和自定義的屬性。可根據項目的實際情況修改。
4)日志文件logback.xml
logback.xml的內容如下：
1 <?xml version="1.0" encoding="UTF-8"?>
2
3 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4 xsi:schemaLocation="http://ch.qos.logback/xml/ns/logback
5 http://ch.qos.logback/xml/ns/logback/logback.xsd
6 http://ch.qos.logback/xml/ns/logback ">
7
8
9 ${APP_Name}
10
11
12
13 %d{yyyyMMddHHmmss}|%-5level| %logger{0}.%M | %msg | %thread %n
14
15
16
17
18
19 ${catalina.home}/logs/app.%d{yyyyMMdd}.log
20 30
21
22
23 %d{yyMMddHHmmss.SSS}|%-5level| %msg%n
24
25
26
27
28
29 ${catalina.home}/logs/run.%d{yyyyMMdd}.log
30 7
31
32
33 %d{yyMMddHHmmss.SSS}|%-5level| %msg%n
34
35
36
37
38
39
40
41
42
43
44
45
日志的級別是info級別 ?可以根據自己項目的實際情況進行設置。
5)StartupEvent.java
1 packagecom.example.demo.init;2
3 importorg.slf4j.Logger;4 importorg.slf4j.LoggerFactory;5 importorg.springframework.context.ApplicationContext;6 importorg.springframework.context.ApplicationListener;7 importorg.springframework.context.event.ContextRefreshedEvent;8
9 /**
10 *11 * Created by wj on 2017/8/28.12 */
13
14 public class StartupEvent implements ApplicationListener{15 private static final Logger log = LoggerFactory.getLogger(StartupEvent.class);16
17 private staticApplicationContext context;18
19 @Override20 public voidonApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {21
22 try{23
24 context =contextRefreshedEvent.getApplicationContext();25
26 SysConfig sysConfig = (SysConfig) context.getBean(SysConfig.class);27
28 //接收UDP消息并保存至redis中
29 UdpServer udpServer = (UdpServer)StartupEvent.getBean(UdpServer.class);30 udpServer.run(sysConfig.getUdpReceivePort());31
32
33 //這里可以開啟多個線程去執行不同的任務34 //此處為工作的內容，不便公開！
35
36
37 } catch(Exception e) {38 log.error("Exception", e);39 }40 }41
42 public staticObject getBean(Class beanName) {43 return context != null ? context.getBean(beanName) : null;44 }45 }
6)UdpServer.java
1 packagecom.example.demo.init;2
3 importcom.example.demo.handle.UdpServerHandler;4 importio.netty.bootstrap.Bootstrap;5 importio.netty.channel.ChannelOption;6 importio.netty.channel.EventLoopGroup;7 importio.netty.channel.nio.NioEventLoopGroup;8 importio.netty.channel.socket.nio.NioDatagramChannel;9 importorg.slf4j.Logger;10 importorg.slf4j.LoggerFactory;11 importorg.springframework.scheduling.annotation.Async;12 importorg.springframework.stereotype.Component;13
14 /**
15 * server服務器16 * Created by wj on 2017/8/30.17 */
18 @Component19 public classUdpServer {20
21 private static final Logger log= LoggerFactory.getLogger(UdpServer.class);22
23 //private static final int PORT = Integer.parseInt(System.getProperty("port", "7686"));
24
25 @Async("myTaskAsyncPool")26 public void run(intudpReceivePort) {27
28 EventLoopGroup group = newNioEventLoopGroup();29 log.info("Server start! Udp Receive msg Port:" +udpReceivePort );30
31 try{32 Bootstrap b = newBootstrap();33 b.group(group)34 .channel(NioDatagramChannel.class)35 .option(ChannelOption.SO_BROADCAST, true)36 .handler(newUdpServerHandler());37
38 b.bind(udpReceivePort).sync().channel().closeFuture().await();39 } catch(InterruptedException e) {40 e.printStackTrace();41 } finally{42 group.shutdownGracefully();43 }44 }45
46 }
此處NioDatagramChannel.class采用的是非阻塞的模式接受UDP消息，若是接受的UDP消息少，可以采用阻塞式的方式接受UDP消息。
UdpServer.run()方法使用@Async將該方法定義成異步的，myTaskAsyncPool是自定義的線程池。
7)UdpServerHandler.java
1 packagecom.example.demo.handle;2
3 importcom.example.demo.init.StartupEvent;4 importcom.example.demo.mod.UdpRecord;5 importcom.example.demo.repository.mysql.UdpRepository;6 importcom.example.demo.repository.redis.RedisRepository;7 importio.netty.buffer.Unpooled;8 importio.netty.channel.ChannelHandlerContext;9 importio.netty.channel.SimpleChannelInboundHandler;10 importio.netty.channel.socket.DatagramPacket;11 importio.netty.util.CharsetUtil;12 importorg.apache.commons.lang3.StringUtils;13 importorg.slf4j.Logger;14 importorg.slf4j.LoggerFactory;15
16 importjava.sql.Timestamp;17 importjava.util.Date;18
19 /**
20 * 接受UDP消息，并保存至redis的list鏈表中21 * Created by wj on 2017/8/30.22 *23 */
24
25 public class UdpServerHandler extends SimpleChannelInboundHandler{26
27 private static final Logger log= LoggerFactory.getLogger(UdpServerHandler.class);28
29 //用來計算server接收到多少UDP消息
30 private static int count = 0;31
32 @Override33 public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throwsException {34
35 String receiveMsg =packet.content().toString(CharsetUtil.UTF_8);36
37 log.info("Received UDP Msg:" +receiveMsg);38
39 UdpRecord udpRecord = newUdpRecord();40
41 //判斷接受到的UDP消息是否正確(未實現)
42 if(StringUtils.isNotEmpty(receiveMsg) ){43
44 //計算接收到的UDP消息的數量
45 count++;46
47 //獲取UdpRepository對象，將接收UDP消息的日志保存至mysql中
48 udpRecord.setUdpMsg(receiveMsg);49 udpRecord.setTime(getTime());50 UdpRepository udpRepository = (UdpRepository) StartupEvent.getBean(UdpRepository.class);51 udpRepository.save(udpRecord);52
53 //獲取RedirRepository對象
54 RedisRepository redisRepository = (RedisRepository) StartupEvent.getBean(RedisRepository.class);55 //將獲取到的UDP消息保存至redis的list列表中
56 redisRepository.lpush("udp:msg", receiveMsg);57 redisRepository.setKey("UDPMsgNumber", String.valueOf(count));58
59
60 //在這里可以返回一個UDP消息給對方，告知已接收到UDP消息，但考慮到這是UDP消息，此處可以注釋掉
61 ctx.write(newDatagramPacket(62 Unpooled.copiedBuffer("QOTM: " + "Got UDP Message!", CharsetUtil.UTF_8), packet.sender()));63
64 }else{65 log.error("Received Error UDP Messsage:" +receiveMsg);66 }67 }68
69 @Override70 public voidchannelReadComplete(ChannelHandlerContext ctx) {71 ctx.flush();72 }73
74 @Override75 public voidexceptionCaught(ChannelHandlerContext ctx, Throwable cause) {76 cause.printStackTrace();77 //We don't close the channel because we can keep serving requests.
78 }79
80 publicTimestamp getTime(){81 Date date = newDate();82 Timestamp time = newTimestamp(date.getTime());83 returntime;84 }85
86 }
此處若不借用ApplicationContext.getBean，是無法獲取到RedisRepository對象的。
注：這里是無法使用注解@Autowired來獲取到redisTemplate對象的。
8)repository
RedisRepository.java
1 packagecom.example.demo.repository.redis;2
3 importorg.slf4j.Logger;4 importorg.slf4j.LoggerFactory;5 importorg.springframework.beans.factory.annotation.Autowired;6 importorg.springframework.data.redis.core.RedisTemplate;7 importorg.springframework.stereotype.Service;8
9 /**
10 * 鏈接redis11 * 實現list lpush和rpop12 * Created by wj on 2017/8/30.13 */
14
15
16 @Service17 public classRedisRepository {18 private static final Logger log = LoggerFactory.getLogger(RedisRepository.class);19
20 @Autowired21 private RedisTemplateredisTemplate;22
23 //----------------String-----------------------
24 public voidsetKey(String key,String value){25 redisTemplate.opsForValue().set(key, value);26 }27
28
29 //----------------list----------------------
30 public Long lpush(String key, String val) throwsException{31 log.info("UDP Msg保存至redis中，key:" + key + ",val:" +val);32 returnredisTemplate.opsForList().leftPush(key, val);33 }34
35 public String rpop(String key) throwsException {36 returnredisTemplate.opsForList().rightPop(key);37 }38
39 }
使用springframework框架中的RedisTemplate類去鏈接redis，此處是將收到的UDP消息左保存(lpush)至list鏈表中,然后右邊彈出(rpop)。
UdpRepository.java
1 packagecom.example.demo.repository.mysql;2
3 importcom.example.demo.mod.UdpRecord;4 importorg.springframework.data.jpa.repository.JpaRepository;5
6 /**
7 * Created by wj on 2017/8/31.8 */
9 public interface UdpRepository extends JpaRepository{10
11 }
定義Spring Data JPA接口，鏈接數據庫。
其中
UdpRecord.java
1 packagecom.example.demo.mod;2
3 importjavax.persistence.Entity;4 importjavax.persistence.GeneratedValue;5 importjavax.persistence.Id;6 importjavax.persistence.Table;7 importjava.sql.Timestamp;8
9 /**
10 * Created by wj on 2017/8/31.11 *12 * 用來記錄接收的UDP消息的日志13 */
14 @Entity15 @Table16 public classUdpRecord {17
18 private longid;19 privateString udpMsg;20 privateTimestamp time;21
22 @Id23 @GeneratedValue24 public longgetId() {25 returnid;26 }27
28 public void setId(longid) {29 this.id =id;30 }31
32 publicString getUdpMsg() {33 returnudpMsg;34 }35
36 public voidsetUdpMsg(String udpMsg) {37 this.udpMsg =udpMsg;38 }39
40 publicTimestamp getTime() {41 returntime;42 }43
44 public voidsetTime(Timestamp time) {45 this.time =time;46 }47 }
注解@Entity和@Table辨明這是一個實體類表格 ，其中的@Id和@GeneratedValue表明id是key值并且是自動遞增的。
9)線程池的相關信息
TaskExecutePool.java
1 packagecom.example.demo.thread;2
3 importcom.example.demo.init.SysConfig;4 importorg.springframework.beans.factory.annotation.Autowired;5 importorg.springframework.context.annotation.Bean;6 importorg.springframework.context.annotation.Configuration;7 importorg.springframework.scheduling.annotation.EnableAsync;8 importorg.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;9
10 importjava.util.concurrent.Executor;11 importjava.util.concurrent.ThreadPoolExecutor;12
13 /**
14 * Created by wangjian on 2017/8/29.15 */
16 @Configuration17 @EnableAsync18 public classTaskExecutePool {19
20 @Autowired21 privateSysConfig config;22
23 @Bean24 publicExecutor myTaskAsyncPool() {25 ThreadPoolTaskExecutor executor = newThreadPoolTaskExecutor();26 executor.setCorePoolSize(config.getCorePoolSize());27 executor.setMaxPoolSize(config.getMaxPoolSize());28 executor.setQueueCapacity(config.getQueueCapacity());29 executor.setKeepAliveSeconds(config.getKeepAliveSeconds());30 executor.setThreadNamePrefix("MyExecutor-");31
32 //rejection-policy：當pool已經達到max size的時候，如何處理新任務33 //CALLER_RUNS：不在新線程中執行任務，而是由調用者所在的線程來執行
34 executor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());35 executor.initialize();36 returnexecutor;37 }38 }
10)配置文件SysConfig.java
1 packagecom.example.demo.init;2
3 importorg.springframework.boot.context.properties.ConfigurationProperties;4 importorg.springframework.stereotype.Component;5
6 /**
7 * Created by wj on 2017/8/30.8 */
9 @Component10 @ConfigurationProperties(prefix="sysfig")11 public classSysConfig {12 private int UdpReceivePort;//UDP消息接收端口13
14 //線程池信息
15 private intCorePoolSize;16
17 private intMaxPoolSize;18
19 private intKeepAliveSeconds;20
21 private intQueueCapacity;22
23 public intgetCorePoolSize() {24 returnCorePoolSize;25 }26
27 public void setCorePoolSize(intcorePoolSize) {28 CorePoolSize =corePoolSize;29 }30
31 public intgetMaxPoolSize() {32 returnMaxPoolSize;33 }34
35 public void setMaxPoolSize(intmaxPoolSize) {36 MaxPoolSize =maxPoolSize;37 }38
39 public intgetKeepAliveSeconds() {40 returnKeepAliveSeconds;41 }42
43 public void setKeepAliveSeconds(intkeepAliveSeconds) {44 KeepAliveSeconds =keepAliveSeconds;45 }46
47 public intgetQueueCapacity() {48 returnQueueCapacity;49 }50
51 public void setQueueCapacity(intqueueCapacity) {52 QueueCapacity =queueCapacity;53 }54
55 public intgetUdpReceivePort() {56 returnUdpReceivePort;57 }58
59 public void setUdpReceivePort(intudpReceivePort) {60 UdpReceivePort =udpReceivePort;61 }62 }
11)小結
其實發送UDP和接收UDP消息的核心代碼很簡單，只是netty框架將其包裝了。
UDP發送消息是
1 byte[] buffer =...2 InetAddress address = InetAddress.getByName("localhost");3
4 DatagramPacket packet = newDatagramPacket(5 buffer, buffer.length, address, 9999);6 DatagramSocket datagramSocket = newDatagramSocket();7 datagramSocket.send(packet);
udp接收消息是
1 DatagramSocket datagramSocket = new DatagramSocket(9999);2
3 byte[] buffer =....4 DatagramPacket packet = newDatagramPacket(buffer, buffer.length);5
6 datagramSocket.receive(packet);
看起來是不是很簡單？？？
12)源代碼下載地址
這里只公布了一個框架，其他很多部分由于涉及到了工作內容不便公布。
有需要的同學可以自行下載對其代碼進行更改。
總結
以上是生活随笔為你收集整理的mysql udp服务器_netty学习：UDP服务器与Spring整合（2）的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： linux 查杀php木马,linux上
- 下一篇： mysql sqldump_mysql