Avro RPC实例

Posted 2021-03-11 hzhuxin

RPC(Remote Procedure Call)：远程过程调用，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的思想。

一个 RPC 的核心功能主要有 5 个部分组成，分别是：客户端、客户端 Stub、网络传输模块、服务端 Stub、服务端等。

Avro除了数据序列化功能外也提供了RPC功能。具体使用如下：

1. 在pom文件中添加相应依赖

        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-ipc</artifactId>
            <version>1.9.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-ipc-netty</artifactId>
            <version>1.9.1</version>
        </dependency>

 修改插件配置

<plugin>
                <groupId>org.apache.avro</groupId>
                <artifactId>avro-maven-plugin</artifactId>
                <version>1.9.1</version>
                <executions>
                    <execution>
                        <phase>generate-sources</phase>
                        <goals>
                            <goal>schema</goal>
                            <goal>idl-protocol</goal>  //添加这一行
                        </goals>
                        <configuration>
                            <sourceDirectory>${project.basedir}/src/main/resources/</sourceDirectory>
                            <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
                        </configuration>
                    </execution>
                </executions>
            </plugin>

 

2. 编写服务API契约（avdl文件）

一个服务API契约在Avro中称之为 Avro Protocol ，Avro 提供了一种IDL语言来简化这个Avro Protocol的编写，如下：

@namespace("com.zpark.demo.avro.rpc")
protocol PersonService{
import schema "person.avsc";  //此处导入schema
  boolean addPerson(com.zpark.Person person); //定义一个远程服务
}

在实际运行时，上述的avdl文件会转成avpr的文件形式，我们可以通过 avroj-tools.jar来进行查看并输出avpr文件，命令如下：

java -jar avro-tools-1.9.1.jar idl person.avdl person.avpr

生成的Avro protocol模式文件内容如下：

{
  "protocol" : "PersonService",
  "namespace" : "com.zpark.demo.avro.rpc",
  "types" : [ {
    "type" : "record",
    "name" : "Person",
    "namespace" : "com.zpark",
    "fields" : [ {
      "name" : "id",
      "type" : "string"
    }, {
      "name" : "name",
      "type" : "string"
    }, {
      "name" : "age",
      "type" : [ "int", "null" ]
    } ]
  } ],
  "messages" : {
    "addPerson" : {
      "request" : [ {
        "name" : "person",
        "type" : "com.zpark.Person"
      } ],
      "response" : "boolean"
    }
  }
}

3.生成服务接口类

  使用 mvn clean install -DskipTests=true 对项目进行打包，就会触发maven插件根据编写的avdl文件生成服务契约接口。

package com.zpark.demo.avro.rpc;

@org.apache.avro.specific.AvroGenerated
public interface PersonService {
  public static final org.apache.avro.Protocol PROTOCOL = org.apache.avro.Protocol.parse("{"protocol":"PersonService","namespace":"com.zpark.demo.avro.rpc","types":[{"type":"record","name":"Person","namespace":"com.zpark","fields":[{"name":"id","type":"string"},{"name":"name","type":"string"},{"name":"age","type":["int","null"]}]}],"messages":{"addPerson":{"request":[{"name":"person","type":"com.zpark.Person"}],"response":"boolean"}}}");
  /**
   */
  boolean addPerson(com.zpark.Person person);

  @SuppressWarnings("all")
  public interface Callback extends PersonService {
    public static final org.apache.avro.Protocol PROTOCOL = com.zpark.demo.avro.rpc.PersonService.PROTOCOL;
    /**
     * @throws java.io.IOException The async call could not be completed.
     */
    void addPerson(com.zpark.Person person, org.apache.avro.ipc.Callback<java.lang.Boolean> callback) throws java.io.IOException;
  }
}

 

4.实现服务接口

public class PersonServiceImpl implements PersonService {
    @Override
    public boolean addPerson(Person person) {
        System.out.println(person);
        return true ;
    }
}

 

5.编写Avro RPC Server端相关代码

public class AvroRpcServer {
    public static void main(String[] args) {
        NettyServer ns = new NettyServer(
                new SpecificResponder(PersonService.class,new PersonServiceImpl()),
                new InetSocketAddress(9998)
        );
        ns.start();
        System.out.println("服务端已启动");
    }
}

 

6.编写Avro RPC Client端相关代码

public class AvroClient {

    public static void main(String[] args) throws Exception{
        NettyTransceiver client = new NettyTransceiver(new InetSocketAddress("127.0.0.1",9998));
        PersonService proxy = SpecificRequestor.getClient(PersonService.class, client);
        Person person = new Person("001","zhangsan",23);
        boolean result = proxy.addPerson(person);
        if(result){
            System.out.println("服务调用成功");
        }else {
            System.out.println("服务调用失败");
        }

    }


}

 

7.测试