SpringBoot Webflux创建TCP/UDP server并使用handler解析数据

SpringBoot Webflux创建TCP/UDP server并使用handler解析数据

目录1.pom依赖引用spring-boot-starter-webflux依赖2.创建UDP/TCPServer3.数据解析handler（具体解析根据vQXSYe协议来）解析UDP数据handler解析TCP数据handler4.测试工具推荐使用SocketTool调试TCP/UDP协议

1.pom依赖

引用spring-boot-starter-webflux依赖

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

4.0.0

org.springframework.boot

spring-boot-starter-parent

2.1.3.RELEASE

com.example

demo

0.0.1-SNAPSHOT

demo

Demo project for Spring Boot

1.8

org.springframework.boot

spring-boot-starter

org.springframework.boot

spring-boot-starter-test

test

org.springframework.boot

spring-boot-starter-webflux

org.springframework.boot

spring-boot-maven-plugin

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

4.0.0

org.springframework.boot

spring-boot-starter-parent

2.1.3.RELEASE

com.example

demo

0.0.1-SNAPSHOT

demo

Demo project for Spring Boot

1.8

org.springframework.boot

spring-boot-starter

org.springframework.boot

spring-boot-starter-test

test

org.springframework.boot

spring-boot-starter-webflux

org.springframework.boot

spring-boot-maven-plugin

2.创建UDP/TCP Server

package com.example.demo;

import com.example.demo.handler.TcpDecoderHanlder;

import com.example.demo.handler.UdpDecoderHanlder;

import org.springframework.boot.CommandLineRunner;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.context.annotation.Bean;

import reactor.core.publisher.Flux;

import reactor.netty.tcp.TcpServer;

import reactor.netty.udp.UdpServer;

import java.time.Duration;

@SpringBootApplication

public class DemoApplication {

public static void main(String[] args) {

SpringApplication.run(DemoApplication.class, args);

}

@Bean

CommandLineRunner serverRunner(UdpDecoderHandler udpDecoderHandler, TcpDecoderHandler tcpDecoderHandler) {

return strings -> {

createUdpServer(udpDecoderHandler);

createTcpServer(tcpDecoderHandler);

};

}

/**

* 创建UDP Server

* @param udpDecoderHanlder： 解析UDP Client上报数据handler

*/

private void createUdpServer(UdpDecoderHandler udpDecoderHandler) {

UdpServer.create()

.handle((in,out) -> {

in.receive()

.asByteArray()

.subscribe();

return Flux.never();

})

.port(8888)

.doOnBound(conn -> conn.addHandlerLast("decoder",udpDecoderHandler)) //可以添加多个handler

.bindNow(Duration.ofSeconds(30));

}

/**

* 创建TCP Server

* @param tcpDecoderHanlder： 解析TCP Client上报数据的handler

*/

private void createTcpServer(TcpDecoderHandler tcpDecoderHandler) {

TcpServer.create()

.handle((in,out) -> {

in.receive()

.asByteArray()

.subscribe();

return Flux.never();

})

.doOnConnection(conn ->

conn.addHandler(tcpDecoderHandler)) //实例只写了如何添加handler,可添加delimiter，tcp生命周期，decoder，encoder等handler

.port(9999)

.bindNow();

}

}

3.数据解析handler（具体解析根据协议来）

解析UDP数据handler

package com.example.demo.handler;

import io.netty.buffer.ByteBuf;

import io.netty.channel.ChannelHandlerContext;

import io.netty.channel.socket.DatagramPacket;

import io.netty.handler.codec.MessageToMessageDecoder;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Service;

import java.util.List;

@Service

public class UdpDecoderHandler extends MessageToMessageDecoder {

private static final Logger LOGGER = LoggerFactory.getLogger(TcpDecoderHandler.class);

@Override

protected void decode(ChannelHandlerContext channelHandlerContext, DatagramPacket byteBuf, List

ByteBuf byteBuf1 = byteBuf.content();

int size = byteBuf1.readableBytes();

byte[] data = new byte[size];

byteBuf1.readBytes(data);

LOGGER.info(new String(data));

}

}

解析TCP数据handler

package com.example.demo.handler;

import io.netty.channel.ChannelHandlerContext;

import io.netty.handler.codec.MessageToMessageDecoder;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Service;

import java.util.List;

@Service

public class TcpDecoderHandler extends MessageToMessageDecoder {

private static final Logger LOGGER = LoggerFactory.getLogger(TcpDecoderHandler.class);

@Override

protected void decode(ChannelHandlerContext channelHandlerContext, Object o, List list){

LOGGER.info("解析client上报数据");

}

}

4.测试工具

推荐使用SocketTool调试TCP/UDP协议

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。