1. Avro RPC简介

1.1. RPC

• RPC逻辑上分为二层，一是传输层，负责网络通信；二是协议层，将数据按照一定协议格式打包和解包
• 从序列化方式来看，Apache Thrift 和Google的Protocol Buffers和Avro应该是属于同一个级别的框架，都能跨语言，性能优秀，数据精简，但是Avro的动态模式（不用生成代码，而且性能很好）这个特点让人非常喜欢，比较适合RPC的数据交换。

1.2. Avro RPC的主要特点

Avro RPC 是一个支持跨语言实现的RPC服务框架。非常轻量级，实现简洁，使用方便，同时支持使用者进行二次开发，逻辑上该框架分为两层：

• 网络传输层使用Netty的Nio实现。
• 协议层可扩展，目前支持的数据序列化方式有Avro，Protocol Buffers ，Json, Hessian,Java序列化。 使用者可以注册自己的协议格式及序列化方式。

Avro RPC主要特点：

• 客户端传输层与应用层逻辑分离，传输层主要职责包括创建连接，连接查找与复用，传输数据，接收服务端回复后回调应用层；
• 客户端支持同步调用和异步调用。服务异步化能很好的提高系统吞吐量，建议使用异步调用。为防止异步发送请求过快，客户端增加了“请求流量限制”功能；
• 服务端有一个协议注册工厂和序列化注册工厂。这样方便针对不同的应用场景来定制服务方式。RPC应该只是服务方式的一种。在分布式的系统架构中，分布式节点之间的通信会存在多种方式，比如MQ的TOP消息，一个消息可以有多个订阅者。因此avro-rpc不仅仅是一个RPC服务框架，还是一个分布式通信的一个基础骨架，提供了很好的扩展性；
• Avro序列化框架是Hadoop下的一个子项目，其特点是数据序列化不带标签，因此序列化后的数据非常小。支持动态解析, 不像Thrift 与 Protocol Buffers必须根据IDL来生成代码，这样侵入性有点强。性能很好，基本上和 Protocol Buffers差不多；

2. Avro RPC开发

2.1 Maven依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>learn</groupId><artifactId>learn.avro</artifactId><version>1.0-SNAPSHOT</version><dependencies><!--avro core--><dependency><groupId>org.apache.avro</groupId><artifactId>avro</artifactId><version>1.7.7</version></dependency><!--avro rpc support--><dependency><groupId>org.apache.avro</groupId><artifactId>avro-ipc</artifactId><version>1.7.7</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.avro</groupId><artifactId>avro-maven-plugin</artifactId><version>1.7.7</version><executions><execution><phase>generate-sources</phase><goals><!--Maven goal that helps for code generation--><goal>schema</goal><!--For RPC used--><goal>protocol</goal><goal>idl-protocol</goal></goals><configuration><sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory><outputDirectory>${project.basedir}/src/main/java/</outputDirectory></configuration></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>1.7</source><target>1.7</target></configuration></plugin></plugins></build>
</project>

2.2 定义协议schema文件(在src/main/avro/mail.avpr)

{"namespace": "examples.avro.rpc","protocol": "Mail","types": [{"name": "Message", "type": "record","fields": [{"name": "to",   "type": "string"},{"name": "from", "type": "string"},{"name": "body", "type": "string"}]}],"messages": {"send": {"request": [{"name": "message", "type": "Message"}],"response": "string"}}
}

2.3 生成代码：

在Intellij Idea的Maven视图中,learn avro->Plugins->avro->avro:protocol,右击avro:protocol，执行Run Maven Build，生成protocol schema对应的Java实体类

2.3.1 Mail接口

/*** Autogenerated by Avro** DO NOT EDIT DIRECTLY*/
package examples.avro.rpc;@SuppressWarnings("all")
@org.apache.avro.specific.AvroGenerated
public interface Mail {public static final org.apache.avro.Protocol PROTOCOL = org.apache.avro.Protocol.parse("{\"protocol\":\"Mail\",\"namespace\":\"example.proto\",\"types\":[{\"type\":\"record\",\"name\":\"Message\",\"fields\":[{\"name\":\"to\",\"type\":\"string\"},{\"name\":\"from\",\"type\":\"string\"},{\"name\":\"body\",\"type\":\"string\"}]}],\"messages\":{\"send\":{\"request\":[{\"name\":\"message\",\"type\":\"Message\"}],\"response\":\"string\"}}}");///Mail接口有1个方法send，参数是Message，Message是一个Avro类，可以序列化和反序列化java.lang.CharSequence send(Message message) throws org.apache.avro.AvroRemoteException;@SuppressWarnings("all")public interface Callback extends Mail {public static final org.apache.avro.Protocol PROTOCOL = Mail.PROTOCOL;void send(Message message, org.apache.avro.ipc.Callback<CharSequence> callback) throws java.io.IOException;}
}

