Commit 0b6a6bd3 authored by wuyuhang's avatar wuyuhang

将控制台输出迁移到 SLF4J

将 System.out/System.err/printStackTrace 替换为 SLF4J 日志调用。按场景使用 info、warn、error,并保留异常堆栈输出。
Co-Authored-By: default avatarClaude <noreply@anthropic.com>
parent 7592ccbe
import com.example.tcpsend.NettyTcpPushServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.OutputStream;
import java.net.InetSocketAddress;
......@@ -25,14 +26,16 @@ import java.util.concurrent.atomic.AtomicBoolean;
*/
public class TcpSender {
private static final Logger log = LoggerFactory.getLogger(TcpSender.class);
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");
log.info("用法: java TcpSender <host> <port> <连接数> <持续秒数> [单次写入buffer大小KB,默认64]");
log.info("示例: java TcpSender 192.168.1.100 9000 8 60 256");
return;
}
......@@ -42,9 +45,8 @@ public class TcpSender {
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");
log.info("[配置] host={}, port={}, 连接数={}, 持续秒数={}, 单次写入大小={}KB",
host, port, connections, durationSeconds, bufferSizeKB);
Thread[] senderThreads = new Thread[connections];
for (int i = 0; i < connections; i++) {
......@@ -72,11 +74,12 @@ public class TcpSender {
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
));
log.info("[t={}s] 瞬时速率={} MB/s ({} Gbps), 累计={} MB, 平均速率={} MB/s",
String.format("%.0f", elapsedSeconds),
String.format("%.2f", instantMBps),
String.format("%.2f", instantGbps),
String.format("%.2f", currentBytes / 1024.0 / 1024.0),
String.format("%.2f", avgMBps));
}
}, "reporter");
reporter.setDaemon(true);
......@@ -92,13 +95,13 @@ public class TcpSender {
}
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
));
log.info("[测试结束] 总耗时={}s, 总发送字节数={} ({} MB / {} GB), 平均速率={} MB/s ({} Gbps)",
durationSeconds,
totalBytes,
String.format("%.2f", totalBytes / 1024.0 / 1024.0),
String.format("%.2f", totalBytes / 1024.0 / 1024.0 / 1024.0),
String.format("%.2f", (totalBytes / 1024.0 / 1024.0) / durationSeconds),
String.format("%.2f", (totalBytes * 8.0 / 1_000_000_000.0) / durationSeconds));
}
/**
......@@ -117,7 +120,7 @@ public class TcpSender {
socket.setTcpNoDelay(true);
socket.connect(new InetSocketAddress(host, port), 5000);
System.out.println("[连接成功] sender-" + connIndex + " -> " + host + ":" + port);
log.info("[连接成功] sender-{} -> {}:{}", connIndex, host, port);
OutputStream out = socket.getOutputStream();
while (running.get() && !Thread.currentThread().isInterrupted()) {
......@@ -125,7 +128,7 @@ public class TcpSender {
totalBytesSent.addAndGet(buffer.length);
}
} catch (Exception e) {
System.out.println("[连接异常] sender-" + connIndex + ": " + e.getMessage());
log.error("[连接异常] sender-{}", connIndex, e);
}
}
}
\ No newline at end of file
......@@ -3,6 +3,8 @@ package com.example.tcpreceiver;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
......@@ -17,6 +19,8 @@ import java.net.InetSocketAddress;
*/
public class ConnectionHandler extends SimpleChannelInboundHandler<ByteBuf> {
private static final Logger log = LoggerFactory.getLogger(ConnectionHandler.class);
private final GlobalStats globalStats;
private long bytesReceived = 0L;
......@@ -34,7 +38,7 @@ public class ConnectionHandler extends SimpleChannelInboundHandler<ByteBuf> {
connectionStartNanos = System.nanoTime();
InetSocketAddress remote = (InetSocketAddress) ctx.channel().remoteAddress();
remoteIp = remote.getAddress().getHostAddress();
System.out.println("[连接建立] " + remoteIp + ":" + remote.getPort());
log.info("[连接建立] {}:{}", remoteIp, remote.getPort());
}
@Override
......@@ -51,17 +55,19 @@ public class ConnectionHandler extends SimpleChannelInboundHandler<ByteBuf> {
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
));
log.info("[连接结束] {}, 耗时={}s, 接收字节数={} ({} MB), 平均速率={} MB/s",
remoteIp,
String.format("%.3f", durationSeconds),
bytesReceived,
String.format("%.2f", mbReceived),
String.format("%.2f", throughputMBps));
globalStats.recordConnectionFinished(remoteIp, bytesReceived);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
System.err.println("[连接异常] " + remoteIp + ": " + cause.getMessage());
log.error("[连接异常] {}", remoteIp, cause);
ctx.close();
}
}
package com.example.tcpreceiver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
import java.io.FileInputStream;
......@@ -17,6 +19,8 @@ import java.util.Map;
*/
public class ServerConfig {
private static final Logger log = LoggerFactory.getLogger(ServerConfig.class);
public int port = 24584;
public boolean useEpoll = true;
public int bossThreads = 1;
......@@ -55,20 +59,20 @@ public class ServerConfig {
if (externalPath != null && Files.exists(Paths.get(externalPath))) {
try (InputStream in = new FileInputStream(externalPath)) {
root = new Yaml().load(in);
System.out.println("[Config] 使用外部配置文件: " + externalPath);
log.info("[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");
log.info("[Config] 外部配置未指定或不存在,使用内置默认 config.yaml");
} else {
System.out.println("[Config] 未找到任何配置文件,使用代码内硬编码默认值");
log.info("[Config] 未找到任何配置文件,使用代码内硬编码默认值");
}
}
} catch (IOException e) {
System.out.println("[Config] 读取配置文件失败,使用默认值: " + e.getMessage());
log.warn("[Config] 读取配置文件失败,使用默认值", e);
}
if (root == null) {
......
......@@ -15,10 +15,9 @@ 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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* TCP接收服务启动类。
......@@ -32,6 +31,8 @@ import java.util.concurrent.TimeUnit;
*/
public class TcpServer {
private static final Logger log = LoggerFactory.getLogger(TcpServer.class);
public static void main(String[] args) throws InterruptedException {
new Thread(() -> {
try {
......@@ -42,11 +43,11 @@ public class TcpServer {
}).start();
String externalConfigPath = args.length > 0 ? args[0] : null;
ServerConfig cfg = ServerConfig.load(externalConfigPath);
System.out.println("[启动] 加载到的配置: " + cfg);
log.info("[启动] 加载到的配置: {}", cfg);
boolean useEpoll = cfg.useEpoll && Epoll.isAvailable();
if (cfg.useEpoll && !Epoll.isAvailable()) {
System.out.println("[启动] 配置要求使用epoll,但当前环境不支持,自动降级为NIO");
log.info("[启动] 配置要求使用epoll,但当前环境不支持,自动降级为NIO");
}
EventLoopGroup bossGroup;
......@@ -59,14 +60,14 @@ public class TcpServer {
? new EpollEventLoopGroup(cfg.workerThreads)
: new EpollEventLoopGroup();
channelClass = EpollServerSocketChannel.class;
System.out.println("[启动] 使用 Epoll 传输");
log.info("[启动] 使用 Epoll 传输");
} else {
bossGroup = new NioEventLoopGroup(cfg.bossThreads);
workerGroup = cfg.workerThreads > 0
? new NioEventLoopGroup(cfg.workerThreads)
: new NioEventLoopGroup();
channelClass = NioServerSocketChannel.class;
System.out.println("[启动] 使用 NIO 传输");
log.info("[启动] 使用 NIO 传输");
}
GlobalStats globalStats = new GlobalStats();
......@@ -112,7 +113,7 @@ public class TcpServer {
));
Channel serverChannel = bootstrap.bind(cfg.port).sync().channel();
System.out.println("[启动] 服务已启动,监听端口: " + cfg.port);
log.info("[启动] 服务已启动,监听端口: {}", cfg.port);
serverChannel.closeFuture().sync();
} finally {
......
......@@ -15,6 +15,8 @@ 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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.LocalTime;
import java.util.Set;
......@@ -27,6 +29,8 @@ import java.util.concurrent.atomic.AtomicLong;
* 客户端连接后全速循环下发固定十六进制报文,无连接时不发包、不消耗CPU
*/
public class NettyTcpPushServer {
private static final Logger log = LoggerFactory.getLogger(NettyTcpPushServer.class);
// ===================== 配置区 =====================
private static final int LISTEN_PORT = 27586;
// Netty 发送缓冲区高低水位,控制推送上限
......@@ -95,7 +99,7 @@ public class NettyTcpPushServer {
// 绑定端口启动服务
ChannelFuture bindFuture = bootstrap.bind(LISTEN_PORT).sync();
System.out.println("TCP推送服务启动成功,监听端口:" + LISTEN_PORT);
log.info("TCP推送服务启动成功,监听端口:{}", LISTEN_PORT);
bindFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
......@@ -122,7 +126,7 @@ public class NettyTcpPushServer {
public void channelActive(ChannelHandlerContext ctx) {
clientChannel = ctx.channel();
ONLINE_CHANNELS.add(clientChannel);
System.out.println("客户端接入:" + clientChannel.remoteAddress() + ",开始限速推送,目标 " + RATE_PER_SECOND + " 包/秒");
log.info("客户端接入:{},开始限速推送,目标 {} 包/秒", clientChannel.remoteAddress(), RATE_PER_SECOND);
startRateLimitedPush(ctx);
if(cfg.dataTime != 0) {
ctx.channel().eventLoop().scheduleWithFixedDelay(() -> {
......@@ -194,7 +198,7 @@ public class NettyTcpPushServer {
if (evt instanceof IdleStateEvent) {
IdleStateEvent e = (IdleStateEvent) evt;
if (e.state() == IdleState.WRITER_IDLE) {
System.out.println("写超时,可能对端消费异常,主动断开:" + ctx.channel().remoteAddress());
log.warn("写超时,可能对端消费异常,主动断开:{}", ctx.channel().remoteAddress());
ctx.close();
}
}
......@@ -206,13 +210,13 @@ public class NettyTcpPushServer {
Channel ch = ctx.channel();
ONLINE_CHANNELS.remove(ch);
stopPush();
System.out.println("客户端断开:" + ch.remoteAddress() + ",停止该通道推送");
log.info("客户端断开:{},停止该通道推送", ch.remoteAddress());
}
/** 异常直接关闭通道 */
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
log.error("通道异常,关闭连接:{}", ctx.channel().remoteAddress(), cause);
ctx.close();
}
}
......
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