轻松使用Hadoop RPC

openkk 7年前
     <p>Hadoop RPC是Hadoop的一个重要部分,提供分布式环境下的对象调用功能,源码在org.apache.hadoop.ipc中。而HBase也几乎完全copy了这部分的源码,只是在配置项上面有所改动。</p>    <p>关于Hadoop RPC的机制分析和源码解读,网上已经有许多资料,一搜一大把,这里就不在描述了。本文通过一个小例子,介绍如何调用Hadoop RPC。</p>    <h2>1. 应用场景</h2>    <p>Hadoop RPC在整个Hadoop中应用非常广泛,Client、DataNode、NameNode之间的通讯全靠它了。</p>    <p>举个例子,我们平时操作HDFS的时候,使用的是FileSystem类,它的内部有个DFSClient对象,这个对象负责与NameNode打交道。在运行时,DFSClient在本地创建一个NameNode的代理,然后就操作这个代理,这个代理就会通过网络,远程调用到NameNode的方法,也能返回值。</p>    <p>在我的应用场景中,需要一个元数据服务器,各节点经常需要去查询元数据,可以使用这套RPC机制。</p>    <h2>2. Protocol</h2>    <p>被远程访问的类,也就是Server端,必须实现VersionedProtocol接口,这个接口只有一个方法getProtocolVersion,用来判断Server和Client端调用的是不是一个版本的,一般Server的代码修改一次,版本号就得改一次。</p>    <p>在例子中,我们定义一个接口MyProtocol,继承VersionedProtocol,里面定义Server端需要实现的方法。</p>    <p>这里MyProtocol接口只有一个方法println,输入一个Text,打印出来,并返回一个Text。</p>    <p>MyProtocol.java代码如下:</p>    <pre class="brush:java; toolbar: true; auto-links: false;">import org.apache.hadoop.io.Text;  import org.apache.hadoop.ipc.VersionedProtocol;    public interface MyProtocol extends VersionedProtocol {      public Text println(Text t);  }</pre>    <p></p>    <h2>3. Server</h2>    <p>Server端实现上述的Protocol接口,里面需要启动一个RPC.Server,它是一个Thread。</p>    <p>构造方法是RPC.getServer(Object instance, String bindAddress, int port, Configuration conf)</p>    <ul style="list-style-type:disc;">     <li>instance:表示提供远程访问的对象,一般Server都会传入this作为参数;</li>     <li>bindAddress:Server绑定的ip地址;</li>     <li>port:Server绑定的端口;</li>     <li>conf:Configuration对象,不用解释了吧。</li>    </ul>    <p>MyServer实现了MyProtocol接口中定义的println方法,将参数打印到控制台,并返回finish。</p>    <p>MyServer.java代码如下:</p>    <pre class="brush:java; toolbar: true; auto-links: false;">import java.io.IOException;  import java.net.UnknownHostException;    import org.apache.hadoop.conf.Configuration;  import org.apache.hadoop.io.Text;  import org.apache.hadoop.ipc.RPC;  import org.apache.hadoop.ipc.RPC.Server;    public class MyServer implements MyProtocol{      private Server server;            public MyServer(){          try {              server = RPC.getServer(this, "localhost", 8888, new Configuration());              server.start();              server.join();          } catch (UnknownHostException e) {              e.printStackTrace();          } catch (IOException e) {              e.printStackTrace();          } catch (InterruptedException e) {              e.printStackTrace();          }      }            @Override      public Text println(Text t){          System.out.println(t);          return new Text("finish");      }        @Override      public long getProtocolVersion(String protocol, long clientVersion) throws IOException {          return 1;      }            public static void main(String[] args) {          new MyServer();      }    }</pre>    <p></p>    <h2>4. Client</h2>    <p>Client端需要创建一个Server的远程代理,并可以通过操作这个代理,来调用到Server端的方法。</p>    <p>创建代理可以调用RPC.waitForProxy(Class protocol, long clientVersion, InetSocketAddress addr, Configuration conf)</p>    <ul style="list-style-type:disc;">     <li>protocol:一个Protocol的class,它必须是继承VersionedProtocol的接口;</li>     <li>clientVersion:客户端的版本号,如果与服务端不一致,则会抛错;</li>     <li>addr:一个InetSocketAddress对象,包含了ip和port;</li>     <li>conf:不解释。</li>    </ul>    <p>这个方法会返回一个VersionedProtocol类型的代理对象,将它强制转型成自己定义的Protocol,接下来就可以操作创建好的代理了。在例子中,我们通过代理来让Server端打印字符串到控制台,并接受返回的消息。</p>    <p>MyClient.java代码如下:</p>    <pre class="brush:java; toolbar: true; auto-links: false;">import java.io.IOException;  import java.net.InetSocketAddress;    import org.apache.hadoop.conf.Configuration;  import org.apache.hadoop.io.Text;  import org.apache.hadoop.ipc.RPC;    public class MyClient {            private MyProtocol proxy;        public MyClient(){          InetSocketAddress addr = new InetSocketAddress("localhost",8888);          try {              proxy = (MyProtocol) RPC.waitForProxy(MyProtocol.class, 1, addr , new Configuration());          } catch (IOException e) {              e.printStackTrace();          }      }            public void println(String s){          System.out.println(proxy.println(new Text(s)));      }        public void close(){          RPC.stopProxy(proxy);      }            public static void main(String[] args) {          MyClient c = new MyClient();          c.println("123");          c.close();      }  }</pre>    <p></p>    <h2>5. 运行</h2>    <p>运行MyServer,控制台显示:</p>    <blockquote>     <p style="margin-left:30px;">2011-12-30 18:49:56 -[INFO] Initializing RPC Metrics with hostName=MyServer, port=8888<br /> 2011-12-30 18:49:56 -[INFO] IPC Server listener on 8888: starting<br /> 2011-12-30 18:49:56 -[INFO] IPC Server Responder: starting<br /> 2011-12-30 18:49:56 -[INFO] IPC Server handler 0 on 8888: starting</p>    </blockquote>    <p>运行MyClient,控制台显示:</p>    <blockquote>     <p>finish</p>    </blockquote>    <p>MyServer端会追加显示:</p>    <blockquote>     <p>123 </p>    </blockquote>    <p></p>    <p></p>    <p></p> 转自:    <a href="/misc/goto?guid=4959500139677589909" target="_blank">http://www.cnblogs.com/hiddenfox/archive/2011/12/30/2305786.html</a>