Skip to content

T
tcp-receiver
Commit b564375d authored by wuyuhang's avatar wuyuhang
Browse files
Options

Initial commit

parents
Hide whitespace changes
Inline Side-by-side
Showing with 1034 additions and 0 deletions
.idea/.gitignore 0 → 100644
View file @ b564375d
# 默认忽略的文件
/shelf/
/workspace.xml
# 基于编辑器的 HTTP 客户端请求
/httpRequests/
# 依赖于环境的 Maven 主目录路径
/mavenHomeManager.xml
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml
.idea/compiler.xml 0 → 100644
View file @ b564375d
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="CompilerConfiguration">
<annotationProcessing>
<profile name="Maven default annotation processors profile" enabled="true">
<sourceOutputDir name="target/generated-sources/annotations" />
<sourceTestOutputDir name="target/generated-test-sources/test-annotations" />
<outputRelativeToContentRoot value="true" />
<module name="tcp-receiver" />
</profile>
</annotationProcessing>
</component>
</project>
\ No newline at end of file
.idea/encodings.xml 0 → 100644
View file @ b564375d
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Encoding">
<file url="file://$PROJECT_DIR$/src/main/java" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/src/main/resources" charset="UTF-8" />
</component>
</project>
\ No newline at end of file
.idea/jarRepositories.xml 0 → 100644
View file @ b564375d
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="RemoteRepositoriesConfiguration">
<remote-repository>
<option name="id" value="esv-snapshots" />
<option name="name" value="esv-snapshots" />
<option name="url" value="http://36.129.47.17:18084/repository/esv-snapshots/" />
</remote-repository>
<remote-repository>
<option name="id" value="esv-releases" />
<option name="name" value="esv-releases" />
<option name="url" value="http://36.129.47.17:18084/repository/maven-releases/" />
</remote-repository>
<remote-repository>
<option name="id" value="cloudera.repo" />
<option name="name" value="cloudera.repo" />
<option name="url" value="https://repository.cloudera.com/artifactory/cloudera-repos/" />
</remote-repository>
<remote-repository>
<option name="id" value="central" />
<option name="name" value="Central Repository" />
<option name="url" value="https://maven.aliyun.com/repository/central/" />
</remote-repository>
<remote-repository>
<option name="id" value="esv-repository" />
<option name="name" value="esv-repository" />
<option name="url" value="http://36.129.47.17:18084/repository/maven-public/" />
</remote-repository>
<remote-repository>
<option name="id" value="aliyun-public" />
<option name="name" value="aliyun-public" />
<option name="url" value="https://maven.aliyun.com/repository/public" />
</remote-repository>
<remote-repository>
<option name="id" value="central" />
<option name="name" value="Maven Central repository" />
<option name="url" value="https://repo1.maven.org/maven2" />
</remote-repository>
<remote-repository>
<option name="id" value="jboss.community" />
<option name="name" value="JBoss Community repository" />
<option name="url" value="https://repository.jboss.org/nexus/content/repositories/public/" />
</remote-repository>
<remote-repository>
<option name="id" value="blade-proxy" />
<option name="name" value="blade-proxy" />
<option name="url" value="http://36.129.47.17:18084/repository/blade-proxy/" />
</remote-repository>
</component>
</project>
\ No newline at end of file
.idea/misc.xml 0 → 100644
View file @ b564375d
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ExternalStorageConfigurationManager" enabled="true" />
<component name="MavenProjectsManager">
<option name="originalFiles">
<list>
<option value="$PROJECT_DIR$/pom.xml" />
</list>
</option>
</component>
<component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" default="true" project-jdk-name="1.8" project-jdk-type="JavaSDK" />
</project>
\ No newline at end of file
.idea/vcs.xml 0 → 100644
View file @ b564375d
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="$PROJECT_DIR$/.." vcs="Git" />
<mapping directory="$PROJECT_DIR$" vcs="Git" />
</component>
</project>
\ No newline at end of file
dependency-reduced-pom.xml 0 → 100644
View file @ b564375d
<?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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.example</groupId>
<artifactId>tcp-receiver</artifactId>
<version>1.0.0</version>
<build>
<finalName>tcp-receiver</finalName>
<plugins>
<plugin>
<artifactId>maven-shade-plugin</artifactId>
<version>3.5.1</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer>
<mainClass>com.example.tcpreceiver.TcpServer</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<properties>
<snakeyaml.version>1.33</snakeyaml.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<netty.version>4.1.110.Final</netty.version>
<maven.compiler.target>1.8</maven.compiler.target>
</properties>
</project>
pom.xml 0 → 100644
View file @ b564375d
<?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>tcp-receiver</artifactId>
<version>1.0.0</version>
<packaging>jar</packaging>
<properties>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<netty.version>4.1.110.Final</netty.version>
<snakeyaml.version>1.33</snakeyaml.version>
<!-- 实际部署到Linux服务器时建议显式指定classifier，比如 linux-x86_64，
以确保拉到epoll native包，见下方 netty-transport-native-epoll 依赖说明 -->
</properties>
<dependencies>
<!-- Netty 核心 -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>${netty.version}</version>
</dependency>
<!-- Linux epoll 原生传输，比NIO Selector性能更好，吞吐量大场景必备 -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<version>${netty.version}</version>
<classifier>linux-x86_64</classifier>
</dependency>
<!-- YAML配置解析 -->
<dependency>
<groupId>org.yaml</groupId>
<artifactId>snakeyaml</artifactId>
<version>${snakeyaml.version}</version>
</dependency>
<!-- 简单日志实现，方便看控制台输出，生产环境可换成logback/log4j2 -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.36</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.7.36</version>
</dependency>
</dependencies>
<build>
<finalName>tcp-receiver</finalName>
<plugins>
<!-- 打包成可直接运行的fat jar -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.5.1</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.example.tcpreceiver.TcpServer</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
src/main/java/TcpSender.java 0 → 100644
View file @ b564375d
import com.example.tcpsend.NettyTcpPushServer;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* TCP压测发送端。
*
* 用法:
* javac TcpSender.java
* java TcpSender <host> <port> <连接数> <持续秒数> [单次写入buffer大小KB，默认64]
*
* 示例:
* java TcpSender 192.168.1.100 9000 8 60 256
* （8个并发连接，往192.168.1.100:9000发60秒，每次write 256KB）
*
* 设计说明:
* - 发送内容随便填充（全0即可），测试目标是吞吐量不是内容本身
* - 每个连接一个线程，死循环write，不等待也不读取响应（纯压测接收端的读取能力）
* - 用一个共享的AtomicLong累加所有线程发出的字节数，主线程每秒打印一次瞬时速率和累计速率
* - 多连接是为了模拟"多个客户端同时发数据"的场景，单连接很难打满万兆网卡（受限于单TCP流和单核处理能力）
*/
public class TcpSender {
private static final AtomicLong totalBytesSent = new AtomicLong(0);
private static final AtomicBoolean running = new AtomicBoolean(true);
public static void main(String[] args) throws InterruptedException {
if (args.length < 4) {
System.out.println("用法: java TcpSender <host> <port> <连接数> <持续秒数> [单次写入buffer大小KB，默认64]");
System.out.println("示例: java TcpSender 192.168.1.100 9000 8 60 256");
return;
}
String host = args[0];
int port = Integer.parseInt(args[1]);
int connections = Integer.parseInt(args[2]);
int durationSeconds = Integer.parseInt(args[3]);
int bufferSizeKB = args.length >= 5 ? Integer.parseInt(args[4]) : 64;
System.out.println("[配置] host=" + host + ", port=" + port
+ ", 连接数=" + connections + ", 持续秒数=" + durationSeconds
+ ", 单次写入大小=" + bufferSizeKB + "KB");
Thread[] senderThreads = new Thread[connections];
for (int i = 0; i < connections; i++) {
int connIndex = i;
senderThreads[i] = new Thread(() -> senderWorker(host, port, bufferSizeKB, connIndex), "sender-" + i);
senderThreads[i].start();
}
// 统计打印线程：每秒打印一次瞬时速率和累计速率
Thread reporter = new Thread(() -> {
long lastBytes = 0;
long startTime = System.nanoTime();
while (running.get()) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
break;
}
long currentBytes = totalBytesSent.get();
long deltaBytes = currentBytes - lastBytes;
lastBytes = currentBytes;
double elapsedSeconds = (System.nanoTime() - startTime) / 1_000_000_000.0;
double instantMBps = deltaBytes / 1024.0 / 1024.0;
double instantGbps = (deltaBytes * 8) / 1_000_000_000.0;
double avgMBps = (currentBytes / 1024.0 / 1024.0) / elapsedSeconds;
System.out.println(String.format(
"[t=%.0fs] 瞬时速率=%.2f MB/s (%.2f Gbps), 累计=%.2f MB, 平均速率=%.2f MB/s",
elapsedSeconds, instantMBps, instantGbps,
currentBytes / 1024.0 / 1024.0, avgMBps
));
}
}, "reporter");
reporter.setDaemon(true);
reporter.start();
// 主线程负责控制总时长，时间到了把running置false，通知所有发送线程退出
Thread.sleep(durationSeconds * 1000L);
running.set(false);
for (Thread t : senderThreads) {
t.interrupt();
t.join(2000);
}
long totalBytes = totalBytesSent.get();
System.out.println(String.format(
"[测试结束] 总耗时=%ds, 总发送字节数=%d (%.2f MB / %.2f GB), 平均速率=%.2f MB/s (%.2f Gbps)",
durationSeconds, totalBytes,
totalBytes / 1024.0 / 1024.0, totalBytes / 1024.0 / 1024.0 / 1024.0,
(totalBytes / 1024.0 / 1024.0) / durationSeconds,
(totalBytes * 8.0 / 1_000_000_000.0) / durationSeconds
));
}
/**
* 单个连接的发送逻辑：建立连接后死循环write固定内容的buffer，直到收到停止信号。
*/
private static void senderWorker(String host, int port, int bufferSizeKB, int connIndex) {
byte[] buffer = new byte[bufferSizeKB * 1024];
// 内容随便填，这里填个固定字节模式，方便和"全0"区分，便于抓包时肉眼确认
for (int i = 0; i < buffer.length; i++) {
buffer[i] = (byte) (i % 256);
}
try (Socket socket = new Socket()) {
// 适当调大发送缓冲区，配合接收端的SO_RCVBUF，减少写阻塞
socket.setSendBufferSize(4 * 1024 * 1024);
socket.setTcpNoDelay(true);
socket.connect(new InetSocketAddress(host, port), 5000);
System.out.println("[连接成功] sender-" + connIndex + " -> " + host + ":" + port);
OutputStream out = socket.getOutputStream();
while (running.get() && !Thread.currentThread().isInterrupted()) {
out.write(buffer);
totalBytesSent.addAndGet(buffer.length);
}
} catch (Exception e) {
System.out.println("[连接异常] sender-" + connIndex + ": " + e.getMessage());
}
}
}
\ No newline at end of file
src/main/java/com/example/tcpreceiver/ConnectionHandler.java 0 → 100644
View file @ b564375d
package com.example.tcpreceiver;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.net.InetSocketAddress;
/**
* 核心连接处理器。
*
* 设计要点：
* - 继承 SimpleChannelInboundHandler<ByteBuf>，框架会在每次channelRead0返回后自动release(msg)，
* 不需要手动管理引用计数，避免内存泄漏。
* - 只做计数，不拷贝、不持有、不写文件，意味着这里的开销极小，瓶颈应该落在网络IO本身。
* - 每个连接独享一个Handler实例（Netty默认每条连接new一个pipeline），所以这里的字段不需要加锁。
*/
public class ConnectionHandler extends SimpleChannelInboundHandler<ByteBuf> {
private final GlobalStats globalStats;
private long bytesReceived = 0L;
private long connectionStartNanos;
private String remoteIp;
public ConnectionHandler(GlobalStats globalStats) {
// 关闭自动release，因为SimpleChannelInboundHandler默认就会release，这里显式调用父类无参构造即可
super();
this.globalStats = globalStats;
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
connectionStartNanos = System.nanoTime();
InetSocketAddress remote = (InetSocketAddress) ctx.channel().remoteAddress();
remoteIp = remote.getAddress().getHostAddress();
System.out.println("[连接建立] " + remoteIp + ":" + remote.getPort());
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
// 只接收不存储不处理：仅累加字节数，内容直接丢弃。
// msg会在本方法返回后由父类自动release，不需要也不应该手动retain/缓存。
bytesReceived += msg.readableBytes();
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
long durationNanos = System.nanoTime() - connectionStartNanos;
double durationSeconds = durationNanos / 1_000_000_000.0;
double mbReceived = bytesReceived / 1024.0 / 1024.0;
double throughputMBps = durationSeconds > 0 ? mbReceived / durationSeconds : 0;
System.out.println(String.format(
"[连接结束] %s, 耗时=%.3fs, 接收字节数=%d (%.2f MB), 平均速率=%.2f MB/s",
remoteIp, durationSeconds, bytesReceived, mbReceived, throughputMBps
));
globalStats.recordConnectionFinished(remoteIp, bytesReceived);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
System.err.println("[连接异常] " + remoteIp + ": " + cause.getMessage());
ctx.close();
}
}
src/main/java/com/example/tcpreceiver/GlobalStats.java 0 → 100644
View file @ b564375d
package com.example.tcpreceiver;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
/**
* 全局统计：
* 1. 按客户端IP维度累加历史总字节数、总连接数（跨多次连接持续累加）
* 2. 维护全局总字节数/总连接数，供定期打印汇总使用
*/
public class GlobalStats {
public static class IpStat {
public final AtomicLong totalBytes = new AtomicLong(0);
public final AtomicLong connectionCount = new AtomicLong(0);
}
private final Map<String, IpStat> ipStatsMap = new ConcurrentHashMap<>();
private final AtomicLong globalTotalBytes = new AtomicLong(0);
private final AtomicLong globalConnectionCount = new AtomicLong(0);
/**
* 一次连接结束时调用，累加该IP的统计和全局统计。
*/
public void recordConnectionFinished(String ip, long bytesReceived) {
IpStat stat = ipStatsMap.computeIfAbsent(ip, k -> new IpStat());
stat.totalBytes.addAndGet(bytesReceived);
stat.connectionCount.incrementAndGet();
globalTotalBytes.addAndGet(bytesReceived);
globalConnectionCount.incrementAndGet();
}
public long getGlobalTotalBytes() {
return globalTotalBytes.get();
}
public long getGlobalConnectionCount() {
return globalConnectionCount.get();
}
public Map<String, IpStat> getIpStatsMap() {
return ipStatsMap;
}
}
src/main/java/com/example/tcpreceiver/ServerConfig.java 0 → 100644
View file @ b564375d
package com.example.tcpreceiver;
import org.yaml.snakeyaml.Yaml;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Map;
/**
* 配置加载类。
* 优先从外部文件路径加载（部署时放在jar包同目录或任意指定路径），
* 找不到外部文件时，回退到classpath内置的默认配置（resources/config.yaml），
* 保证服务在没有外部配置时也能用默认参数跑起来。
*/
public class ServerConfig {
public int port = 24584;
public boolean useEpoll = true;
public int bossThreads = 1;
public int workerThreads = 0;
public int soBacklog = 1024;
public boolean tcpNoDelay = true;
public boolean soKeepAlive = true;
public int soRcvbuf = 4 * 1024 * 1024;
public int soSndbuf = 64 * 1024;
public boolean usePooledAllocator = true;
public int recvBufferAllocatorMax = 1024 * 1024;
public int recvBufferAllocatorMin = 64 * 1024;
public int recvBufferAllocatorInitial = 256 * 1024;
public boolean enablePerIpStats = true;
public int globalStatsIntervalSeconds = 30;
public int rate;
public int dataLength;
public int dataTime;
/**
* 加载配置。
*
* @param externalPath 外部配置文件路径，可为null。
* 为null或文件不存在时，使用classpath下的config.yaml作为默认值。
*/
@SuppressWarnings("unchecked")
public static ServerConfig load(String externalPath) {
ServerConfig cfg = new ServerConfig();
Map<String, Object> root = null;
try {
if (externalPath != null && Files.exists(Paths.get(externalPath))) {
try (InputStream in = new FileInputStream(externalPath)) {
root = new Yaml().load(in);
System.out.println("[Config] 使用外部配置文件: " + externalPath);
}
} else {
InputStream in = ServerConfig.class.getClassLoader().getResourceAsStream("config.yaml");
if (in != null) {
root = new Yaml().load(in);
in.close();
System.out.println("[Config] 外部配置未指定或不存在，使用内置默认 config.yaml");
} else {
System.out.println("[Config] 未找到任何配置文件，使用代码内硬编码默认值");
}
}
} catch (IOException e) {
System.out.println("[Config] 读取配置文件失败，使用默认值: " + e.getMessage());
}
if (root == null) {
return cfg;
}
Map<String, Object> server = (Map<String, Object>) root.get("server");
if (server != null) {
cfg.port = getInt(server, "port", cfg.port);
cfg.useEpoll = getBool(server, "useEpoll", cfg.useEpoll);
cfg.bossThreads = getInt(server, "bossThreads", cfg.bossThreads);
cfg.workerThreads = getInt(server, "workerThreads", cfg.workerThreads);
cfg.soBacklog = getInt(server, "soBacklog", cfg.soBacklog);
cfg.tcpNoDelay = getBool(server, "tcpNoDelay", cfg.tcpNoDelay);
cfg.soKeepAlive = getBool(server, "soKeepAlive", cfg.soKeepAlive);
cfg.soRcvbuf = getInt(server, "soRcvbuf", cfg.soRcvbuf);
cfg.soSndbuf = getInt(server, "soSndbuf", cfg.soSndbuf);
cfg.usePooledAllocator = getBool(server, "usePooledAllocator", cfg.usePooledAllocator);
cfg.recvBufferAllocatorMax = getInt(server, "recvBufferAllocatorMax", cfg.recvBufferAllocatorMax);
cfg.recvBufferAllocatorMin = getInt(server, "recvBufferAllocatorMin", cfg.recvBufferAllocatorMin);
cfg.recvBufferAllocatorInitial = getInt(server, "recvBufferAllocatorInitial", cfg.recvBufferAllocatorInitial);
}
Map<String, Object> sendServer = (Map<String, Object>) root.get("sendServer");
if (sendServer != null) {
cfg.rate = getInt(sendServer, "rate", cfg.rate);
cfg.dataLength = getInt(sendServer, "dataLength", cfg.dataLength);
cfg.dataTime = getInt(sendServer, "dataTime", cfg.dataTime);
}
Map<String, Object> stats = (Map<String, Object>) root.get("stats");
if (stats != null) {
cfg.enablePerIpStats = getBool(stats, "enablePerIpStats", cfg.enablePerIpStats);
cfg.globalStatsIntervalSeconds = getInt(stats, "globalStatsIntervalSeconds", cfg.globalStatsIntervalSeconds);
}
return cfg;
}
private static int getInt(Map<String, Object> map, String key, int def) {
Object v = map.get(key);
if (v == null) return def;
return ((Number) v).intValue();
}
private static boolean getBool(Map<String, Object> map, String key, boolean def) {
Object v = map.get(key);
if (v == null) return def;
return (Boolean) v;
}
@Override
public String toString() {
return "ServerConfig{" +
"port=" + port +
", useEpoll=" + useEpoll +
", bossThreads=" + bossThreads +
", workerThreads=" + workerThreads +
", soBacklog=" + soBacklog +
", tcpNoDelay=" + tcpNoDelay +
", soKeepAlive=" + soKeepAlive +
", soRcvbuf=" + soRcvbuf +
", soSndbuf=" + soSndbuf +
", usePooledAllocator=" + usePooledAllocator +
", recvBufferAllocatorMax=" + recvBufferAllocatorMax +
", recvBufferAllocatorMin=" + recvBufferAllocatorMin +
", recvBufferAllocatorInitial=" + recvBufferAllocatorInitial +
", enablePerIpStats=" + enablePerIpStats +
", globalStatsIntervalSeconds=" + globalStatsIntervalSeconds +
'}';
}
}
src/main/java/com/example/tcpreceiver/TcpServer.java 0 → 100644
View file @ b564375d
package com.example.tcpreceiver;
import com.example.tcpsend.NettyTcpPushServer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.ServerChannel;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* TCP接收服务启动类。
*
* 启动方式：
* java -jar tcp-receiver.jar [外部配置文件路径]
*
* 不传参数时，使用jar内置的默认 config.yaml。
* 生产部署时建议把config.yaml拷贝到jar同目录，按服务器配置调整后通过参数传入路径，
* 这样换服务器或调参数都不需要重新打包。
*/
public class TcpServer {
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
try {
NettyTcpPushServer.main(args);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}).start();
String externalConfigPath = args.length > 0 ? args[0] : null;
ServerConfig cfg = ServerConfig.load(externalConfigPath);
System.out.println("[启动] 加载到的配置: " + cfg);
boolean useEpoll = cfg.useEpoll && Epoll.isAvailable();
if (cfg.useEpoll && !Epoll.isAvailable()) {
System.out.println("[启动] 配置要求使用epoll，但当前环境不支持，自动降级为NIO");
}
EventLoopGroup bossGroup;
EventLoopGroup workerGroup;
Class<? extends ServerChannel> channelClass;
if (useEpoll) {
bossGroup = new EpollEventLoopGroup(cfg.bossThreads);
workerGroup = cfg.workerThreads > 0
? new EpollEventLoopGroup(cfg.workerThreads)
: new EpollEventLoopGroup();
channelClass = EpollServerSocketChannel.class;
System.out.println("[启动] 使用 Epoll 传输");
} else {
bossGroup = new NioEventLoopGroup(cfg.bossThreads);
workerGroup = cfg.workerThreads > 0
? new NioEventLoopGroup(cfg.workerThreads)
: new NioEventLoopGroup();
channelClass = NioServerSocketChannel.class;
System.out.println("[启动] 使用 NIO 传输");
}
GlobalStats globalStats = new GlobalStats();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(channelClass)
.option(ChannelOption.SO_BACKLOG, cfg.soBacklog)
.childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) {
// 每条连接独立的Handler实例，内部计数字段互不影响
ch.pipeline().addLast(new ConnectionHandler(globalStats));
}
});
// ===== 子连接（实际数据通道）参数 =====
bootstrap.childOption(ChannelOption.TCP_NODELAY, cfg.tcpNoDelay);
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, cfg.soKeepAlive);
if (cfg.soRcvbuf > 0) {
bootstrap.childOption(ChannelOption.SO_RCVBUF, cfg.soRcvbuf);
}
if (cfg.soSndbuf > 0) {
bootstrap.childOption(ChannelOption.SO_SNDBUF, cfg.soSndbuf);
}
// 内存分配器：池化直接内存可以显著降低高吞吐场景下的GC压力和内存拷贝开销
if (cfg.usePooledAllocator) {
bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
} else {
bootstrap.childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT);
}
// 自适应接收缓冲区：根据实际数据量在min~max之间动态调整单次读取大小，
// 大流量场景适当调大可以减少系统调用次数
bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
new AdaptiveRecvByteBufAllocator(
cfg.recvBufferAllocatorMin,
cfg.recvBufferAllocatorInitial,
cfg.recvBufferAllocatorMax
));
Channel serverChannel = bootstrap.bind(cfg.port).sync().channel();
System.out.println("[启动] 服务已启动，监听端口: " + cfg.port);
serverChannel.closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
src/main/java/com/example/tcpsend/NettyTcpPushServer.java 0 → 100644
View file @ b564375d
package com.example.tcpsend;
import com.example.tcpreceiver.ServerConfig;
import com.example.tcpreceiver.TcpServer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.handler.traffic.ChannelTrafficShapingHandler;
import io.netty.util.concurrent.ScheduledFuture;
import java.time.LocalTime;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
* Netty高性能TCP推送服务端
* 客户端连接后全速循环下发固定十六进制报文，无连接时不发包、不消耗CPU
*/
public class NettyTcpPushServer {
// ===================== 配置区 =====================
private static final int LISTEN_PORT = 27586;
// Netty 发送缓冲区高低水位，控制推送上限
private static final int HIGH_WATER_MARK = 1024 * 1024 * 8;
private static final int LOW_WATER_MARK = 1024 * 1024 * 4;
// 固定下发报文十六进制
static final String PUSH_HEX_MSG = "8001001003046e0004066610021100b50c000000000000000000000000008a0301870500000000000000308f050000000000ffffffffffffffff328f050000000000338f050000000000348f050000000000358f050000000000368f050000000000378f050000000000388f050000000000398f0500000000003a8f0500000000003b8f0500000000003c8f0500000000003d8f0500000000003e8f0500000000003f8f050000000000408f050000000000418f050000000000428f050000000000438f050000000000448f050000000000458f050000000000468f050000000000478f050000000000488f050000000000498f0500000000004a8f0500000000004b8f0500000000004c8f0500000000004d8f0500000000004e8f0500000000004f8f050000000000508f050000000000518f050000000000528f050000000000538f050000000000548f050000000000558f050000000000568f050000000000578f050000000000588f050000000000598f0500000000005a8f0500000000005b8f0500000000005c8f0500000000005d8f0500000000005e8f0500000000005f8f050000000000608f050000000000618f050000000000628f050000000000638f050000000000648f050000000000658f050000000000668f050000000000678f050000000000688f050000000000698f0500000000006a8f0500000000006b8f0500000000006c8f0500000000006d8f0500000000006e8f0500000000006f8f050000000000708f050000000000718f050000000000728f050000000000738f050000000000748f050000000000758f050000000000768f050000000000778f050000000000788f050000000000798f0500000000007a8f0500000000007b8f0500000000007c8f0500000000007d8f0500000000007e8f0500000000007f8f050000000000808f050000000000818f050000000000828f050000000000838f050000000000848f050000000000858f050000000000868f050000000000878f050000000000888f050000000000898f0500000000008a8f0500000000008b8f0500000000008c8f0500000000008d8f0500000000008e8f0500000000008f8f050000000000908f050000000000918f050000000000928f050000000000938f050000000000948f050000000000958f050000000000968f050000000000978f050000000000988f050000000000998f0500000000009a8f0500000000009b8f0500000000009c8f0500000000009d8f0500000000009e8f0500000000009f8f050000000000";
static final String END_HEX_MSG = "8001001003046e0004066607040200b50c000000000000000000000000008a03018705000000000000003005050000000000ffffffffffffffff328f050000000000338f050000000000348f050000000000358f050000000000368f050000000000378f050000000000388f050000000000398f0500000000003a8f0500000000003b8f0500000000003c8f0500000000003d8f0500000000003e8f0500000000003f8f050000000000408f050000000000418f050000000000428f050000000000438f050000000000448f050000000000458f050000000000468f050000000000478f050000000000488f050000000000498f0500000000004a8f0500000000004b8f0500000000004c8f0500000000004d8f0500000000004e8f0500000000004f8f050000000000508f050000000000518f050000000000528f050000000000538f050000000000548f050000000000558f050000000000568f050000000000578f050000000000588f050000000000598f0500000000005a8f0500000000005b8f0500000000005c8f0500000000005d8f0500000000005e8f0500000000005f8f050000000000608f050000000000618f050000000000628f050000000000638f050000000000648f050000000000658f050000000000668f050000000000678f050000000000688f050000000000698f0500000000006a8f0500000000006b8f0500000000006c8f0500000000006d8f0500000000006e8f0500000000006f8f050000000000708f050000000000718f050000000000728f050000000000738f050000000000748f050000000000758f050000000000768f050000000000778f050000000000788f050000000000798f0500000000007a8f0500000000007b8f0500000000007c8f0500000000007d8f0500000000007e8f0500000000007f8f050000000000808f050000000000818f050000000000828f050000000000838f050000000000848f050000000000858f050000000000868f050000000000878f050000000000888f050000000000898f0500000000008a8f0500000000008b8f0500000000008c8f0500000000008d8f0500000000008e8f0500000000008f8f050000000000908f050000000000918f050000000000928f050000000000938f050000000000948f050000000000958f050000000000968f050000000000978f050000000000988f050000000000998f0500000000009a8f0500000000009b8f0500000000009c8f0500000000009d8f0500000000009e8f0500000000009f8f050000000000";
// ===================== 全局静态资源 =====================
// 全局预编译报文Buf，所有通道复用
private static ByteBuf GLOBAL_PUSH_BUF;
private static ByteBuf GLOBAL_END_BUF;
// 在线客户端通道集合
public static final Set<Channel> ONLINE_CHANNELS = ConcurrentHashMap.newKeySet();
public static final AtomicLong TOTAL_BYTE_SENT = new AtomicLong(0);
public static final AtomicLong END_BYTE_SENT = new AtomicLong(0);
// 单条报文字节长度
public static int MSG_BYTE_LEN;
public static int END_BYTE_LEN;
public static ServerConfig cfg = null;
static {
// 启动一次性转换十六进制报文
byte[] msgBytes = hexToBytes(PUSH_HEX_MSG);
MSG_BYTE_LEN = msgBytes.length;
GLOBAL_PUSH_BUF = PooledByteBufAllocator.DEFAULT.buffer(MSG_BYTE_LEN);
GLOBAL_PUSH_BUF.writeBytes(msgBytes);
// 统计初始字节基数
TOTAL_BYTE_SENT.set(0);
// 启动一次性转换十六进制报文
byte[] endMsgBytes = hexToBytes(END_HEX_MSG);
END_BYTE_LEN = endMsgBytes.length;
GLOBAL_END_BUF = PooledByteBufAllocator.DEFAULT.buffer(END_BYTE_LEN);
GLOBAL_END_BUF.writeBytes(endMsgBytes);
// 统计初始字节基数
END_BYTE_SENT.set(0);
}
public static void main(String[] args) throws InterruptedException {
String externalConfigPath = args.length > 0 ? args[0] : null;
cfg = ServerConfig.load(externalConfigPath);
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.option(ChannelOption.SO_BACKLOG, 1024)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.childOption(ChannelOption.TCP_NODELAY, true) // 关闭Nagle，极致推送
.childOption(ChannelOption.SO_SNDBUF, 4 * 1024 * 1024)
.childOption(ChannelOption.SO_RCVBUF, 4 * 1024 * 1024)
.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, HIGH_WATER_MARK)
.childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, LOW_WATER_MARK)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
// pipeline.addLast(new LoggingHandler(LogLevel.DEBUG)); // 调试打开，压测关闭
pipeline.addLast(new IdleStateHandler(0, 50, 0));
pipeline.addLast(new PushHandler());
}
});
// 绑定端口启动服务
ChannelFuture bindFuture = bootstrap.bind(LISTEN_PORT).sync();
System.out.println("TCP推送服务启动成功，监听端口：" + LISTEN_PORT);
bindFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
/** 推送处理器，单个客户端通道独立循环推送 */
static class PushHandler extends ChannelInboundHandlerAdapter {
private Channel clientChannel;
private ScheduledFuture<?> pushTask;
// 全局统计指标
public AtomicLong TOTAL_PACK_SENT = new AtomicLong(0);
// ===== 限速配置 =====
private static final int RATE_PER_SECOND = 134 * cfg.rate; // 目标：每秒发送包数，按需调整
private static final int SLICE_PER_SECOND = 100; // 把1秒拆成100片，越大越平滑，但调度开销也越大
private static final int PACKETS_PER_SLICE = Math.max(1, RATE_PER_SECOND / SLICE_PER_SECOND);
private static final int SLICE_INTERVAL_MS = 1000 / SLICE_PER_SECOND;
private final int MAX_PUSH_SIZE = cfg.dataLength;
private boolean MAX_TIME = false;
@Override
public void channelActive(ChannelHandlerContext ctx) {
clientChannel = ctx.channel();
ONLINE_CHANNELS.add(clientChannel);
System.out.println("客户端接入：" + clientChannel.remoteAddress() + "，开始限速推送，目标 " + RATE_PER_SECOND + " 包/秒");
startRateLimitedPush(ctx);
if(cfg.dataTime != 0) {
ctx.channel().eventLoop().scheduleWithFixedDelay(() -> {
MAX_TIME = true;
}, cfg.dataTime, cfg.dataTime, TimeUnit.SECONDS);
} else {
MAX_TIME = true;
}
}
private void startRateLimitedPush(ChannelHandlerContext ctx) {
// scheduleAtFixedRate运行在该Channel所属的EventLoop线程上，天然串行、线程安全
pushTask = ctx.channel().eventLoop().scheduleAtFixedRate(() -> {
if (!clientChannel.isActive()) {
stopPush();
return;
}
if (!clientChannel.isWritable()) {
// 缓冲区高水位，本时间片跳过，不空转、不阻塞，下个时间片自然重试
return;
}
for (int i = 0; i < PACKETS_PER_SLICE; i++) {
ByteBuf sendBuf = GLOBAL_PUSH_BUF.duplicate().retain();
LocalTime now = LocalTime.now();
long nanoOfDay = now.toNanoOfDay(); // 纳秒
sendBuf.setIntLE(26, (int) (nanoOfDay / 100_000L));
clientChannel.write(sendBuf);
TOTAL_PACK_SENT.incrementAndGet();
TOTAL_BYTE_SENT.addAndGet(MSG_BYTE_LEN);
if (TOTAL_PACK_SENT.get() >= MAX_PUSH_SIZE && MAX_TIME) {
sendEndAndStop();
return;
}
}
clientChannel.flush();
}, 0, SLICE_INTERVAL_MS, TimeUnit.MILLISECONDS);
}
private void sendEndAndStop() {
for (int j = 0; j < 5; j++) {
ByteBuf endBuf = GLOBAL_END_BUF.duplicate().retain();
if (j == 4) {
clientChannel.writeAndFlush(endBuf).addListener(ChannelFutureListener.CLOSE);
} else {
clientChannel.write(endBuf);
}
END_BYTE_SENT.addAndGet(END_BYTE_LEN);
}
stopPush();
}
private void stopPush() {
if (pushTask != null) {
pushTask.cancel(true);
pushTask = null;
clientChannel.close();
}
}
/** 缓冲区水位变化回调，可写后恢复推送 */
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws InterruptedException {
/*if(clientChannel.isWritable()) {
startPushLoop();
}*/
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof IdleStateEvent) {
IdleStateEvent e = (IdleStateEvent) evt;
if (e.state() == IdleState.WRITER_IDLE) {
System.out.println("写超时，可能对端消费异常，主动断开：" + ctx.channel().remoteAddress());
ctx.close();
}
}
}
/** 客户端断开，停止推送，移除在线集合 */
@Override
public void channelInactive(ChannelHandlerContext ctx) {
Channel ch = ctx.channel();
ONLINE_CHANNELS.remove(ch);
stopPush();
System.out.println("客户端断开：" + ch.remoteAddress() + "，停止该通道推送");
}
/** 异常直接关闭通道 */
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
public static byte[] hexToBytes(String hexStr) {
if (hexStr == null || hexStr.trim().isEmpty()) {
return new byte[0];
}
int length = hexStr.length();
byte[] bytes = new byte[length / 2];
for (int i = 0; i < length; i += 2) {
int high = Character.digit(hexStr.charAt(i), 16);
int low = Character.digit(hexStr.charAt(i + 1), 16);
bytes[i / 2] = (byte) ((high << 4) | low);
}
return bytes;
}
}
\ No newline at end of file
src/main/resources/config.yaml 0 → 100644
View file @ b564375d
# ====================================
# TCP接收服务 配置文件
# 部署到不同配置的服务器时，主要调整这里
# ====================================
server:
# 监听端口
port: 24584
sendRate: 800
# 是否优先使用Linux epoll原生传输（强烈建议Linux服务器开启，性能优于NIO）
# 如果不是Linux环境或没有对应native包，自动降级为NIO
useEpoll: true
# Boss线程数：负责接收新连接，一般1-2个足够，不需要很多
bossThreads: 1
# Worker线程数：负责实际IO读写，决定了并发处理能力
# 建议设置为 CPU核心数 或 CPU核心数*2，根据实际压测调整
# 0表示使用Netty默认值（CPU核数*2）
workerThreads: 0
# ===== TCP相关参数 =====
# 服务端accept队列大小，高并发连接场景适当调大，避免握手被丢弃
soBacklog: 1024
# 是否禁用Nagle算法。只接收数据场景通常建议true，降低延迟
tcpNoDelay: true
# 是否开启TCP keepalive，检测死连接
soKeepAlive: true
# 接收缓冲区大小（字节）。万兆网大流量场景建议调大，比如 1MB ~ 4MB
# 0表示不显式设置，使用系统默认值
soRcvbuf: 4194304
# 发送缓冲区大小（字节），本服务只接收基本不发送，可以设小一点或者不设
soSndbuf: 65536
# ===== Netty内部缓冲区/内存相关 =====
# 是否使用池化的直接内存分配器（推荐true，减少GC压力和内存拷贝）
usePooledAllocator: true
# 每次读取的最大缓冲区大小提示（字节），配合自适应ByteBuf分配器
# 大流量场景可以适当调大，减少读取系统调用次数
recvBufferAllocatorMax: 1048576
recvBufferAllocatorMin: 65536
recvBufferAllocatorInitial: 262144
sendServer:
rate: 30
dataLength: 120000
dataTime: 30
# 统计相关
stats:
# 是否按客户端IP维度汇总统计（多次连接累加）
enablePerIpStats: true
# 是否定期打印全局汇总统计（所有IP总和），单位：秒。0表示关闭定期打印
globalStatsIntervalSeconds: 30
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment