面向服務的體系架構 SOA（一） --- 基於TCP、HTTP協議的RPC

1.1 基於TCP協議的RPC

1.1.1 RPC名詞解釋

　　RPC的全稱是Remote Process Call，即遠程過程調用，RPC的實現包括客戶端和服務端，即服務調用方和服務提供方。服務調用方發送RPC請求到服務提供方，服務提供方根據請求的參數執行請求方法，並將結果返回給服務調用方，一次RPC調用完成。

1.1.2 對象的序列化

　　在網絡上傳輸的數據，無論何種類型，最終都需要轉化為二進制流。在面向對象的程序設計中，客戶端將對象轉化為二進制流發送給服務端，服務端接收數據后將二進制流轉化為對象，java中將這兩種轉化方式稱為對象的序列化和反序列化。下面介紹java內置的序列化方式和基於java的Hessian序列化方式：

java內置的序列化和反序列化關鍵代碼：

　　

        //序列化操作
        Person person = new Person();
        ByteArrayOutputStream os = new ByteArrayOutputStream();
        ObjectOutputStream out = new ObjectOutputStream(os);
        out.writeObject(person);
        byte[] byteArray = os.toByteArray();
        
        //反序列化操作
        ByteArrayInputStream is = new ByteArrayInputStream(byteArray);
        ObjectInputStream in = new ObjectInputStream(is);
        Person newPerson = new Person();
        newPerson = (Person) in.readObject();

基於java的Hessian序列化和反序列化關鍵代碼：

        //序列化操作
        ByteArrayOutputStream osH = new ByteArrayOutputStream();
        HessianOutput outH = new HessianOutput(osH);
        outH.writeObject(person);
        byte[] byteArrayH = osH.toByteArray();
        
        //反序列化操作
        ByteArrayInputStream isH = new ByteArrayInputStream(byteArrayH);
        HessianInput inH = new HessianInput(isH);
        newPerson = (Person) inH.readObject();

1.1.3 基於TCP協議實現RPC

我們利用java的SocketAPI實現一個簡單的RPC調用，服務的接口和實現比較簡單，根據傳入的參數來判斷返回"hello" or "bye bye"。

public interface SayHelloService {

    public String sayHello(String arg);
}

public class SayHelloServiceImpl implements SayHelloService {

    public String sayHello(String arg) {
        return "hello".equals(arg) ? "hello" : "bye bye";
    }
    
}

服務消費者Consumer類：

/**
 * 基於TCP協議實現RPC -- 服務消費者
 * @author admin
 *
 */
public class Consumer {
    
    public static void main(String[] args) throws Exception {
        //接口名稱
        String interfaceName = SayHelloService.class.getName();
        //需要執行遠程的方法
        Method method = SayHelloService.class.getMethod("sayHello", String.class);
        //傳遞到遠程的參數
        Object [] arguments = {"hello"};
        Socket socket = new Socket("127.0.0.1", 1234);
        //將方法名和參數傳遞到遠端
        ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
        out.writeUTF(interfaceName);//接口名稱
        out.writeUTF(method.getName());//方法名稱
        out.writeObject(method.getParameterTypes());//方法參數類型
        out.writeObject(arguments);//傳遞的參數
        System.out.println("發送信息到服務端，發送的信息為:" + arguments[0]);
        //從遠端讀取返回結果
        ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
        String result = (String) in.readObject();
        System.out.println("服務返回的結果為：" + result); 
    }
}

服務提供者Provider類：

/**
 * 基於TCP協議實現RPC -- 服務提供者
 * @author admin
 *
 */
public class Provider {
    
    public static void main(String[] args) throws Exception {
        ServerSocket server = new ServerSocket(1234);
        Map<Object, Object> services = new HashMap<Object, Object>();  
        services.put(SayHelloService.class.getName(), new SayHelloServiceImpl());
        while(true) {
            System.out.println("服務提供者啟動，等待客戶端調用…………"); 
            Socket socket = server.accept();
            //讀取服務信息
            ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
            String interfaceName = in.readUTF();
            String methodName = in.readUTF();
            Class<?>[] parameterTypes = (Class<?>[]) in.readObject();
            Object [] arguments = (Object[]) in.readObject();
            System.out.println("客戶端調用服務端接口" + interfaceName + "的" + methodName + "方法");
            //執行調用
            Class serviceClass = Class.forName(interfaceName);//得到接口的class
            Object service = services.get(interfaceName);//取得服務實現的對象
            Method method = serviceClass.getMethod(methodName, parameterTypes);//獲得要調用的方法
            Object result = method.invoke(service, arguments);
            ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
            out.writeObject(result);
            System.out.println("服務端返回結果為：" + result);
        }
    }
}

