RPC_AVRO范例

1.下载与安装

　　官方网站：http://avro.apache.org/

　　下载地址：http://labs.renren.com/apache-mirror//avro/avro-1.5.1/avro-src-1.5.1.tar.gz

　　安装之前确保已经装了maven

1. cd /usr/local/src
2. wget http://labs.renren.com/apache-mirror//avro/avro-1.5.1/avro-src-1.5.1.tar.gz
3. tar zxvf avro-src-1.5.1.tar.gz
4. cd avro-src-1.5.1/lang/java
5. mvn clean install -DskipTests

　　安装后，avro-1.5.1.jar位于avro-src-1.5.1/lang/java/avro/target

 

2.消息结构与服务接口

　　Avro的模式主要由JSON对象来表示，Avro支持8种基本类型(Primitive Type)和6种复杂类型(Complex Type：records、enums、arrays、maps、unions 和fixed)，基本类型可以由JSON字符串来表示。

　　Avro支持两种序列化编码方式：二进制编码和JSON编码，使用二进制编码会高效序列化，并且序列化后得到的结果会比较小。

　　基本类型：

1. null: no value
2. boolean: a binary value
3. int: 32-bit signed integer
4. long: 64-bit signed integer
5. float: single precision (32-bit) IEEE 754 floating-point number
6. double: double precision (64-bit) IEEE 754 floating-point number
7. bytes: sequence of 8-bit unsigned bytes
8. string: unicode character sequence

　　首先编写一个message.avpr文件，定义一个消息结构。

1. {
2.     "namespace": "avro",
3.     "protocol": "messageProtocol",
4.     "doc": "This is a message.",
5.     "name": "Message",
6.  
7.     "types": [
8.         {"name":"message", "type":"record",
9.             "fields":[
10.                 {"name":"name", "type":"string"},
11.                 {"name":"type", "type":"int"},
12.                 {"name":"price", "type":"double"},
13.                 {"name":"valid", "type":"boolean"},
14.                 {"name":"content", "type":"bytes"}
15.         ]}
16.     ],
17.  
18.     "messages":    {
19.         "sendMessage":{
20.             "doc" : "test",
21.             "request" :[{"name":"message","type":"message" }],
22.             "response" :"message"
23.         }         
24.     }   
25. }

　　其中定义了1种类型叫做message，有5个成员name、type、price、valid、content。还定义了1个消息服务叫做sendMessage，输入有一个参数，类型是message，返回message。

3.序列化

　　Avro有两种序列化编码：binary和JSON。

3.1.Binary Encoding

　　基本类型：

　　　　null：0字节

　　　　boolean：1个字节——0（false）或1（true）

　　　　int和long使用变长的zig-zag编码

　　　　float：4个字节

　　　　double：8个字节

　　　　bytes：1个long，后边跟着字节序列

　　　　string：1个long，后边跟着UTF－8编码的字符

3.2.records

　　按字段声明的顺序编码值，如下面一个record schema：

1. {
2.     "type": "record",
3.     "name": "test",
4.     "fields" : [
5.         {"name": "a", "type": "long"},
6.         {"name": "b", "type": "string"}]
7. }

　　实例化这个record，a字段的值是27（编码为0×36），b字段的值是“foo”（编码为06 66 6f 6f）,那么这个record编码结果是：

1. 36 06 66 6f 6f

3.3.enums

　　一个enum被编码为一个int，比如，考虑这个enum。

1. {"type": "enum", "name": "Foo", "symbols": ["A", "B", "C", "D"] }

　　这将被编码为一个取值范围为[0,3]的int，0表示“A”，3表示“D”。

3.4.arrays

　　arrays编码为block序列，每个block包含一个long的count值，紧跟着的是array items，一个block的count为0表示该block是array的结尾。

3.5.maps

　　mapss编码为block序列，每个block包含一个long的count值，紧跟着的是key/value对，一个block的count为0表示该block是map的结尾。

3.6.union

　　union编码以一个long值开始，表示后边的数据是union中的哪种数据类型。

3.7.fixed

　　编码为指定数目的字节。

4.rpc通信实现

　　Avro的RPC实现不需要定义服务接口，但需要从.avpr文件中解析协议，协议中定义了消息结构和消息服务。message.avpr中定义了一个类型叫message，定义了一个服务叫sendMessage。

　　工具类Utils.java：

