當前位置：首頁 > 运维知识 > 数据库 >内容正文

数据库

mysql udp服务器_netty学习：UDP服务器与Spring整合（2）

發布時間：2025/3/15 数据库 25 豆豆

生活随笔收集整理的這篇文章主要介紹了 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"?>

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

6 com.example

7 udplearning

8 0.0.1-SNAPSHOT

9 jar

11 udplearning

12 Demo project for Spring Boot

15 org.springframework.boot

16 spring-boot-starter-parent

17 1.5.6.RELEASE

22 UTF-8

23 UTF-8

24 3.4

25 1.8

33 io.netty

34 netty-all

35 4.0.49.Final

40 org.apache.commons

41 commons-lang3

42 ${commons-lang3.version}

48 org.springframework.boot

49 spring-boot-starter-data-jpa

52 org.springframework.boot

53 spring-boot-starter-data-redis

56 org.springframework.boot

57 spring-boot-starter-jdbc

60 org.springframework.boot

61 spring-boot-starter-web

64 org.springframework.boot

65 spring-boot-starter-web-services

69 mysql

70 mysql-connector-java

71 runtime

74 org.springframework.boot

75 spring-boot-starter-test

76 test

83 org.springframework.boot

84 spring-boot-maven-plugin

3)配置文件application.properties

application.properties的內容：

1 spring.profiles.active=test2

3 spring.messages.encoding=utf-8

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

27 #UDP消息接收打端口28 sysfig.udpReceivePort = 7686

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"?>

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 ">

9 ${APP_Name}

13 %d{yyyyMMddHHmmss}|%-5level| %logger{0}.%M | %msg | %thread %n

19 ${catalina.home}/logs/app.%d{yyyyMMdd}.log

20 30

23 %d{yyMMddHHmmss.SSS}|%-5level| %msg%n

29 ${catalina.home}/logs/run.%d{yyyyMMdd}.log

30 7

33 %d{yyMMddHHmmss.SSS}|%-5level| %msg%n

日志的級別是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 */

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

33 //這里可以開啟多個線程去執行不同的任務34 //此處為工作的內容，不便公開！

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"));

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 */

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

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 */

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

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