欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

面向服务的体系架构 SOA(一) --- 基于TCP、HTTP协议的RPC

程序员文章站 2022-03-10 19:28:32
1.1 基于TCP协议的RPC 1.1.1 RPC名词解释 RPC的全称是Remote Process Call,即远程过程调用,RPC的实现包括客户端和服务端,即服务调用方和服务提供方。服务调用方发送RPC请求到服务提供方,服务提供方根据请求的参数执行请求方法,并将结果返回给服务调用方,一次RPC ......

1.1 基于TCP协议的RPC

1.1.1 RPC名词解释

  RPC的全称是Remote Process Call,即远程过程调用,RPC的实现包括客户端和服务端,即服务调用方和服务提供方。服务调用方发送RPC请求到服务提供方,服务提供方根据请求的参数执行请求方法,并将结果返回给服务调用方,一次RPC调用完成。

1.1.2 对象的序列化

  在网络上传输的数据,无论何种类型,最终都需要转化为二进制流。在面向对象的程序设计中,客户端将对象转化为二进制流发送给服务端,服务端接收数据后将二进制流转化为对象,java中将这两种转化方式称为对象的序列化和反序列化。下面介绍java内置的序列化方式和基于java的Hessian序列化方式:

java内置的序列化和反序列化关键代码:

  

 1         //序列化操作
 2         Person person = new Person();
 3         ByteArrayOutputStream os = new ByteArrayOutputStream();
 4         ObjectOutputStream out = new ObjectOutputStream(os);
 5         out.writeObject(person);
 6         byte[] byteArray = os.toByteArray();
 7         
 8         //反序列化操作
 9         ByteArrayInputStream is = new ByteArrayInputStream(byteArray);
10         ObjectInputStream in = new ObjectInputStream(is);
11         Person newPerson = new Person();
12         newPerson = (Person) in.readObject();

基于java的Hessian序列化和反序列化关键代码:

 1         //序列化操作
 2         ByteArrayOutputStream osH = new ByteArrayOutputStream();
 3         HessianOutput outH = new HessianOutput(osH);
 4         outH.writeObject(person);
 5         byte[] byteArrayH = osH.toByteArray();
 6         
 7         //反序列化操作
 8         ByteArrayInputStream isH = new ByteArrayInputStream(byteArrayH);
 9         HessianInput inH = new HessianInput(isH);
10         newPerson = (Person) inH.readObject();

1.1.3 基于TCP协议实现RPC

我们利用java的SocketAPI实现一个简单的RPC调用,服务的接口和实现比较简单,根据传入的参数来判断返回"hello" or "bye bye"。

 1 public interface SayHelloService {
 2 
 3     public String sayHello(String arg);
 4 }
 5 
 6 public class SayHelloServiceImpl implements SayHelloService {
 7 
 8     public String sayHello(String arg) {
 9         return "hello".equals(arg) ? "hello" : "bye bye";
10     }
11     
12 }

服务消费者Consumer类:

 1 /**
 2  * 基于TCP协议实现RPC -- 服务消费者
 3  * @author admin
 4  *
 5  */
 6 public class Consumer {
 7     
 8     public static void main(String[] args) throws Exception {
 9         //接口名称
10         String interfaceName = SayHelloService.class.getName();
11         //需要执行远程的方法
12         Method method = SayHelloService.class.getMethod("sayHello", String.class);
13         //传递到远程的参数
14         Object [] arguments = {"hello"};
15         Socket socket = new Socket("127.0.0.1", 1234);
16         //将方法名和参数传递到远端
17         ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
18         out.writeUTF(interfaceName);//接口名称
19         out.writeUTF(method.getName());//方法名称
20         out.writeObject(method.getParameterTypes());//方法参数类型
21         out.writeObject(arguments);//传递的参数
22         System.out.println("发送信息到服务端,发送的信息为:" + arguments[0]);
23         //从远端读取返回结果
24         ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
25         String result = (String) in.readObject();
26         System.out.println("服务返回的结果为:" + result); 
27     }
28 }

服务提供者Provider类:

 1 /**
 2  * 基于TCP协议实现RPC -- 服务提供者
 3  * @author admin
 4  *
 5  */
 6 public class Provider {
 7     
 8     public static void main(String[] args) throws Exception {
 9         ServerSocket server = new ServerSocket(1234);
10         Map<Object, Object> services = new HashMap<Object, Object>();  
11         services.put(SayHelloService.class.getName(), new SayHelloServiceImpl());
12         while(true) {
13             System.out.println("服务提供者启动,等待客户端调用…………"); 
14             Socket socket = server.accept();
15             //读取服务信息
16             ObjectInputStream in = new ObjectInputStream(socket.getInputStream());
17             String interfaceName = in.readUTF();
18             String methodName = in.readUTF();
19             Class<?>[] parameterTypes = (Class<?>[]) in.readObject();
20             Object [] arguments = (Object[]) in.readObject();
21             System.out.println("客户端调用服务端接口" + interfaceName + "的" + methodName + "方法");
22             //执行调用
23             Class serviceClass = Class.forName(interfaceName);//得到接口的class
24             Object service = services.get(interfaceName);//取得服务实现的对象
25             Method method = serviceClass.getMethod(methodName, parameterTypes);//获得要调用的方法
26             Object result = method.invoke(service, arguments);
27             ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream());
28             out.writeObject(result);
29             System.out.println("服务端返回结果为:" + result);
30         }
31     }
32 }

在真实的生产环境中往往是多个客户端同时请求服务端,服务端则需要同时接收和处理多个客户端请求消息,涉及并发处理、服务路由、负载均衡等现实问题,以上代码显然不能完成。

 

1.2 基于HTTP协议的RPC

1.2.1 HTTP协议栈

HTTP的全称是HyperText Transfer Protocol,即超文本传输协议,当今普遍采用的版本是HTTP1.1。HTTP协议属于应用层协议,它构建在TCP和IP协议之上,处于TCP/IP架构的顶端,为了更好的理解HTTP协议,我们基于java的SocketAPI设计一个简单的应用层通信协议,来窥探协议实现的一些过程与细节。

面向服务的体系架构 SOA(一) --- 基于TCP、HTTP协议的RPC

客户端向服务端发送一条命令,服务端接收到命令后,会判断命令是否为"HELLO",若是则返回客户端"hello!",否则返回客户端"bye bye"。

 1 /**
 2  * 协议请求
 3  * 
 4  * @author admin
 5  *
 6  */
 7 public class Request {
 8 
 9     /**
10      * 协议编码 0:GBK;1:UTF-8
11      */
12     private byte encode;
13     /**
14      * 命令
15      */
16     private String command;
17     /**
18      * 命令长度
19      */
20     private int commandLength;
21 
22     public byte getEncode() {
23         return encode;
24     }
25 
26     public void setEncode(byte encode) {
27         this.encode = encode;
28     }
29 
30     public String getCommand() {
31         return command;
32     }
33 
34     public void setCommand(String command) {
35         this.command = command;
36     }
37 
38     public int getCommandLength() {
39         return commandLength;
40     }
41 
42     public void setCommandLength(int commandLength) {
43         this.commandLength = commandLength;
44     }
45 
46 }
 1 /**
 2  * 协议响应
 3  * 
 4  * @author admin
 5  *
 6  */
 7 public class Response {
 8     /**
 9      * 编码
10      */
11     private byte encode;
12     /**
13      * 响应
14      */
15     private String response;
16     /**
17      * 响应长度
18      */
19     private int responseLength;
20 
21     public byte getEncode() {
22         return encode;
23     }
24 
25     public void setEncode(byte encode) {
26         this.encode = encode;
27     }
28 
29     public String getResponse() {
30         return response;
31     }
32 
33     public void setResponse(String response) {
34         this.response = response;
35     }
36 
37     public int getResponseLength() {
38         return responseLength;
39     }
40 
41     public void setResponseLength(int responseLength) {
42         this.responseLength = responseLength;
43     }
44 
45     @Override
46     public String toString() {
47         return "Response [encode=" + encode + ", response=" + response + ", responseLength=" + responseLength + "]";
48     }
49     
50 }

客户端发送以及服务端响应处理代码:

 1 /**
 2  * 服务端
 3  * @author admin
 4  *
 5  */
 6 public class Server {
 7     
 8     public static void main(String[] args) throws Exception {
 9         ServerSocket server = new ServerSocket(1234);
10         while(true) {
11             Socket client = server.accept();
12             //读取请求数据
13             Request request = ProtocolUtil.readRequest(client.getInputStream());
14             //封装响应数据
15             Response response = new Response();
16             response.setEncode(Encode.UTF8.getValue());
17             response.setResponse(request.getCommand().equals("HELLO") ? "hello!" : "bye bye");
18             response.setResponseLength(response.getResponse().length());
19             //响应到客户端
20             ProtocolUtil.writeResponse(client.getOutputStream(), response);
21         }
22     }
23 }
24 
25 /**
26  * 客户端
27  * @author admin
28  *
29  */
30 public class Client {
31     
32     public static void main(String[] args) throws Exception {
33         //组装请求数据
34         Request request = new Request();
35         request.setCommand("HELLO");
36         request.setCommandLength(request.getCommand().length());
37         request.setEncode(Encode.UTF8.getValue());
38         Socket client = new Socket("127.0.0.1", 1234);
39         //发送请求
40         ProtocolUtil.writeRequest(client.getOutputStream(), request);
41         //读取相应
42         Response response = ProtocolUtil.readResponse(client.getInputStream());
43         System.out.println(response);
44     }
45 }