1. package avro;
2.  
3. import java.io.File;
4. import java.io.IOException;
5. import java.net.URL;
6.  
7. import org.apache.avro.Protocol;
8.  
9. public class Utils {
10.     public static Protocol getProtocol() {
11.         Protocol protocol = null;
12.         try {
13.             URL url = Utils.class.getClassLoader().getResource("message.avpr");
14.             protocol = Protocol.parse(new File(url.getPath()));
15.         } catch (IOException e) {
16.             e.printStackTrace();
17.         }
18.         return protocol;
19.     }
20. }

　　服务端实现Server.java：

1. package avro;
2.  
3. import org.apache.avro.Protocol;
4. import org.apache.avro.Protocol.Message;
5. import org.apache.avro.generic.GenericRecord;
6. import org.apache.avro.ipc.HttpServer;
7. import org.apache.avro.ipc.generic.GenericResponder;
8.  
9. public class Server extends GenericResponder {
10.     private Protocol protocol = null;
11.     private int port;
12.  
13.     public Server(Protocol protocol, int port) {
14.         super(protocol);
15.         this.protocol = protocol;
16.         this.port = port;
17.     }
18.  
19.     public Object respond(Message message, Object request) throws Exception {
20.         GenericRecord req = (GenericRecord) request;
21.         GenericRecord msg = (GenericRecord)(req.get("message"));
22.         // process the request
23.         …
24.         return msg;
25.     }
26.  
27.     public void run() {
28.         try {
29.             HttpServer server = new HttpServer(this, port);
30.  
31.             server.start();
32.         } catch (Exception e) {
33.             e.printStackTrace();
34.         }
35.     }
36.  
37.     public static void main(String[] args) {
38.         if (args.length != 1) {
39.             System.out.println("Usage: Server port");
40.             System.exit(0);
41.         }
42.         int port = Integer.parseInt(args[0]);
43.         new Server(Utils.getProtocol(), port).run();
44.     }
45. }

　　客户端实现Client.java：

1. package avro;
2.  
3. import java.net.URL;
4. import java.nio.ByteBuffer;
5. import java.util.Arrays;
6.  
7. import org.apache.avro.util.Utf8;
8. import org.apache.avro.Protocol;
9. import org.apache.avro.generic.GenericData;
10. import org.apache.avro.generic.GenericRecord;
11. import org.apache.avro.ipc.HttpTransceiver;
12. import org.apache.avro.ipc.Transceiver;
13. import org.apache.avro.ipc.generic.GenericRequestor;
14.  
15. public class Client {
16.     private Protocol protocol = null;
17.     private String host = null;
18.     private int port = 0;
19.     private int size = 0;
20.     private int count = 0;
21.  
22.     public Client(Protocol protocol, String host, int port, int size, int count) {
23.         this.protocol = protocol;
24.         this.host = host;
25.         this.port = port;
26.         this.size = size;
27.         this.count = count;
28.     }
29.  
30.     public long sendMessage() throws Exception {
31.         GenericRecord requestData = new GenericData.Record(
32.                 protocol.getType("message"));
33.         // initiate the request data
34.         …
35.  
36.         GenericRecord request = new GenericData.Record(protocol.getMessages()
37.                 .get("sendMessage").getRequest());
38.         request.put("message", requestData);
39.  
40.         Transceiver t = new HttpTransceiver(new URL("http://" + host + ":"
41.                 + port));
42.         GenericRequestor requestor = new GenericRequestor(protocol, t);
43.  
44.         long start = System.currentTimeMillis();
45.         for (int i = 0; i < count; i++) {
46.             requestor.request("sendMessage", request);
47.         }
48.         long end = System.currentTimeMillis();
49.         System.out.println(end - start);
50.         return end - start;
51.     }
52.  
53.     public long run() {
54.         long res = 0;
55.         try {
56.             res = sendMessage();
57.         } catch (Exception e) {
58.             e.printStackTrace();
59.         }
60.         return res;
61.     }
62.  
63.     public static void main(String[] args) throws Exception {
64.         if (args.length != 4) {
65.             System.out.println("Usage: Client host port dataSize count");
66.             System.exit(0);
67.         }
68.  
69.         String host = args[0];
70.         int port = Integer.parseInt(args[1]);
71.         int size = Integer.parseInt(args[2]);
72.         int count = Integer.parseInt(args[3]);
73.         new Client(Utils.getProtocol(), host, port, size, count).run();
74.     }
75. }

5.参考资料

　　(1) Avro Documentation: http://avro.apache.org/docs/current/index.html