简介基于P2P的局域网即时通信系统Java版是一份面向Java课程设计、网络编程实训或毕业设计参考的完整项目资源适合想深入理解P2P模型与Socket编程的中高级学习者。资源围绕图形界面的局域网消息系统展开程序同时扮演服务器与客户端服务端固定占用3000端口包含用户注册与分组、扫描网段内在线对等方、对等方列表动态获取与应答、通过TCP连接收发消息和文件等功能消息格式由开发者自定义涵盖用户名、IP地址等必要信息。界面部分涉及对等方列表、消息显示区、输入框及文件传输进程控制能完整体现P2P通信的流程。压缩包整体约1.34MBzip格式轻量易用目前已有603人学习适合正在完成类似课设选题、需要借鉴整体架构与编码思路的学习者。读者可以从中学习对等方发现协议设计、在线用户列表管理、多线程收发处理并快速迁移到自己的项目中。1. 局域网里的P2P即时通信到底在解决什么问题很多公司内部团队早上第一件事是打开自己熟悉的聊天工具但消息要绕到云端再绕回来中间只要断一次同事间的“在吗”就要变成排障工单。换一个思路在内网里每台机器本来就是一个可用的节点为什么不能让消息在两个节点之间直连这就是P2P即时通信的核心——不设中心服务器每个节点同时扮演客户端和服务端通过局域网广播发现彼此用TCP连接传递消息。这个标题里的“基于P2P的局域网即时通信系统Java版”通常对应这样的场景内网无外网、不允许部署中心服务器或者只是想在开发机上做一个轻量工具。Java在这里不是必要条件但用它实现的好处是跨平台、并发模型成熟Swing/JavaFX又能快速拼出聊天界面。本文要做的就是把这个系统的骨架拆开节点发现、消息收发、状态保活、可靠性保证并给出能直接编译运行的Java代码结构。适合对Java网络编程、多线程、Socket有一定基础想自己动手搭一套内网聊天工具的工程师。2. 设计节点发现与消息协议UDP广播 TCP单播P2P系统第一步是“找到对方”。在局域网内最常规的方案是UDP负责节点发现TCP负责实际消息传输。这一章先把协议定下来再给出最小可用的发现代码。2.1 为什么节点发现用UDP而不是TCPTCP是面向连接的字节流通信双方必须提前知道对方的IP和端口才能建立连接。而P2P场景下节点是动态加入的你根本不知道此刻有哪些机器在线所以需要一个“喊一嗓子”的机制。UDP广播正好能做这件事向子网广播地址例如192.168.1.255发送一个分组子网内所有监听该端口的主机都能收到。但这不代表UDP只会被用于发现。如果你用UDP直接传聊天内容就要自己处理分片、乱序和丢包而TCP把这些问题都解决了且内网质量足够好TCP握手开销完全可以接受。所以我的技术选型是UDP广播发送hello和world报文双方交换IP后再用TCP长连接传递chat消息。这已经在很多开源的局域网聊天工具里被验证过是“常见做法”里非常稳妥的一条。需要注意UDP广播只能覆盖同一广播域。如果公司把网络划分了多个VLAN或者开启了交换机隔离广播就会被限制在某个小范围里。此时需要升级方案比如引入一个“引导节点”保存在线列表或者改用组播地址。但这个标题限定在局域网我默认按单子网场景来写。2.2 自研信令报文的JSON格式节点发现和消息传输需要共用一套报文结构。Java生态里Jackson和Gson都很好用JSON本身调试时也能直接打印观察。我一般会定义下面这张表里的字段字段类型必填说明typestring是报文类型hello / world / chat / ack / online / offlinefromstring是发送者用户名tostring是接收者用户名群发时为 allmsgIdstring是UUID用于消息去重和ACK关联contentstring是文本内容timestamplong是毫秒时间戳“hello”表示新节点加入并向外广播“world”是收到hello的节点回发自己的身份这样双方就建立了映射关系。“chat”是真正的聊天数据里面同样带msgId。“ack”是接收方对某条msgId的确认。这张表里最关键的是msgId后续去重、重发、ACK关联都靠它来串联。2.3 节点发现的最小Java实现下面这段代码会向局域网广播一条hello报文并在同一端口监听其他节点的回复。先看发送端// 1. 用独立DatagramSocket发送广播hello try (DatagramSocket socket new DatagramSocket()) { socket.setBroadcast(true); // 必须开启广播权限 JSONObject hello new JSONObject(); hello.put(type, hello); hello.put(from, username); hello.put(msgId, UUID.randomUUID().toString()); // broadcastTarget通常是192.168.x.255简化处理 byte[] bytes hello.toString().getBytes(StandardCharsets.UTF_8); DatagramPacket packet new DatagramPacket(bytes, bytes.length, InetAddress.getByName(broadcastTarget), 8899); socket.send(packet); } catch (IOException e) { e.printStackTrace(); }这段代码里有两个参数需要重点解释广播地址broadcastTarget必须指向子网的定向广播地址Windows和Linux下常见写法是192.168.1.255不要写成255.255.255.255后者在某些路由器上不会转发端口8899需要与接收端保持一致且需要确保操作系统防火墙放行UDP包。try-with-resources会在send后立刻关闭Socket这里只是为了演示生产建议用一个常驻Socket反复发送。接收端是常驻监听线程// 2. 接收hello/world/chat报文 DatagramSocket recvSocket new DatagramSocket(8899); byte[] buf new byte[8192]; // 包体预估不超过8KB DatagramPacket packet new DatagramPacket(buf, buf.length); while (running) { recvSocket.receive(packet); // 阻塞等待 String json new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); JSONObject msg JSONObject.parseObject(json); // 根据msg的type分发处理并记录来源地址 handleMessage(msg, packet.getAddress()); }receive是阻塞方法所以这段代码必须跑在独立线程里。buf大小8192是经验值比以太网MTU(1500)大几个分片如果报文体超过这个值就会在底层被丢弃Java层看不到任何异常。所有收到的报文都要带着packet.getAddress()因为UDP包没有连接你后续要回world或发TCP握手都需要知道对方IP。3. 用 Java NIO 还是传统 BIO先跑通最小可通信节点发现节点拿到IP后下一步就是建立TCP连接传递消息。这里常见的选型矛盾是该用NIO还是BIO3.1 局域网即时通信场景下不需要NIONIO的核心优势是Selector可以让一个线程管理成百上千个连接适合高并发、连接数多的场景。但局域网即时通信的节点数量通常是个位数到几十个一台开发机上开几十个线程完全没有压力。BIO的模型简单每个连接一个线程读写阻塞但代码直观调试容易。Java里的虚拟线程在JDK 21之后也能缓解线程开销但同样的逻辑在传统BIO里写起来也一样清晰。我给出的结论这个标题的项目用BIO 固定线程池而不是NIO。原因有几点一是代码可读性BIO的网络代码几乎人人能看懂后续接手的人维护成本低二是消息模型是短连接还是长连接聊天场景需要收对方的主动消息所以每个点对点连接应该是长连接BIO长连接一样能稳定跑三是NIO的Selector在处理“半包粘包”时还要配合ByteBuffer和自定义协议边界而本文用行分隔符就能规避。当然如果你的需求要扩展到几百个节点或者需要承载和多个客户端同时连接那时候再把传输层改为Netty而不是直接用原生NIO。Netty已经把拆包、重连、心跳框架都做好了比你自己堆Selector靠谱得多。3.2 一个可复现的TCP消息收发类下面是一个最小的TCP服务端它监听每个接入的Socket在线程池里处理消息。这段代码的核心是accept阻塞接收连接每个连接交给独立线程处理处理完的消息如果type是chat就转发给目标用户。public class TcpMessageServer { private final ServerSocket serverSocket; private final ExecutorService workerPool; private boolean running true; public TcpMessageServer(int port, int poolSize) throws IOException { this.serverSocket new ServerSocket(port); this.workerPool Executors.newFixedThreadPool(poolSize); } public void start() { while (running) { try { Socket socket serverSocket.accept(); workerPool.submit(() - handleClient(socket)); } catch (IOException e) { if (running) e.printStackTrace(); } } } private void handleClient(Socket socket) { try (BufferedReader reader new BufferedReader(new InputStreamReader( socket.getInputStream(), StandardCharsets.UTF_8)); PrintWriter writer new PrintWriter(socket.getOutputStream(), true)) { String line; while ((line reader.readLine()) ! null) { JSONObject msg JSONObject.parseObject(line); String type msg.getString(type); if (chat.equals(type)) { // 转发给目标节点或直接处理 routeMessage(msg); // 回ACK JSONObject ack new JSONObject(); ack.put(type, ack); ack.put(msgId, msg.getString(msgId)); writer.println(ack.toJSONString()); } } } catch (IOException e) { // 单个连接异常不影响其他连接 } } }这里有一个容易被新手忽略的参数PrintWriter的第二个参数autoFlush必须传入true否则println不会立即发送数据消息会积压在缓冲区里。读者能看到我用了readLine这要求发送方在发送JSON时必须换行这是最简单可靠的应用层消息边界。3.3 连接管理在线用户表的维护P2P系统里每个节点都要保存一份在线用户表。这个表的作用是收到消息时能找到对应的Socket收到离线通知时能清理连接。我通常这样定义public class PeerRegistry { // key用户名value目标节点的IP和Socket private final MapString, Socket socketMap new ConcurrentHashMap(); private final MapString, String userIpMap new ConcurrentHashMap(); public void register(String username, String ip, Socket socket) { socketMap.put(username, socket); userIpMap.put(username, ip); } public void offline(String username) { socketMap.remove(username); userIpMap.remove(username); } }ConcurrentHashMap保证多线程环境下读写安全因为消息收发和心跳检测可能同时操作这张表。用户名的唯一性由协议层保证例如使用主机名拼接随机数的方式生成。当A要传消息给B时A先从本机的PeerRegistry里查到B的IP和Socket如果没有就说明B在线但没有与A建立直接连接这时可以退化到UDP TODO不局域网里一般建议用用户表里B的IP直接发起TCP连接。需要注意的是P2P连接并不要求完全网格化。可以在节点之间建立“按需连接”A和B聊天才建立连接不聊天就只维持UDP发现的在线状态。这样可以减少连接数也更符合P2P的自治特征。4. 在线状态、心跳与消息ACK从“能通”到“可靠”两道程序之间能互发消息只是第一步真正的即时通信系统要解决节点掉线、消息丢失的问题。这一章实现心跳机制、ACK确认和重发。4.1 用UDP或TCP做心跳检测最常见的做法是每隔几秒向已知在线节点发送一条typeheart的JSON报文。心跳可以走UDP因为它不关心响应顺序只需知道对方是否还活着也可以走TCP在已有连接上发送心跳可以同时保活连接。如果5秒发送一次心跳连续2次没有收到回应就把对方标记为离线。这个阈值要结合局域网情况调整Wi-Fi环境丢包率高建议把超时放宽到15秒有线局域网可以压缩到10秒。下面是心跳发送的代码使用ScheduledExecutorService周期执行。ScheduledExecutorService scheduler Executors.newScheduledThreadPool(1); scheduler.scheduleAtFixedRate(() - { JSONObject ping new JSONObject(); ping.put(type, heart); ping.put(from, username); // 遍历已知节点通过TCP连接发送 for (PeerInfo peer : peerRegistry.getAllPeers()) { sendTcpMessage(peer.getIp(), peer.getPort(), ping.toJSONString()); } }, 0, 5, TimeUnit.SECONDS);sendTcpMessage内部复用了已有Socket还是新建连接取决于你的连接策略。如果每个节点持有与其他节点的长连接那就直接在该Socket上输出如果采用按需连接这里就会变成“发送消息时临时建连”。频繁创建TCP连接会带来端口和句柄开销所以我倾向于维护一个连接池复用空闲连接。4.2 ACK确认与超时重发TCP底层虽然有ACK但应用层的ACK仍然是必要的。为什么因为TCP只能保证字节流到达对端的内核栈但无法保证对端进程已经成功解析并写入UI层。如果聊天消息在界面上没有展示用户就会认为“没收到”。所以发送方收到typeack的确认后才真正确定消息已消费。实现ACK机制时发送方需要为每条chat消息保存一个状态public class OutgoingMessage { private String msgId; private String peerUser; private String content; private long sendTime; private int retryCount; private volatile boolean acked; }发送后启动一个定时任务检查acked状态超时未确认则重发public void resendUnackedMessages() { for (OutgoingMessage m : pendingMessages.values()) { if (!m.isAcked() m.getRetryCount() 3) { sendTcpMessage(m.getPeerUser(), m.getContent()); m.incrementRetryCount(); m.setSendTime(System.currentTimeMillis()); } } }这里的三次重试是经验值重试次数太多会反复占用带宽太少在无线网络下又容易丢失。重试间隔可以设置为2秒、4秒、8秒指数退避的方式比固定5秒更友好因为它能避免对端短时间内收到多份重复消息造成尾包炸裂。4.3 消息去重与离线消息的处理重发机制会带来重复消息所以接收方必须根据msgId去重。维护一个最近的msgId队列只要看到已存在的msgId就丢弃private final SetString seenMsgIds Collections.newSetFromMap( new LinkedHashMapString, Boolean(1024, 0.75f, true) { Override protected boolean removeEldestEntry(Map.Entry eldest) { return size() 2048; // 只保留最近2048条 } });这里用LinkedHashMap的accessOrdertrue实现LRU当msgId超过2048个就淘汰最旧的避免内存膨胀。需要特别说明的是这个去重队列只适合短会话场景如果长时间运行建议把消息持久化到本地数据库去重的key改为“msgId sender”。离线消息是P2P模式最难处理的问题。因为节点之间点对点直接通信发送者无法确认目标是否在线时对方显然收不到消息。在这个标题的定位下有三种常见做法发送方在发消息前询问PeerRegistry里目标是否存在不存在就提示“对方离线消息未发送”或者借助一个可选的中继节点缓存消息但这会引入半中心化架构或者在局域网环境下依赖“在线率”足够高直接丢弃离线消息。第三点虽然粗暴但在内部工具脚本里反而是最常见的大家认为“你都离线了上线自然会看到我之前发的公告”。本文倾向于第一种发送前检查在线状态。5. 多客户端联测的验证方法和三个最容易踩的坑5.1 用三个终端验证系统是否正常验证P2P通信系统是否正常不能在同一个机器上模拟两个进程就结束。至少需要两台物理机或一台物理机加两个虚拟机因为Windows和Linux对UDP广播的处理不同。我一般在机器A上启动一个节点绑定端口8899机器B上启动另一个节点绑定端口8899然后用第三个终端发送一条chat消息观察两个节点的日志窗口。建议先做“最小验证”在A上用netstat -an | grep 8899确认UDP和TCP端口都在监听在B上用tcpdump -i eth0 udp port 8899捕获广播包确认hello报文能穿过网络。如果B捕获不到广播包不用继续查应用代码直接看交换机或防火墙配置。5.2 坑一广播地址写错导致节点互相看不见很多人直接写255.255.255.255作为广播地址这在部分Windows环境可以工作但在Linux和跨网卡环境下会失败。正确做法是从网卡接口获取广播地址或者用IP和子网掩码按位运算byte[] ip InetAddress.getLocalHost().getAddress(); byte[] subnet {0xffff, 0xff, 0xff, 0x00}; // 24位子网掩码 byte[] broadcast new byte[4]; for (int i 0; i 4; i) { broadcast[i] (byte) (ip[i] | ~subnet[i]); }这样算出来的广播地址才是目标的子网广播地址。如果你在使用无线网卡建议打印出本机IP和子网掩码确认。5.3 坑二防火墙拦截UDP广播或TCP端口UDP广播经常被Windows防火墙的“网络发现”规则拦截。你可以给Java程序手动添加 firewall 例外也可以暂时关闭防火墙测试只在内网可信环境。TCP端口如果被占用或未放行连接尝试会超时。这里有一个快速判断方法在发送节点上用telnet 目标IP 8898测试TCP端口是否可达如果telnet超时先处理防火墙再查代码。5.4 坑三心跳线程没考虑反射到UI线程Java Swing聊天界面要求所有UI更新必须在Event Dispatch ThreadEDT执行。如果你在心跳线程里直接调用textarea.append()轻则界面卡顿重则抛出未检查异常导致线程死亡。常见做法是维护一个阻塞队列心跳与UI线程间接通信BlockingQueueJSONObject eventQueue new LinkedBlockingQueue(); // 心跳线程处理消息时发送到队列 eventQueue.put(msg); // EDT线程定时拉取队列并更新界面 SwingUtilities.invokeLater(() - { JSONObject m eventQueue.poll(); if (m ! null) textArea.append(m.getString(from) : m.getString(content) \n); });这个队列同时也承担了解耦作用即使底层网络报文的接收线程崩溃UI线程依然能持续读取最后收到的数据。对于P2P这种节点间直连的架构队列隔离是保证长时间稳定运行的关键细节值得在最开始设计时就放进代码里。本文还有配套的精品资源点击获取 基于P2P的局域网即时通信Java实现：从UDP广播到TCP消息
简介基于P2P的局域网即时通信系统Java版是一份面向Java课程设计、网络编程实训或毕业设计参考的完整项目资源适合想深入理解P2P模型与Socket编程的中高级学习者。资源围绕图形界面的局域网消息系统展开程序同时扮演服务器与客户端服务端固定占用3000端口包含用户注册与分组、扫描网段内在线对等方、对等方列表动态获取与应答、通过TCP连接收发消息和文件等功能消息格式由开发者自定义涵盖用户名、IP地址等必要信息。界面部分涉及对等方列表、消息显示区、输入框及文件传输进程控制能完整体现P2P通信的流程。压缩包整体约1.34MBzip格式轻量易用目前已有603人学习适合正在完成类似课设选题、需要借鉴整体架构与编码思路的学习者。读者可以从中学习对等方发现协议设计、在线用户列表管理、多线程收发处理并快速迁移到自己的项目中。1. 局域网里的P2P即时通信到底在解决什么问题很多公司内部团队早上第一件事是打开自己熟悉的聊天工具但消息要绕到云端再绕回来中间只要断一次同事间的“在吗”就要变成排障工单。换一个思路在内网里每台机器本来就是一个可用的节点为什么不能让消息在两个节点之间直连这就是P2P即时通信的核心——不设中心服务器每个节点同时扮演客户端和服务端通过局域网广播发现彼此用TCP连接传递消息。这个标题里的“基于P2P的局域网即时通信系统Java版”通常对应这样的场景内网无外网、不允许部署中心服务器或者只是想在开发机上做一个轻量工具。Java在这里不是必要条件但用它实现的好处是跨平台、并发模型成熟Swing/JavaFX又能快速拼出聊天界面。本文要做的就是把这个系统的骨架拆开节点发现、消息收发、状态保活、可靠性保证并给出能直接编译运行的Java代码结构。适合对Java网络编程、多线程、Socket有一定基础想自己动手搭一套内网聊天工具的工程师。2. 设计节点发现与消息协议UDP广播 TCP单播P2P系统第一步是“找到对方”。在局域网内最常规的方案是UDP负责节点发现TCP负责实际消息传输。这一章先把协议定下来再给出最小可用的发现代码。2.1 为什么节点发现用UDP而不是TCPTCP是面向连接的字节流通信双方必须提前知道对方的IP和端口才能建立连接。而P2P场景下节点是动态加入的你根本不知道此刻有哪些机器在线所以需要一个“喊一嗓子”的机制。UDP广播正好能做这件事向子网广播地址例如192.168.1.255发送一个分组子网内所有监听该端口的主机都能收到。但这不代表UDP只会被用于发现。如果你用UDP直接传聊天内容就要自己处理分片、乱序和丢包而TCP把这些问题都解决了且内网质量足够好TCP握手开销完全可以接受。所以我的技术选型是UDP广播发送hello和world报文双方交换IP后再用TCP长连接传递chat消息。这已经在很多开源的局域网聊天工具里被验证过是“常见做法”里非常稳妥的一条。需要注意UDP广播只能覆盖同一广播域。如果公司把网络划分了多个VLAN或者开启了交换机隔离广播就会被限制在某个小范围里。此时需要升级方案比如引入一个“引导节点”保存在线列表或者改用组播地址。但这个标题限定在局域网我默认按单子网场景来写。2.2 自研信令报文的JSON格式节点发现和消息传输需要共用一套报文结构。Java生态里Jackson和Gson都很好用JSON本身调试时也能直接打印观察。我一般会定义下面这张表里的字段字段类型必填说明typestring是报文类型hello / world / chat / ack / online / offlinefromstring是发送者用户名tostring是接收者用户名群发时为 allmsgIdstring是UUID用于消息去重和ACK关联contentstring是文本内容timestamplong是毫秒时间戳“hello”表示新节点加入并向外广播“world”是收到hello的节点回发自己的身份这样双方就建立了映射关系。“chat”是真正的聊天数据里面同样带msgId。“ack”是接收方对某条msgId的确认。这张表里最关键的是msgId后续去重、重发、ACK关联都靠它来串联。2.3 节点发现的最小Java实现下面这段代码会向局域网广播一条hello报文并在同一端口监听其他节点的回复。先看发送端// 1. 用独立DatagramSocket发送广播hello try (DatagramSocket socket new DatagramSocket()) { socket.setBroadcast(true); // 必须开启广播权限 JSONObject hello new JSONObject(); hello.put(type, hello); hello.put(from, username); hello.put(msgId, UUID.randomUUID().toString()); // broadcastTarget通常是192.168.x.255简化处理 byte[] bytes hello.toString().getBytes(StandardCharsets.UTF_8); DatagramPacket packet new DatagramPacket(bytes, bytes.length, InetAddress.getByName(broadcastTarget), 8899); socket.send(packet); } catch (IOException e) { e.printStackTrace(); }这段代码里有两个参数需要重点解释广播地址broadcastTarget必须指向子网的定向广播地址Windows和Linux下常见写法是192.168.1.255不要写成255.255.255.255后者在某些路由器上不会转发端口8899需要与接收端保持一致且需要确保操作系统防火墙放行UDP包。try-with-resources会在send后立刻关闭Socket这里只是为了演示生产建议用一个常驻Socket反复发送。接收端是常驻监听线程// 2. 接收hello/world/chat报文 DatagramSocket recvSocket new DatagramSocket(8899); byte[] buf new byte[8192]; // 包体预估不超过8KB DatagramPacket packet new DatagramPacket(buf, buf.length); while (running) { recvSocket.receive(packet); // 阻塞等待 String json new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); JSONObject msg JSONObject.parseObject(json); // 根据msg的type分发处理并记录来源地址 handleMessage(msg, packet.getAddress()); }receive是阻塞方法所以这段代码必须跑在独立线程里。buf大小8192是经验值比以太网MTU(1500)大几个分片如果报文体超过这个值就会在底层被丢弃Java层看不到任何异常。所有收到的报文都要带着packet.getAddress()因为UDP包没有连接你后续要回world或发TCP握手都需要知道对方IP。3. 用 Java NIO 还是传统 BIO先跑通最小可通信节点发现节点拿到IP后下一步就是建立TCP连接传递消息。这里常见的选型矛盾是该用NIO还是BIO3.1 局域网即时通信场景下不需要NIONIO的核心优势是Selector可以让一个线程管理成百上千个连接适合高并发、连接数多的场景。但局域网即时通信的节点数量通常是个位数到几十个一台开发机上开几十个线程完全没有压力。BIO的模型简单每个连接一个线程读写阻塞但代码直观调试容易。Java里的虚拟线程在JDK 21之后也能缓解线程开销但同样的逻辑在传统BIO里写起来也一样清晰。我给出的结论这个标题的项目用BIO 固定线程池而不是NIO。原因有几点一是代码可读性BIO的网络代码几乎人人能看懂后续接手的人维护成本低二是消息模型是短连接还是长连接聊天场景需要收对方的主动消息所以每个点对点连接应该是长连接BIO长连接一样能稳定跑三是NIO的Selector在处理“半包粘包”时还要配合ByteBuffer和自定义协议边界而本文用行分隔符就能规避。当然如果你的需求要扩展到几百个节点或者需要承载和多个客户端同时连接那时候再把传输层改为Netty而不是直接用原生NIO。Netty已经把拆包、重连、心跳框架都做好了比你自己堆Selector靠谱得多。3.2 一个可复现的TCP消息收发类下面是一个最小的TCP服务端它监听每个接入的Socket在线程池里处理消息。这段代码的核心是accept阻塞接收连接每个连接交给独立线程处理处理完的消息如果type是chat就转发给目标用户。public class TcpMessageServer { private final ServerSocket serverSocket; private final ExecutorService workerPool; private boolean running true; public TcpMessageServer(int port, int poolSize) throws IOException { this.serverSocket new ServerSocket(port); this.workerPool Executors.newFixedThreadPool(poolSize); } public void start() { while (running) { try { Socket socket serverSocket.accept(); workerPool.submit(() - handleClient(socket)); } catch (IOException e) { if (running) e.printStackTrace(); } } } private void handleClient(Socket socket) { try (BufferedReader reader new BufferedReader(new InputStreamReader( socket.getInputStream(), StandardCharsets.UTF_8)); PrintWriter writer new PrintWriter(socket.getOutputStream(), true)) { String line; while ((line reader.readLine()) ! null) { JSONObject msg JSONObject.parseObject(line); String type msg.getString(type); if (chat.equals(type)) { // 转发给目标节点或直接处理 routeMessage(msg); // 回ACK JSONObject ack new JSONObject(); ack.put(type, ack); ack.put(msgId, msg.getString(msgId)); writer.println(ack.toJSONString()); } } } catch (IOException e) { // 单个连接异常不影响其他连接 } } }这里有一个容易被新手忽略的参数PrintWriter的第二个参数autoFlush必须传入true否则println不会立即发送数据消息会积压在缓冲区里。读者能看到我用了readLine这要求发送方在发送JSON时必须换行这是最简单可靠的应用层消息边界。3.3 连接管理在线用户表的维护P2P系统里每个节点都要保存一份在线用户表。这个表的作用是收到消息时能找到对应的Socket收到离线通知时能清理连接。我通常这样定义public class PeerRegistry { // key用户名value目标节点的IP和Socket private final MapString, Socket socketMap new ConcurrentHashMap(); private final MapString, String userIpMap new ConcurrentHashMap(); public void register(String username, String ip, Socket socket) { socketMap.put(username, socket); userIpMap.put(username, ip); } public void offline(String username) { socketMap.remove(username); userIpMap.remove(username); } }ConcurrentHashMap保证多线程环境下读写安全因为消息收发和心跳检测可能同时操作这张表。用户名的唯一性由协议层保证例如使用主机名拼接随机数的方式生成。当A要传消息给B时A先从本机的PeerRegistry里查到B的IP和Socket如果没有就说明B在线但没有与A建立直接连接这时可以退化到UDP TODO不局域网里一般建议用用户表里B的IP直接发起TCP连接。需要注意的是P2P连接并不要求完全网格化。可以在节点之间建立“按需连接”A和B聊天才建立连接不聊天就只维持UDP发现的在线状态。这样可以减少连接数也更符合P2P的自治特征。4. 在线状态、心跳与消息ACK从“能通”到“可靠”两道程序之间能互发消息只是第一步真正的即时通信系统要解决节点掉线、消息丢失的问题。这一章实现心跳机制、ACK确认和重发。4.1 用UDP或TCP做心跳检测最常见的做法是每隔几秒向已知在线节点发送一条typeheart的JSON报文。心跳可以走UDP因为它不关心响应顺序只需知道对方是否还活着也可以走TCP在已有连接上发送心跳可以同时保活连接。如果5秒发送一次心跳连续2次没有收到回应就把对方标记为离线。这个阈值要结合局域网情况调整Wi-Fi环境丢包率高建议把超时放宽到15秒有线局域网可以压缩到10秒。下面是心跳发送的代码使用ScheduledExecutorService周期执行。ScheduledExecutorService scheduler Executors.newScheduledThreadPool(1); scheduler.scheduleAtFixedRate(() - { JSONObject ping new JSONObject(); ping.put(type, heart); ping.put(from, username); // 遍历已知节点通过TCP连接发送 for (PeerInfo peer : peerRegistry.getAllPeers()) { sendTcpMessage(peer.getIp(), peer.getPort(), ping.toJSONString()); } }, 0, 5, TimeUnit.SECONDS);sendTcpMessage内部复用了已有Socket还是新建连接取决于你的连接策略。如果每个节点持有与其他节点的长连接那就直接在该Socket上输出如果采用按需连接这里就会变成“发送消息时临时建连”。频繁创建TCP连接会带来端口和句柄开销所以我倾向于维护一个连接池复用空闲连接。4.2 ACK确认与超时重发TCP底层虽然有ACK但应用层的ACK仍然是必要的。为什么因为TCP只能保证字节流到达对端的内核栈但无法保证对端进程已经成功解析并写入UI层。如果聊天消息在界面上没有展示用户就会认为“没收到”。所以发送方收到typeack的确认后才真正确定消息已消费。实现ACK机制时发送方需要为每条chat消息保存一个状态public class OutgoingMessage { private String msgId; private String peerUser; private String content; private long sendTime; private int retryCount; private volatile boolean acked; }发送后启动一个定时任务检查acked状态超时未确认则重发public void resendUnackedMessages() { for (OutgoingMessage m : pendingMessages.values()) { if (!m.isAcked() m.getRetryCount() 3) { sendTcpMessage(m.getPeerUser(), m.getContent()); m.incrementRetryCount(); m.setSendTime(System.currentTimeMillis()); } } }这里的三次重试是经验值重试次数太多会反复占用带宽太少在无线网络下又容易丢失。重试间隔可以设置为2秒、4秒、8秒指数退避的方式比固定5秒更友好因为它能避免对端短时间内收到多份重复消息造成尾包炸裂。4.3 消息去重与离线消息的处理重发机制会带来重复消息所以接收方必须根据msgId去重。维护一个最近的msgId队列只要看到已存在的msgId就丢弃private final SetString seenMsgIds Collections.newSetFromMap( new LinkedHashMapString, Boolean(1024, 0.75f, true) { Override protected boolean removeEldestEntry(Map.Entry eldest) { return size() 2048; // 只保留最近2048条 } });这里用LinkedHashMap的accessOrdertrue实现LRU当msgId超过2048个就淘汰最旧的避免内存膨胀。需要特别说明的是这个去重队列只适合短会话场景如果长时间运行建议把消息持久化到本地数据库去重的key改为“msgId sender”。离线消息是P2P模式最难处理的问题。因为节点之间点对点直接通信发送者无法确认目标是否在线时对方显然收不到消息。在这个标题的定位下有三种常见做法发送方在发消息前询问PeerRegistry里目标是否存在不存在就提示“对方离线消息未发送”或者借助一个可选的中继节点缓存消息但这会引入半中心化架构或者在局域网环境下依赖“在线率”足够高直接丢弃离线消息。第三点虽然粗暴但在内部工具脚本里反而是最常见的大家认为“你都离线了上线自然会看到我之前发的公告”。本文倾向于第一种发送前检查在线状态。5. 多客户端联测的验证方法和三个最容易踩的坑5.1 用三个终端验证系统是否正常验证P2P通信系统是否正常不能在同一个机器上模拟两个进程就结束。至少需要两台物理机或一台物理机加两个虚拟机因为Windows和Linux对UDP广播的处理不同。我一般在机器A上启动一个节点绑定端口8899机器B上启动另一个节点绑定端口8899然后用第三个终端发送一条chat消息观察两个节点的日志窗口。建议先做“最小验证”在A上用netstat -an | grep 8899确认UDP和TCP端口都在监听在B上用tcpdump -i eth0 udp port 8899捕获广播包确认hello报文能穿过网络。如果B捕获不到广播包不用继续查应用代码直接看交换机或防火墙配置。5.2 坑一广播地址写错导致节点互相看不见很多人直接写255.255.255.255作为广播地址这在部分Windows环境可以工作但在Linux和跨网卡环境下会失败。正确做法是从网卡接口获取广播地址或者用IP和子网掩码按位运算byte[] ip InetAddress.getLocalHost().getAddress(); byte[] subnet {0xffff, 0xff, 0xff, 0x00}; // 24位子网掩码 byte[] broadcast new byte[4]; for (int i 0; i 4; i) { broadcast[i] (byte) (ip[i] | ~subnet[i]); }这样算出来的广播地址才是目标的子网广播地址。如果你在使用无线网卡建议打印出本机IP和子网掩码确认。5.3 坑二防火墙拦截UDP广播或TCP端口UDP广播经常被Windows防火墙的“网络发现”规则拦截。你可以给Java程序手动添加 firewall 例外也可以暂时关闭防火墙测试只在内网可信环境。TCP端口如果被占用或未放行连接尝试会超时。这里有一个快速判断方法在发送节点上用telnet 目标IP 8898测试TCP端口是否可达如果telnet超时先处理防火墙再查代码。5.4 坑三心跳线程没考虑反射到UI线程Java Swing聊天界面要求所有UI更新必须在Event Dispatch ThreadEDT执行。如果你在心跳线程里直接调用textarea.append()轻则界面卡顿重则抛出未检查异常导致线程死亡。常见做法是维护一个阻塞队列心跳与UI线程间接通信BlockingQueueJSONObject eventQueue new LinkedBlockingQueue(); // 心跳线程处理消息时发送到队列 eventQueue.put(msg); // EDT线程定时拉取队列并更新界面 SwingUtilities.invokeLater(() - { JSONObject m eventQueue.poll(); if (m ! null) textArea.append(m.getString(from) : m.getString(content) \n); });这个队列同时也承担了解耦作用即使底层网络报文的接收线程崩溃UI线程依然能持续读取最后收到的数据。对于P2P这种节点间直连的架构队列隔离是保证长时间稳定运行的关键细节值得在最开始设计时就放进代码里。本文还有配套的精品资源点击获取