Apache Avro RPC简单示例 -

fengyilin

浏览: 194814 次
性别:
来自: 南京

最近访客更多访客>>

hanfu_wei

jafeyyu

haydenzhu

map60148

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

Apache Avro RPC简单示例

博客分类：

apache avro

mvn 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;
  }
}