上一篇文章中，介绍了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项目时候选择组件后自动导入的。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion> <groupId>com.example</groupId>
<artifactId>udplearning</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging> <name>udplearning</name>
<description>Demo project for Spring Boot</description> <parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent> <properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<commons-lang3.version>3.4</commons-lang3.version>
<java.version>1.8</java.version>
</properties> <dependencies> <!-- netty --> <dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.0.49.Final</version>
</dependency> <dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>${commons-lang3.version}</version>
</dependency> <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web-services</artifactId>
</dependency> <dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies> <build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build> </project>
3）配置文件application.properties
application.properties的内容：
spring.profiles.active=test spring.messages.encoding=utf-8 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的内容如下：
context.listener.classes=com.example.demo.init.StartupEvent #mysql
spring.jpa.show-sql=true
spring.jpa.database=mysql
#spring.jpa.hibernate.ddl-auto=update
spring.datasource.url=jdbc:mysql://127.0.0.1/test
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
spring.datasource.jdbc-interceptors=ConnectionState;SlowQueryReport(threshold=0) spring.session.store-type=none # (RedisProperties)
spring.redis.database=3
spring.redis.host=127.0.0.1
spring.redis.port=6379
spring.redis.password=123456
spring.redis.pool.max-active=8
spring.redis.pool.max-wait=-1
spring.redis.pool.max-idle=8
spring.redis.pool.min-idle=0
spring.redis.timeout=0 #UDP消息接收打端口
sysfig.udpReceivePort = 7686 #线程池
spring.task.pool.corePoolSize = 5
spring.task.pool.maxPoolSize = 100
spring.task.pool.keepAliveSeconds = 100
spring.task.pool.queueCapacity = 100
其中配置了context.listener.classes=com.example.demo.init.StartupEvent，将StartupEvent类作为Spring boot启动后执行文件。
其中还配置了一些mysql、redis和自定义的属性。可根据项目的实际情况修改。
4）日志文件logback.xml
logback.xml的内容如下：
<?xml version="1.0" encoding="UTF-8"?>
<configuration xmlns="http://ch.qos.logback/xml/ns/logback"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://ch.qos.logback/xml/ns/logback
http://ch.qos.logback/xml/ns/logback/logback.xsd
http://ch.qos.logback/xml/ns/logback ">
<property name="APP_Name" value="udplearning" />
<timestamp key="bySecond" datePattern="yyyyMMdd'T'HHmmss" />
<contextName>${APP_Name}</contextName> <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{yyyyMMddHHmmss}|%-5level| %logger{0}.%M | %msg | %thread %n</pattern>
</encoder>
</appender> <appender name="FILELOG" class="ch.qos.logback.core.rolling.RollingFileAppender">
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<fileNamePattern>${catalina.home}/logs/app.%d{yyyyMMdd}.log</fileNamePattern>
<maxHistory>30</maxHistory>
</rollingPolicy>
<encoder>
<pattern>%d{yyMMddHHmmss.SSS}|%-5level| %msg%n</pattern>
</encoder>
</appender> <appender name="RUNLOG" class="ch.qos.logback.core.rolling.RollingFileAppender">
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<fileNamePattern>${catalina.home}/logs/run.%d{yyyyMMdd}.log</fileNamePattern>
<maxHistory>7</maxHistory>
</rollingPolicy>
<encoder>
<pattern>%d{yyMMddHHmmss.SSS}|%-5level| %msg%n</pattern>
</encoder>
</appender> <logger name="com.example.demo" level="debug" additivity="false">
<appender-ref ref="STDOUT" />
<appender-ref ref="FILELOG" />
</logger> <root level="info">
<appender-ref ref="STDOUT" />
</root>
</configuration>
日志的级别是info级别 可以根据自己项目的实际情况进行设置。
5）StartupEvent.java
package com.example.demo.init; import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent; /**
*
* Created by wj on 2017/8/28.
*/ public class StartupEvent implements ApplicationListener<ContextRefreshedEvent> {
private static final Logger log = LoggerFactory.getLogger(StartupEvent.class); private static ApplicationContext context; @Override
public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) { try { context = contextRefreshedEvent.getApplicationContext(); SysConfig sysConfig = (SysConfig) context.getBean(SysConfig.class); //接收UDP消息并保存至redis中
UdpServer udpServer = (UdpServer)StartupEvent.getBean(UdpServer.class);
udpServer.run(sysConfig.getUdpReceivePort()); // 这里可以开启多个线程去执行不同的任务
// 此处为工作的内容，不便公开！ } catch (Exception e) {
log.error("Exception", e);
}
} public static Object getBean(Class beanName) {
return context != null ? context.getBean(beanName) : null;
}
}
6)UdpServer.java
package com.example.demo.init; import com.example.demo.handle.UdpServerHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component; /**
* server服务器
* Created by wj on 2017/8/30.
*/
@Component
public class UdpServer { private static final Logger log= LoggerFactory.getLogger(UdpServer.class); // private static final int PORT = Integer.parseInt(System.getProperty("port", "7686")); @Async("myTaskAsyncPool")
public void run(int udpReceivePort) { EventLoopGroup group = new NioEventLoopGroup();
log.info("Server start! Udp Receive msg Port:" + udpReceivePort ); try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioDatagramChannel.class)
.option(ChannelOption.SO_BROADCAST, true)
.handler(new UdpServerHandler()); b.bind(udpReceivePort).sync().channel().closeFuture().await();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
} }
此处NioDatagramChannel.class采用的是非阻塞的模式接受UDP消息，若是接受的UDP消息少，可以采用阻塞式的方式接受UDP消息。
UdpServer.run()方法使用@Async将该方法定义成异步的，myTaskAsyncPool是自定义的线程池。
7）UdpServerHandler.java
package com.example.demo.handle; import com.example.demo.init.StartupEvent;
import com.example.demo.mod.UdpRecord;
import com.example.demo.repository.mysql.UdpRepository;
import com.example.demo.repository.redis.RedisRepository;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.CharsetUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import java.sql.Timestamp;
import java.util.Date; /**
* 接受UDP消息，并保存至redis的list链表中
* Created by wj on 2017/8/30.
*
*/ public class UdpServerHandler extends SimpleChannelInboundHandler<DatagramPacket> { private static final Logger log= LoggerFactory.getLogger(UdpServerHandler.class); //用来计算server接收到多少UDP消息
private static int count = 0; @Override
public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception { String receiveMsg = packet.content().toString(CharsetUtil.UTF_8); log.info("Received UDP Msg:" + receiveMsg); UdpRecord udpRecord = new UdpRecord(); //判断接受到的UDP消息是否正确（未实现）
if (StringUtils.isNotEmpty(receiveMsg) ){ //计算接收到的UDP消息的数量
count++; //获取UdpRepository对象，将接收UDP消息的日志保存至mysql中
udpRecord.setUdpMsg(receiveMsg);
udpRecord.setTime(getTime());
UdpRepository udpRepository = (UdpRepository) StartupEvent.getBean(UdpRepository.class);
udpRepository.save(udpRecord); //获取RedirRepository对象
RedisRepository redisRepository = (RedisRepository) StartupEvent.getBean(RedisRepository.class);
//将获取到的UDP消息保存至redis的list列表中
redisRepository.lpush("udp:msg", receiveMsg);
redisRepository.setKey("UDPMsgNumber", String.valueOf(count)); // 在这里可以返回一个UDP消息给对方，告知已接收到UDP消息，但考虑到这是UDP消息，此处可以注释掉
ctx.write(new DatagramPacket(
Unpooled.copiedBuffer("QOTM: " + "Got UDP Message!" , CharsetUtil.UTF_8), packet.sender())); }else{
log.error("Received Error UDP Messsage:" + receiveMsg);
}
} @Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
// We don't close the channel because we can keep serving requests.
} public Timestamp getTime(){
Date date = new Date();
Timestamp time = new Timestamp(date.getTime());
return time;
} }
此处若不借用ApplicationContext.getBean，是无法获取到RedisRepository对象的。
注：这里是无法使用注解@Autowired来获取到redisTemplate对象的。
8)repository
RedisRepository.java
package com.example.demo.repository.redis; import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service; /**
* 链接redis
* 实现list lpush和rpop
* Created by wj on 2017/8/30.
*/ @Service
public class RedisRepository {
private static final Logger log = LoggerFactory.getLogger(RedisRepository.class); @Autowired
private RedisTemplate<String, String> redisTemplate; //----------------String-----------------------
public void setKey(String key,String value){
redisTemplate.opsForValue().set(key, value);
} //----------------list----------------------
public Long lpush(String key, String val) throws Exception{
log.info("UDP Msg保存至redis中，key:" + key + ",val:" + val);
return redisTemplate.opsForList().leftPush(key, val);
} public String rpop(String key) throws Exception {
return redisTemplate.opsForList().rightPop(key);
} }
使用springframework框架中的RedisTemplate类去链接redis，此处是将收到的UDP消息左保存（lpush）至list链表中,然后右边弹出（rpop）。
UdpRepository.java
package com.example.demo.repository.mysql; import com.example.demo.mod.UdpRecord;
import org.springframework.data.jpa.repository.JpaRepository; /**
* Created by wj on 2017/8/31.
*/
public interface UdpRepository extends JpaRepository<UdpRecord,Long> { }
定义Spring Data JPA接口，链接数据库。
其中
UdpRecord.java
package com.example.demo.mod; import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.Id;
import javax.persistence.Table;
import java.sql.Timestamp; /**
* Created by wj on 2017/8/31.
*
* 用来记录接收的UDP消息的日志
*/
@Entity
@Table
public class UdpRecord { private long id;
private String udpMsg;
private Timestamp time; @Id
@GeneratedValue
public long getId() {
return id;
} public void setId(long id) {
this.id = id;
} public String getUdpMsg() {
return udpMsg;
} public void setUdpMsg(String udpMsg) {
this.udpMsg = udpMsg;
} public Timestamp getTime() {
return time;
} public void setTime(Timestamp time) {
this.time = time;
}
}
注解@Entity和@Table辨明这是一个实体类表格 ，其中的@Id和@GeneratedValue表明id是key值并且是自动递增的。
9）线程池的相关信息
TaskExecutePool.java
package com.example.demo.thread; import com.example.demo.init.SysConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor; /**
* Created by wangjian on 2017/8/29.
*/
@Configuration
@EnableAsync
public class TaskExecutePool { @Autowired
private SysConfig config; @Bean
public Executor myTaskAsyncPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(config.getCorePoolSize());
executor.setMaxPoolSize(config.getMaxPoolSize());
executor.setQueueCapacity(config.getQueueCapacity());
executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
executor.setThreadNamePrefix("MyExecutor-"); // rejection-policy：当pool已经达到max size的时候，如何处理新任务
// CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}
10)配置文件SysConfig.java
package com.example.demo.init; import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component; /**
* Created by wj on 2017/8/30.
*/
@Component
@ConfigurationProperties(prefix="sysfig")
public class SysConfig {
private int UdpReceivePort;//UDP消息接收端口 //线程池信息
private int CorePoolSize; private int MaxPoolSize; private int KeepAliveSeconds; private int QueueCapacity; public int getCorePoolSize() {
return CorePoolSize;
} public void setCorePoolSize(int corePoolSize) {
CorePoolSize = corePoolSize;
} public int getMaxPoolSize() {
return MaxPoolSize;
} public void setMaxPoolSize(int maxPoolSize) {
MaxPoolSize = maxPoolSize;
} public int getKeepAliveSeconds() {
return KeepAliveSeconds;
} public void setKeepAliveSeconds(int keepAliveSeconds) {
KeepAliveSeconds = keepAliveSeconds;
} public int getQueueCapacity() {
return QueueCapacity;
} public void setQueueCapacity(int queueCapacity) {
QueueCapacity = queueCapacity;
} public int getUdpReceivePort() {
return UdpReceivePort;
} public void setUdpReceivePort(int udpReceivePort) {
UdpReceivePort = udpReceivePort;
}
}
11）小结
其实发送UDP和接收UDP消息的核心代码很简单，只是netty框架将其包装了。
UDP发送消息是
byte[] buffer = ...
InetAddress address = InetAddress.getByName("localhost"); DatagramPacket packet = new DatagramPacket(
buffer, buffer.length, address, 9999);
DatagramSocket datagramSocket = new DatagramSocket();
datagramSocket.send(packet);
udp接收消息是
DatagramSocket datagramSocket = new DatagramSocket(9999); byte[] buffer =....
DatagramPacket packet = new DatagramPacket(buffer, buffer.length); datagramSocket.receive(packet);
看起来是不是很简单？？？
12）源代码下载地址
https://github.com/wj302763621/learning_udp.git
这里只公布了一个框架，其他很多部分由于涉及到了工作内容不便公布。
有需要的同学可以自行下载对其代码进行更改。