在真實的生產環境中往往是多個客戶端同時請求服務端，服務端則需要同時接收和處理多個客戶端請求消息，涉及並發處理、服務路由、負載均衡等現實問題，以上代碼顯然不能完成。

 

1.2 基於HTTP協議的RPC

1.2.1 HTTP協議棧

HTTP的全稱是HyperText Transfer Protocol，即超文本傳輸協議，當今普遍采用的版本是HTTP1.1。HTTP協議屬於應用層協議，它構建在TCP和IP協議之上，處於TCP/IP架構的頂端，為了更好的理解HTTP協議，我們基於java的SocketAPI設計一個簡單的應用層通信協議，來窺探協議實現的一些過程與細節。

客戶端向服務端發送一條命令，服務端接收到命令后，會判斷命令是否為"HELLO"，若是則返回客戶端"hello!",否則返回客戶端"bye bye"。

/**
 * 協議請求
 * 
 * @author admin
 *
 */
public class Request {

    /**
     * 協議編碼 0：GBK；1：UTF-8
     */
    private byte encode;
    /**
     * 命令
     */
    private String command;
    /**
     * 命令長度
     */
    private int commandLength;

    public byte getEncode() {
        return encode;
    }

    public void setEncode(byte encode) {
        this.encode = encode;
    }

    public String getCommand() {
        return command;
    }

    public void setCommand(String command) {
        this.command = command;
    }

    public int getCommandLength() {
        return commandLength;
    }

    public void setCommandLength(int commandLength) {
        this.commandLength = commandLength;
    }

}

/**
 * 協議響應
 * 
 * @author admin
 *
 */
public class Response {
    /**
     * 編碼
     */
    private byte encode;
    /**
     * 響應
     */
    private String response;
    /**
     * 響應長度
     */
    private int responseLength;

    public byte getEncode() {
        return encode;
    }

    public void setEncode(byte encode) {
        this.encode = encode;
    }

    public String getResponse() {
        return response;
    }

    public void setResponse(String response) {
        this.response = response;
    }

    public int getResponseLength() {
        return responseLength;
    }

    public void setResponseLength(int responseLength) {
        this.responseLength = responseLength;
    }

    @Override
    public String toString() {
        return "Response [encode=" + encode + ", response=" + response + ", responseLength=" + responseLength + "]";
    }
    
}

客戶端發送以及服務端響應處理代碼：

/**
 * 服務端
 * @author admin
 *
 */
public class Server {
    
    public static void main(String[] args) throws Exception {
        ServerSocket server = new ServerSocket(1234);
        while(true) {
            Socket client = server.accept();
            //讀取請求數據
            Request request = ProtocolUtil.readRequest(client.getInputStream());
            //封裝響應數據
            Response response = new Response();
            response.setEncode(Encode.UTF8.getValue());
            response.setResponse(request.getCommand().equals("HELLO") ? "hello!" : "bye bye");
            response.setResponseLength(response.getResponse().length());
            //響應到客戶端
            ProtocolUtil.writeResponse(client.getOutputStream(), response);
        }
    }
}

/**
 * 客戶端
 * @author admin
 *
 */
public class Client {
    
    public static void main(String[] args) throws Exception {
        //組裝請求數據
        Request request = new Request();
        request.setCommand("HELLO");
        request.setCommandLength(request.getCommand().length());
        request.setEncode(Encode.UTF8.getValue());
        Socket client = new Socket("127.0.0.1", 1234);
        //發送請求
        ProtocolUtil.writeRequest(client.getOutputStream(), request);
        //讀取相應
        Response response = ProtocolUtil.readResponse(client.getInputStream());
        System.out.println(response);
    }
}

ProtocolUtil 類：

public class ProtocolUtil {
    
