用Java 队列 结合 rgrpc 实现文件传输的server端
时间: 2024-05-06 18:20:01 浏览: 56
以下是使用Java队列结合gRPC实现文件传输的server端的示例代码:
```java
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class FileTransferServer {
private static final int PORT = 9000;
public static void main(String[] args) throws IOException, InterruptedException {
Server server = ServerBuilder.forPort(PORT)
.addService(new FileTransferImpl())
.build();
server.start();
System.out.println("Server started, listening on " + PORT);
server.awaitTermination();
}
static class FileTransferImpl extends FileTransferGrpc.FileTransferImplBase {
private BlockingQueue<FileChunk> queue = new LinkedBlockingQueue<>();
private FileOutputStream outputStream;
private String fileName;
private long fileSize;
@Override
public void uploadFile(FileName request, StreamObserver<FileStatus> responseObserver) {
fileName = request.getName();
fileSize = request.getSize();
try {
outputStream = new FileOutputStream(fileName);
} catch (IOException e) {
responseObserver.onError(e);
}
FileStatus fileStatus = FileStatus.newBuilder().setStatus(FileStatus.Status.STARTED).build();
responseObserver.onNext(fileStatus);
new Thread(() -> {
while (true) {
try {
FileChunk fileChunk = queue.take();
if (fileChunk.getData().equals(ByteString.EMPTY)) {
break;
}
outputStream.write(fileChunk.getData().toByteArray());
} catch (IOException | InterruptedException e) {
responseObserver.onError(e);
break;
}
}
try {
outputStream.close();
} catch (IOException e) {
responseObserver.onError(e);
}
FileStatus fileStatus = FileStatus.newBuilder().setStatus(FileStatus.Status.COMPLETED).build();
responseObserver.onNext(fileStatus);
responseObserver.onCompleted();
}).start();
}
@Override
public void uploadFileChunk(FileChunk request, StreamObserver<ChunkStatus> responseObserver) {
ChunkStatus chunkStatus = ChunkStatus.newBuilder().setStatus(ChunkStatus.Status.RECEIVED).build();
responseObserver.onNext(chunkStatus);
try {
queue.put(request);
} catch (InterruptedException e) {
responseObserver.onError(e);
}
}
@Override
public void uploadFileEnd(Empty request, StreamObserver<FileStatus> responseObserver) {
FileChunk fileChunk = FileChunk.newBuilder().setData(ByteString.EMPTY).build();
try {
queue.put(fileChunk);
} catch (InterruptedException e) {
responseObserver.onError(e);
}
}
}
}
```
在上面的代码中,我们定义了一个`FileTransferImpl`类,它实现了gRPC生成的`FileTransferGrpc.FileTransferImplBase`接口。在这个类中,我们创建了一个阻塞队列`queue`来缓存文件块数据,并创建了一个`FileOutputStream`对象来将文件块数据写入到磁盘上。我们使用了一个线程来从队列中取出文件块数据,并将其写入到磁盘上。
在`uploadFile`方法中,我们先获取了客户端传来的文件名和文件大小,并创建了一个`FileOutputStream`对象来将文件数据写入到磁盘上。然后,我们向客户端发送了一个`FileStatus`对象,表示文件传输已经开始。
在`uploadFileChunk`方法中,我们将客户端传来的文件块数据放入到队列中,并向客户端发送了一个`ChunkStatus`对象,表示已经接收到了该文件块数据。
在`uploadFileEnd`方法中,我们向队列中放入了一个`FileChunk`对象,它的数据字段设为了一个空的`ByteString`对象,表示文件传输已经结束。
最后,我们在`main`方法中启动了gRPC服务器,并将实现了接口的`FileTransferImpl`类添加到服务器中。当客户端连接到服务器时,服务器会向客户端发送一个`FileStatus`对象,表示文件传输已经开始。然后,当客户端传输完所有的文件块数据后,服务器会向客户端发送一个`FileStatus`对象,表示文件传输已经完成。
相关推荐
![pdf](https://img-home.csdnimg.cn/images/20210720083512.png)
![pdf](https://img-home.csdnimg.cn/images/20210720083512.png)
![pdf](https://img-home.csdnimg.cn/images/20210720083512.png)
![-](https://img-home.csdnimg.cn/images/20210720083327.png)
![-](https://img-home.csdnimg.cn/images/20210720083327.png)
![-](https://img-home.csdnimg.cn/images/20210720083327.png)
![-](https://img-home.csdnimg.cn/images/20210720083327.png)
![-](https://img-home.csdnimg.cn/images/20210720083327.png)
![-](https://csdnimg.cn/download_wenku/file_type_column_c1.png)
![-](https://csdnimg.cn/download_wenku/file_type_column_c1.png)
![-](https://csdnimg.cn/download_wenku/file_type_column_c1.png)
![-](https://csdnimg.cn/download_wenku/file_type_column_c1.png)
![-](https://csdnimg.cn/download_wenku/file_type_column_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)