自定義通訊協議傳輸protobuf

阿新 • • 發佈：2021-02-09

自定義協議

本次我們使用的是datalength+databytes 的自定義長度的協議
datalength：就是後面databytes的長度
databytes：protobuf 3.0協議資料序列化後的資料內容，長度不限

操作流程

客戶端：將資料按照通訊協議封裝，推送給服務端

服務端：服務端接收到資料後，按照協議解析出資料

服務端程式碼Demo

public class ServerTestService {
	private static Socket client;
    private static void listen() {
        try {
            DataInputStream dis = 
 new DataInputStream(client.getInputStream());
            DataOutputStream dos = new DataOutputStream(client.getOutputStream());
            int i= 0;
            while (true) {
                i++;
                int length = dis.readInt();
                byte[] message = new byte[length];
                dis. 
readFully(message);
                ImScrmWwProto.TransportMessage transportMessage = ImScrmWwProto.TransportMessage.parseFrom(message);
                System.out.println("服務端[" + i + "]接收：length: " + length + " message: "+transportMessage);
                dos.writeInt 
(length);
                dos.write(message);
                dos.flush();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @SneakyThrows
    public static void main(String[] args) {
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress("127.0.0.1", 8090));
        client = serverSocket.accept();
        listen();
    }
}

客戶端程式碼

public class ClientMessageService {
 private static Socket socket;

    public static void main(String[] args) {
        try {
            socket = new Socket("127.0.0.1",8090);
            SendMessageRequest request = SendMessageRequest.builder()
                    .id(1)
                    .accessToken("accessToken11111111111")
                    .messageType(1001)
                    .requestMessageId(1222222)
                    .content("any object")
                    .build();
            send(request);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void send(SendMessageRequest request) {

        ImScrmWwProto.TransportMessage.Builder transportMessageBuild = ImScrmWwProto.TransportMessage.newBuilder();
        transportMessageBuild.setId(request.getId());
        transportMessageBuild.setAccessToken(request.getAccessToken());
        transportMessageBuild.setMsgType(MessageTypeConverter.of(request.getMessageType()));
        transportMessageBuild.setRefMessageId(request.getRequestMessageId());
        transportMessageBuild.setContent(Any.pack(transportMessageBuild.build()));

        ImScrmWwProto.TransportMessage transportMessage = transportMessageBuild.build();
        byte[] payload = transportMessage.toByteArray();
        try {
            int i = 0;
            while (true) {
                i++;
                System.out.println("客戶端["+i + "]傳送 length：" + payload.length + " message: " + transportMessage);
                DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
                dos.writeInt(payload.length);
                dos.write(payload);
                dos.flush();
                Thread.sleep(10000);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}