netty学习：UDP服务器与Spring整合（2）

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

spring.profiles.active=test spring.messages.encoding=utf-8 logging.config=classpath:logback.xml

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

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

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;
}
}

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();
}
} }

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;
} }

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

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> { }

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;
}
}

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;
}
}

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;
}
}

byte[] buffer = ...
InetAddress address = InetAddress.getByName("localhost"); DatagramPacket packet = new DatagramPacket(
buffer, buffer.length, address, 9999);
DatagramSocket datagramSocket = new DatagramSocket();
datagramSocket.send(packet);

DatagramSocket datagramSocket = new DatagramSocket(9999); byte[] buffer =....
DatagramPacket packet = new DatagramPacket(buffer, buffer.length); datagramSocket.receive(packet);