ProtocolUtil 类:

 1 public class ProtocolUtil {
 2     
 3     public static void writeRequest(OutputStream out, Request request) {
 4         try {
 5             out.write(request.getEncode());
 6             //write一个int值会截取其低8位传输,丢弃其高24位,因此需要将基本类型转化为字节流
 7             //java采用Big Endian字节序,而所有的网络协议也都是以Big Endian字节序来进行传输,所以再进行数据的传输和接收时,需要先将数据转化成Big Endian字节序
 8             //out.write(request.getCommandLength());
 9             out.write(int2ByteArray(request.getCommandLength()));
10             out.write(Encode.GBK.getValue() == request.getEncode() ? request.getCommand().getBytes("GBK") : request.getCommand().getBytes("UTF8"));
11             out.flush();
12         } catch (Exception e) {
13             System.err.println(e.getMessage());
14         }
15     }
16     
17     /**
18      * 将响应输出到客户端
19      * @param os
20      * @param response
21      */
22     public static void writeResponse(OutputStream out, Response response) {
23         try {
24             out.write(response.getEncode());
25             out.write(int2ByteArray(response.getResponseLength()));
26             out.write(Encode.GBK.getValue() == response.getEncode() ? response.getResponse().getBytes("GBK") : response.getResponse().getBytes("UTF8"));
27             out.flush();
28         } catch (Exception e) {
29             System.err.println(e.getMessage());
30         }
31     }
32     
33     public static Request readRequest(InputStream is) {
34         Request request = new Request();
35         try {
36             //读取编码
37             byte [] encodeByte = new byte[1];
38             is.read(encodeByte);
39             byte encode = encodeByte[0];
40             //读取命令长度
41             byte [] commandLengthByte = new byte[4];//缓冲区
42             is.read(commandLengthByte);
43             int commandLength = byte2Int(commandLengthByte);
44             //读取命令
45             byte [] commandByte = new byte[commandLength];
46             is.read(commandByte);
47             String command = Encode.GBK.getValue() == encode ? new String(commandByte, "GBK") : new String(commandByte, "UTF8");
48             //组装请求返回
49             request.setEncode(encode);
50             request.setCommand(command);
51             request.setCommandLength(commandLength);
52         } catch (Exception e) {
53             System.err.println(e.getMessage());
54         }
55         return request;
56     }
57     
58     public static Response readResponse(InputStream is) {
59         Response response = new Response();
60         try {
61             byte [] encodeByte = new byte[1];
62             is.read(encodeByte);
63             byte encode = encodeByte[0];
64             byte [] responseLengthByte = new byte[4];
65             is.read(responseLengthByte);
66             int commandLength = byte2Int(responseLengthByte);
67             byte [] responseByte = new byte[commandLength];
68             is.read(responseByte);
69             String resContent = Encode.GBK.getValue() == encode ? new String(responseByte, "GBK") : new String(responseByte, "UTF8");
70             response.setEncode(encode);
71             response.setResponse(resContent);
72             response.setResponseLength(commandLength);
73         } catch (Exception e) {
74             System.err.println(e.getMessage());
75         }
76         return response;
77     }
78     
79     public static int byte2Int(byte [] bytes) {
80         int num = bytes[3] & 0xFF;
81         num |= ((bytes[2] << 8) & 0xFF00);
82         num |= ((bytes[1] << 16) & 0xFF0000);
83         num |= ((bytes[0] << 24) & 0xFF000000);
84         return num;
85     }
86     
87     public static byte[] int2ByteArray(int i) {
88         byte [] result = new byte[4];
89         result[0] = (byte) ((i >> 24) & 0xFF);
90         result[1] = (byte) ((i >> 16) & 0xFF);
91         result[2] = (byte) ((i >> 8) & 0xFF);
92         result[3] = (byte) (i & 0xFF);
93         return result;
94     }
95     
96 }

1.2.2 HTTP请求与响应

下图是HTTP请求与响应的过程步骤,在此不详细赘述。

面向服务的体系架构 SOA(一) --- 基于TCP、HTTP协议的RPC

1.2.3 通过HttpClient发送HTTP请求

HttpClient对HTTP协议通信的过程进行了封装,下面是简单的通过HttpClient发送HTTP GET请求,并获取服务端响应的代码:

 1       //url前加上http协议头,标明该请求为http请求
 2         String url = "https://www.baidu.com";
 3         //组装请求
 4         HttpClient httpClient = new DefaultHttpClient();
 5         HttpGet httpGet = new HttpGet(url);
 6         //接收响应
 7         HttpResponse response = httpClient.execute(httpGet);
 8         HttpEntity entity = response.getEntity();
 9         byte[] byteArray = EntityUtils.toByteArray(entity);
10         String result = new String(byteArray, "utf8");
11         System.out.println(result);

1.2.4 使用HTTP协议的优势

  随着请求规模的扩展,基于TCP协议的RPC的实现,需要考虑多线程并发、锁、I/O等复杂的底层细节,在大流量高并发的压力下,任何一个小的错误都可能被无限放大,最终导致程序宕机。而对于基于HTTP协议的实现来说,很多成熟的开源web容易已经帮其处理好了这些事情,如Apache,Tomcat,Jboss等,开发人员可将更多的精力集中在业务实现上,而非处理底层细节。