    public static void writeRequest(OutputStream out, Request request) {
        try {
            out.write(request.getEncode());
            //write一個int值會截取其低8位傳輸，丟棄其高24位，因此需要將基本類型轉化為字節流
            //java采用Big Endian字節序，而所有的網絡協議也都是以Big Endian字節序來進行傳輸，所以再進行數據的傳輸和接收時，需要先將數據轉化成Big Endian字節序
            //out.write(request.getCommandLength());
            out.write(int2ByteArray(request.getCommandLength()));
            out.write(Encode.GBK.getValue() == request.getEncode() ? request.getCommand().getBytes("GBK") : request.getCommand().getBytes("UTF8"));
            out.flush();
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
    }
    
    /**
     * 將響應輸出到客戶端
     * @param os
     * @param response
     */
    public static void writeResponse(OutputStream out, Response response) {
        try {
            out.write(response.getEncode());
            out.write(int2ByteArray(response.getResponseLength()));
            out.write(Encode.GBK.getValue() == response.getEncode() ? response.getResponse().getBytes("GBK") : response.getResponse().getBytes("UTF8"));
            out.flush();
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
    }
    
    public static Request readRequest(InputStream is) {
        Request request = new Request();
        try {
            //讀取編碼
            byte [] encodeByte = new byte[1];
            is.read(encodeByte);
            byte encode = encodeByte[0];
            //讀取命令長度
            byte [] commandLengthByte = new byte[4];//緩沖區
            is.read(commandLengthByte);
            int commandLength = byte2Int(commandLengthByte);
            //讀取命令
            byte [] commandByte = new byte[commandLength];
            is.read(commandByte);
            String command = Encode.GBK.getValue() == encode ? new String(commandByte, "GBK") : new String(commandByte, "UTF8");
            //組裝請求返回
            request.setEncode(encode);
            request.setCommand(command);
            request.setCommandLength(commandLength);
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
        return request;
    }
    
    public static Response readResponse(InputStream is) {
        Response response = new Response();
        try {
            byte [] encodeByte = new byte[1];
            is.read(encodeByte);
            byte encode = encodeByte[0];
            byte [] responseLengthByte = new byte[4];
            is.read(responseLengthByte);
            int commandLength = byte2Int(responseLengthByte);
            byte [] responseByte = new byte[commandLength];
            is.read(responseByte);
            String resContent = Encode.GBK.getValue() == encode ? new String(responseByte, "GBK") : new String(responseByte, "UTF8");
            response.setEncode(encode);
            response.setResponse(resContent);
            response.setResponseLength(commandLength);
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
        return response;
    }
    
    public static int byte2Int(byte [] bytes) {
        int num = bytes[3] & 0xFF;
        num |= ((bytes[2] << 8) & 0xFF00);
        num |= ((bytes[1] << 16) & 0xFF0000);
        num |= ((bytes[0] << 24) & 0xFF000000);
        return num;
    }
    
    public static byte[] int2ByteArray(int i) {
        byte [] result = new byte[4];
        result[0] = (byte) ((i >> 24) & 0xFF);
        result[1] = (byte) ((i >> 16) & 0xFF);
        result[2] = (byte) ((i >> 8) & 0xFF);
        result[3] = (byte) (i & 0xFF);
        return result;
    }
    
}

1.2.2 HTTP請求與響應

下圖是HTTP請求與響應的過程步驟，在此不詳細贅述。

1.2.3 通過HttpClient發送HTTP請求

HttpClient對HTTP協議通信的過程進行了封裝，下面是簡單的通過HttpClient發送HTTP GET請求，並獲取服務端響應的代碼：

　　　　  //url前加上http協議頭，標明該請求為http請求
        String url = "https://www.baidu.com";
        //組裝請求
        HttpClient httpClient = new DefaultHttpClient();
        HttpGet httpGet = new HttpGet(url);
        //接收響應
        HttpResponse response = httpClient.execute(httpGet);
        HttpEntity entity = response.getEntity();
        byte[] byteArray = EntityUtils.toByteArray(entity);
        String result = new String(byteArray, "utf8");
        System.out.println(result);

1.2.4 使用HTTP協議的優勢

　　隨着請求規模的擴展，基於TCP協議的RPC的實現，需要考慮多線程並發、鎖、I/O等復雜的底層細節，在大流量高並發的壓力下，任何一個小的錯誤都可能被無限放大，最終導致程序宕機。而對於基於HTTP協議的實現來說，很多成熟的開源web容易已經幫其處理好了這些事情，如Apache，Tomcat，Jboss等，開發人員可將更多的精力集中在業務實現上，而非處理底層細節。