2.3.2 Message类(根据schema文件生成)

/*** Autogenerated by Avro* * DO NOT EDIT DIRECTLY*/
package examples.avro.rpc;
@SuppressWarnings("all")
@org.apache.avro.specific.AvroGenerated
public class Message extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Message\",\"namespace\":\"example.proto\",\"fields\":[{\"name\":\"to\",\"type\":\"string\"},{\"name\":\"from\",\"type\":\"string\"},{\"name\":\"body\",\"type\":\"string\"}]}");public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }@Deprecated public java.lang.CharSequence to;@Deprecated public java.lang.CharSequence from;@Deprecated public java.lang.CharSequence body;/*** Default constructor.  Note that this does not initialize fields* to their default values from the schema.  If that is desired then* one should use <code>newBuilder()</code>. */public Message() {}/*** All-args constructor.*/public Message(java.lang.CharSequence to, java.lang.CharSequence from, java.lang.CharSequence body) {this.to = to;this.from = from;this.body = body;}public org.apache.avro.Schema getSchema() { return SCHEMA$; }// Used by DatumWriter.  Applications should not call. public java.lang.Object get(int field$) {switch (field$) {case 0: return to;case 1: return from;case 2: return body;default: throw new org.apache.avro.AvroRuntimeException("Bad index");}}// Used by DatumReader.  Applications should not call. @SuppressWarnings(value="unchecked")public void put(int field$, java.lang.Object value$) {switch (field$) {case 0: to = (java.lang.CharSequence)value$; break;case 1: from = (java.lang.CharSequence)value$; break;case 2: body = (java.lang.CharSequence)value$; break;default: throw new org.apache.avro.AvroRuntimeException("Bad index");}}/*** Gets the value of the 'to' field.*/public java.lang.CharSequence getTo() {return to;}/*** Sets the value of the 'to' field.* @param value the value to set.*/public void setTo(java.lang.CharSequence value) {this.to = value;}/*** Gets the value of the 'from' field.*/public java.lang.CharSequence getFrom() {return from;}/*** Sets the value of the 'from' field.* @param value the value to set.*/public void setFrom(java.lang.CharSequence value) {this.from = value;}/*** Gets the value of the 'body' field.*/public java.lang.CharSequence getBody() {return body;}/*** Sets the value of the 'body' field.* @param value the value to set.*/public void setBody(java.lang.CharSequence value) {this.body = value;}/** Creates a new Message RecordBuilder */public static Message.Builder newBuilder() {return new Message.Builder();}/** Creates a new Message RecordBuilder by copying an existing Builder */public static Message.Builder newBuilder(Message.Builder other) {return new Message.Builder(other);}/** Creates a new Message RecordBuilder by copying an existing Message instance */public static Message.Builder newBuilder(Message other) {return new Message.Builder(other);}/*** RecordBuilder for Message instances.*/public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Message>implements org.apache.avro.data.RecordBuilder<Message> {private java.lang.CharSequence to;private java.lang.CharSequence from;private java.lang.CharSequence body;/** Creates a new Builder */private Builder() {super(Message.SCHEMA$);}/** Creates a Builder by copying an existing Builder */private Builder(Message.Builder other) {super(other);if (isValidValue(fields()[0], other.to)) {this.to = data().deepCopy(fields()[0].schema(), other.to);fieldSetFlags()[0] = true;}if (isValidValue(fields()[1], other.from)) {this.from = data().deepCopy(fields()[1].schema(), other.from);fieldSetFlags()[1] = true;}if (isValidValue(fields()[2], other.body)) {this.body = data().deepCopy(fields()[2].schema(), other.body);fieldSetFlags()[2] = true;}}/** Creates a Builder by copying an existing Message instance */private Builder(Message other) {super(Message.SCHEMA$);if (isValidValue(fields()[0], other.to)) {this.to = data().deepCopy(fields()[0].schema(), other.to);fieldSetFlags()[0] = true;}if (isValidValue(fields()[1], other.from)) {this.from = data().deepCopy(fields()[1].schema(), other.from);fieldSetFlags()[1] = true;}if (isValidValue(fields()[2], other.body)) {this.body = data().deepCopy(fields()[2].schema(), other.body);fieldSetFlags()[2] = true;}}/** Gets the value of the 'to' field */public java.lang.CharSequence getTo() {return to;}/** Sets the value of the 'to' field */public Message.Builder setTo(java.lang.CharSequence value) {validate(fields()[0], value);this.to = value;fieldSetFlags()[0] = true;return this; }/** Checks whether the 'to' field has been set */public boolean hasTo() {return fieldSetFlags()[0];}/** Clears the value of the 'to' field */public Message.Builder clearTo() {to = null;fieldSetFlags()[0] = false;return this;}/** Gets the value of the 'from' field */public java.lang.CharSequence getFrom() {return from;}/** Sets the value of the 'from' field */public Message.Builder setFrom(java.lang.CharSequence value) {validate(fields()[1], value);this.from = value;fieldSetFlags()[1] = true;return this; }/** Checks whether the 'from' field has been set */public boolean hasFrom() {return fieldSetFlags()[1];}/** Clears the value of the 'from' field */public Message.Builder clearFrom() {from = null;fieldSetFlags()[1] = false;return this;}/** Gets the value of the 'body' field */public java.lang.CharSequence getBody() {return body;}/** Sets the value of the 'body' field */public Message.Builder setBody(java.lang.CharSequence value) {validate(fields()[2], value);this.body = value;fieldSetFlags()[2] = true;return this; }/** Checks whether the 'body' field has been set */public boolean hasBody() {return fieldSetFlags()[2];}/** Clears the value of the 'body' field */public Message.Builder clearBody() {body = null;fieldSetFlags()[2] = false;return this;}@Overridepublic Message build() {try {Message record = new Message();record.to = fieldSetFlags()[0] ? this.to : (java.lang.CharSequence) defaultValue(fields()[0]);record.from = fieldSetFlags()[1] ? this.from : (java.lang.CharSequence) defaultValue(fields()[1]);record.body = fieldSetFlags()[2] ? this.body : (java.lang.CharSequence) defaultValue(fields()[2]);return record;} catch (Exception e) {throw new org.apache.avro.AvroRuntimeException(e);}}}
}

