当前位置：首页 > news >正文

java实现Windows 命名管道：千万别在读取时“死等”

news 2026/5/12 13:40:38

1.代码背景

博主想实现一个跨平台通信机制来连接LSP服务端，想通过unix domian socket来实现这个。在java实现客户端连接时，出现了问题，JDK 原生搞不定 Windows 命名管道，Java 的网络库只认 Socket 家族，不认“文件”家族。所以想直接访问这个通道，就必须用 Kernel32 。

2. 问题分析

描述：在 Windows 上，如果用一个线程 ReadFile 阻塞等数据，主线程想 WriteFile 发消息时会直接卡死。博主在测试过程中，发现能收到服务端的通知，但是在发送消息后，服务端无法收到。经过判断，是在发送时卡住，然后研究出具体原因。由此记录一下过程
原因：同步模式下，读和写共用同一个句柄锁。读不完，锁不放，写不进。
避坑指南：在 Windows 环境下，必须先“轮询”再“读取” 。
- 使用 PeekNamedPipe 检查是否有数据。
- 没数据就 Thread.sleep(20) ，让出句柄锁。
选择同步的理由
- 代码简洁度：重叠 I/O 在 Java/JNA 里的实现非常臃肿，容易引入内存泄漏或指针错误。
- 兼容性：LSP 服务端通常一次只允许一个连接，双句柄方案可能会导致服务端报错“管道忙”。

3.代码实现

packageorg.example.lspClient;importcom.sun.jna.platform.win32.Kernel32;importcom.sun.jna.platform.win32.WinBase;importcom.sun.jna.platform.win32.WinNT;importcom.sun.jna.ptr.IntByReference;importorg.tinylog.Logger;importjava.io.*;importjava.net.StandardProtocolFamily;importjava.net.UnixDomainSocketAddress;importjava.nio.ByteBuffer;importjava.nio.channels.SocketChannel;importjava.nio.charset.StandardCharsets;importjava.nio.file.Path;importjava.util.Arrays;importjava.util.function.Consumer;/** * 跨平台 UDS 传输层，具有 LSP 协议感知能力 */publicclassUdsTransportimplementsCloseable{privatefinalbooleanisWindows;privateWinNT.HANDLE hPipe;privateSocketChannelchannel;privatefinalByteArrayOutputStreambuffer=newByteArrayOutputStream();privateConsumer<String>messageListener;privatevolatilebooleanrunning=true;publicUdsTransport(StringsocketPath)throwsIOException{Stringos=System.getProperty("os.name").toLowerCase();this.isWindows=os.contains("win");if(isWindows){connectWindowsPipe(socketPath);}else{connectUnixSocket(socketPath);}startReadingThread();}publicvoidsetOnMessage(Consumer<String>listener){this.messageListener=listener;}privatevoidconnectWindowsPipe(StringpipeName)throwsIOException{this.hPipe=Kernel32.INSTANCE.CreateFile(pipeName,WinNT.GENERIC_READ|WinNT.GENERIC_WRITE,0,null,WinNT.OPEN_EXISTING,0,null);if(WinBase.INVALID_HANDLE_VALUE.equals(hPipe)){interror=Kernel32.INSTANCE.GetLastError();thrownewIOException("Failed to connect to Named Pipe: "+pipeName+" (Error: "+error+")");}}privatevoidconnectUnixSocket(StringsocketPath)throwsIOException{this.channel=SocketChannel.open(StandardProtocolFamily.UNIX);this.channel.connect(UnixDomainSocketAddress.of(Path.of(socketPath)));}publicvoidsendMessage(StringjsonContent)throwsIOException{byte[]body=jsonContent.getBytes(StandardCharsets.UTF_8);Stringheader="Content-Length: "+body.length+"\r\n\r\n";writeRaw(header.getBytes(StandardCharsets.US_ASCII));writeRaw(body);}privatevoidwriteRaw(byte[]bytes)throwsIOException{if(isWindows){IntByReferencewritten=newIntByReference();booleansuccess=Kernel32.INSTANCE.WriteFile(hPipe,bytes,bytes.length,written,null);if(!success){interr=Kernel32.INSTANCE.GetLastError();thrownewIOException("WriteFile failed with error: "+err);}if(written.getValue()<bytes.length){Logger.warn("Partial write detected: "+written.getValue()+"/"+bytes.length);return;}}else{ByteBuffernioBuffer=ByteBuffer.wrap(bytes);while(nioBuffer.hasRemaining()){intwritten=channel.write(nioBuffer);if(written==0){thrownewIOException("Channel write stalled");}}}Logger.info("Wrote bytes to: "+Arrays.toString(bytes));}privatevoidstartReadingThread(){Threadthread=newThread(()->{byte[]readBuffer=newbyte[8192];try{while(running){intn=readToBuffer(readBuffer);if(n>0){synchronized(buffer){buffer.write(readBuffer,0,n);}processBuffer();}else{// 给WriteFile 竞争句柄锁的机会Thread.sleep(20);}}}catch(Exceptione){if(running)System.err.println("Transport read error: "+e.getMessage());}},"UdsTransport-Reader");thread.setDaemon(true);thread.start();}privateintreadToBuffer(byte[]target)throwsIOException{if(isWindows){IntByReferenceavail=newIntByReference();// 轮询是否有数据booleanpeekOk=Kernel32.INSTANCE.PeekNamedPipe(hPipe,null,0,null,avail,null);if(!peekOk){interr=Kernel32.INSTANCE.GetLastError();if(err==109)thrownewEOFException("Pipe closed");return0;}if(avail.getValue()==0){return0;// 当前没数据，返回 0 让出 CPU 和句柄锁}IntByReferenceread=newIntByReference();// 此时 ReadFile 不会阻塞，因为它已经知道有数据了booleansuccess=Kernel32.INSTANCE.ReadFile(hPipe,target,target.length,read,null);if(!success){interr=Kernel32.INSTANCE.GetLastError();if(err==109)thrownewEOFException("Pipe closed");if(err==234)returntarget.length;// ERROR_MORE_DATAthrownewIOException("ReadFile failed: "+err);}returnread.getValue();}else{ByteBuffernioBuf=ByteBuffer.wrap(target);intn=channel.read(nioBuf);if(n==-1)thrownewEOFException("Socket closed");returnn;}}privatevoidprocessBuffer(){synchronized(buffer){while(true){byte[]data=buffer.toByteArray();if(data.length==0)break;StringtempStr=newString(data,StandardCharsets.US_ASCII);intheaderEnd=tempStr.indexOf("\r\n\r\n");if(headerEnd==-1)break;intcontentLength=-1;String[]lines=tempStr.substring(0,headerEnd).split("\r\n");for(Stringline:lines){if(line.toLowerCase().startsWith("content-length:")){contentLength=Integer.parseInt(line.substring(15).trim());}}if(contentLength==-1)break;inttotalSize=headerEnd+4+contentLength;if(data.length<totalSize)break;byte[]bodyBytes=newbyte[contentLength];System.arraycopy(data,headerEnd+4,bodyBytes,0,contentLength);Stringmessage=newString(bodyBytes,StandardCharsets.UTF_8);// 异步回调if(messageListener!=null){messageListener.accept(message);}byte[]remaining=newbyte[data.length-totalSize];System.arraycopy(data,totalSize,remaining,0,remaining.length);buffer.reset();try{buffer.write(remaining);}catch(IOExceptionignored){}}}}@Overridepublicvoidclose()throwsIOException{running=false;if(isWindows){if(hPipe!=null)Kernel32.INSTANCE.CloseHandle(hPipe);}else{if(channel!=null)channel.close();}}}