Apache Avro RPC简单示例

一.简介
Apache Avro 是一个序列化系统，提供如下功能：
  1.丰富的数据结构
  2.压缩的、快速的、二进制数据格式
  3.存储持久化数据的容器文件
  4.RPC功能
  5.可以简单实现与动态语言的集成
特点：
Avro在数据读写过程中都利用基于JSON格式的Schemas，因而不用在序列化时对每个数据都要加一个类型等头部信息，从而使序列化保持快而小。并且因为数据加上对应的schema是自描述的，这样也可以促进在动态、脚本语言中的使用。
二、示例
示例参考【https://github.com/phunt/avro-rpc-quickstart】
本示例利用Mail协议简单模拟一个远程服务，Avro将利用这个服务来发送消息
1.编辑avpr协议文件

{"namespace": "example.proto",   （1）
 "protocol": "Mail",  （2）

 "types": [  （3）
     {"name": "Message", "type": "record",
      "fields": [
          {"name": "to",   "type": "string"},
          {"name": "from", "type": "string"},
          {"name": "body", "type": "string"}
      ]
     }
 ],

 "messages": { （4）
     "send": {
         "request": [{"name": "message", "type": "Message"}],
         "response": "string"
     }
 }
}

（1）定义命名空间
（2）定义协议名称
（3）定义类型信息
（4）定义rpc Messages

2.编写pom文件

<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/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>example.qs</groupId>
    <artifactId>avro-rpc-quickstart</artifactId>
    <packaging>jar</packaging>
    <version>1.8.1-SNAPSHOT</version>
    <name>avro-rpc-quickstart</name>
    <url>http://maven.apache.org</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <compiler-plugin.version>3.5</compiler-plugin.version>
        <avro.version>1.8.1</avro.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.10</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.6.4</version>
            <scope>compile</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>${avro.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-ipc</artifactId>
            <version>${avro.version}</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>${compiler-plugin.version}</version>
            </plugin>
            <plugin>
                <groupId>org.apache.avro</groupId>
                <artifactId>avro-maven-plugin</artifactId>
                <version>${avro.version}</version>
                <executions>
                    <execution>
                        <id>schemas</id>
                        <phase>generate-sources</phase>
                        <goals>
                            <goal>schema</goal>
                            <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>
        </plugins>
    </build>

</project>

3.在项目目录下执行 mvn compile ，执行结束后生成 Mail.java 和Message.java文件
4.编写测试程序

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package example;

import java.io.IOException;
import java.net.InetSocketAddress;

import org.apache.avro.ipc.NettyServer;
import org.apache.avro.ipc.NettyTransceiver;
import org.apache.avro.ipc.Server;
import org.apache.avro.ipc.specific.SpecificRequestor;
import org.apache.avro.ipc.specific.SpecificResponder;
import org.apache.avro.util.Utf8;

import example.proto.Mail;
import example.proto.Message;

/**
 * Start a server, attach a client, and send a message.
 */
public class Main {
    public static class MailImpl implements Mail {
        // in this simple example just return details of the message
        public Utf8 send(Message message) {
            System.out.println("Sending message");
            return new Utf8("Sending message to " + message.getTo().toString()
                    + " from " + message.getFrom().toString()
                    + " with body " + message.getBody().toString());
        }
    }

    private static Server server;

    private static void startServer() throws IOException {
        server = new NettyServer(new SpecificResponder(Mail.class, new MailImpl()), new InetSocketAddress(65111));
        // the server implements the Mail protocol (MailImpl)
    }

    public static void main(String[] args) throws IOException {
        if (args.length != 3) {
            System.out.println("Usage: <to> <from> <body>");
            System.exit(1);
        }

        System.out.println("Starting server");
        // usually this would be another app, but for simplicity
        startServer();
        System.out.println("Server started");

        NettyTransceiver client = new NettyTransceiver(new InetSocketAddress(65111));
        // client code - attach to the server and send a message
        Mail proxy = (Mail) SpecificRequestor.getClient(Mail.class, client);
        System.out.println("Client built, got proxy");

        // fill in the Message record and send it
        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("Calling proxy.send with message:  " + message.toString());
        System.out.println("Result: " + proxy.send(message));

        // cleanup
        client.close();
        server.close();
    }
}

5.执行测试，出入参数 tianjin beijing hello
  执行结果

Client built, got proxy
Calling proxy.send with message:  {"to": "tianjin", "from": "beijing", "body": "hello"}
Sending message
Result: Sending message to tianjin from beijing with body hello

6.工程代码参考附件


*当数据类型是 Map<String,String>
默认情况下，avpr文件中数据类型是string时，利用avro-maven-plugin生成的数据类型是CharSequence，比如：

{"namespace": "example.proto",
 "protocol": "Mail",

 "messages": {
     "send": {
         "request": [{"name": "jobId", "type": "string"},{"name": "message", "type": {"type":"map","values":"string"}}],
         "response": "string"
     }
 }
}

默认生成的类文件：

/**
 * Autogenerated by Avro
 *
 * DO NOT EDIT DIRECTLY
 */
package example.proto;

@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\":[],\"messages\":{\"send\":{\"request\":[{\"name\":\"jobId\",\"type\":\"string\"},{\"name\":\"message\",\"type\":{\"type\":\"map\",\"values\":\"string\"}}],\"response\":\"string\"}}}");
  /**
   */
  java.lang.CharSequence send(java.lang.CharSequence jobId, java.util.Map<java.lang.CharSequence,java.lang.CharSequence> message) throws org.apache.avro.AvroRemoteException;

  @SuppressWarnings("all")
  public interface Callback extends Mail {
    public static final org.apache.avro.Protocol PROTOCOL = example.proto.Mail.PROTOCOL;
    /**
     * @throws java.io.IOException The async call could not be completed.
     */
    void send(java.lang.CharSequence jobId, java.util.Map<java.lang.CharSequence,java.lang.CharSequence> message, org.apache.avro.ipc.Callback<java.lang.CharSequence> callback) throws java.io.IOException;
  }
}

如果我们想在程序中使用String类型时，需要在avro-maven-plugin 中加入<stringType>String</stringType>

<groupId>org.apache.avro</groupId>
<artifactId>avro-maven-plugin</artifactId>
<version>${avro.version}</version>
<configuration>
	<stringType>String</stringType>
</configuration>

此时生成的Java文件

/**
 * Autogenerated by Avro
 *
 * DO NOT EDIT DIRECTLY
 */
package example.proto;

@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\":[],\"messages\":{\"send\":{\"request\":[{\"name\":\"jobId\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"message\",\"type\":{\"type\":\"map\",\"values\":{\"type\":\"string\",\"avro.java.string\":\"String\"},\"avro.java.string\":\"String\"}}],\"response\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}}}");
  /**
   */
  java.lang.String send(java.lang.String jobId, java.util.Map<java.lang.String,java.lang.String> message) throws org.apache.avro.AvroRemoteException;

  @SuppressWarnings("all")
  public interface Callback extends Mail {
    public static final org.apache.avro.Protocol PROTOCOL = example.proto.Mail.PROTOCOL;
    /**
     * @throws java.io.IOException The async call could not be completed.
     */
    void send(java.lang.String jobId, java.util.Map<java.lang.String,java.lang.String> message, org.apache.avro.ipc.Callback<java.lang.String> callback) throws java.io.IOException;
  }
}