2.3.3 AvroServer类

package examples.avro.rpc;import org.apache.avro.ipc.NettyServer;
import org.apache.avro.ipc.Server;
import org.apache.avro.ipc.specific.SpecificResponder;
import org.apache.avro.util.Utf8;import java.io.IOException;
import java.net.InetSocketAddress;//Server端的实现Mai服务
class MailImpl implements Mail {public Utf8 send(Message message) {System.out.println("Message Received：" + message);return new Utf8("Received your message： " + message.getFrom().toString()+ " with body " + message.getBody().toString());}
}public class AvroServer {private static Server server;public static void main(String[] args) throws Exception {System.out.println("Starting server");startServer();Thread.sleep(1000);System.out.println("Server started");Thread.sleep(60 * 1000);server.close();}private static void startServer() throws IOException {server = new NettyServer(new SpecificResponder(Mail.class, new MailImpl()), new InetSocketAddress(65111));}
}

2.3.3 AvroClient类

package examples.avro.rpc;import org.apache.avro.ipc.NettyTransceiver;
import org.apache.avro.ipc.specific.SpecificRequestor;
import org.apache.avro.util.Utf8;import java.net.InetSocketAddress;public class AvroClient {public static void main(String[] args) throws Exception {NettyTransceiver client = new NettyTransceiver(new InetSocketAddress(65111));///获取Mail接口的proxy实现Mail proxy = SpecificRequestor.getClient(Mail.class, client);System.out.println("Client of Mail Proxy is built");// fill in the Message record and send itargs = new String[]{"to:Tom", "from:Jack", "body:How are you"};Message message = new Message();message.setTo(new Utf8(args[0]));message.setFrom(new Utf8(args[1]));message.setBody(new Utf8(args[2]));System.out.println("RPC call with message:  " + message.toString());///底层给服务器发送send方法调用System.out.println("Result: " + proxy.send(message));// cleanupclient.close();}
}

本文支持对Avro RPC的粗浅尝试，Avro Client端用的同步通信方式