分布式架构基础:Java RMI详解

GitHub: https://github.com/jayknoxqu/rmi-example

RMI简介

​ Java RMI，即 远程方法调用(Remote Method Invocation)，一种用于实现远程过程调用(RPC)(Remote procedure call)的Java API， 能直接传输序列化后的Java对象和分布式垃圾收集。它的实现依赖于Java虚拟机(JVM)，因此它仅支持从一个JVM到另一个JVM的调用。

rmi架构图

rmi的实现

(1) 直接使用Registry实现rmi

服务端:
public class RegistryService {
    public static void main(String[] args) {
        try {
            // 本地主机上的远程对象注册表Registry的实例,默认端口1099
            Registry registry = LocateRegistry.createRegistry(1099);
            // 创建一个远程对象
            HelloRegistryFacade hello = new HelloRegistryFacadeImpl();
            // 把远程对象注册到RMI注册服务器上，并命名为HelloRegistry
            registry.rebind("HelloRegistry", hello);
            System.out.println("======= 启动RMI服务成功! =======");
        } catch (RemoteException e) {
            e.printStackTrace();
        }
    }
}
接口:

继承Remote接口

public interface HelloRegistryFacade extends Remote {

    String helloWorld(String name) throws RemoteException;

}
接口实现:

继承UnicastRemoteObject

public class HelloRegistryFacadeImpl extends UnicastRemoteObject implements HelloRegistryFacade{

    public HelloRegistryFacadeImpl() throws RemoteException {
        super();
    }

    @Override
    public String helloWorld(String name) {
        return "[Registry] 你好! " + name;
    }

}
客户端:
public class RegistryClient {
    public static void main(String[] args) {
        try {
            Registry registry = LocateRegistry.getRegistry(1099);
            HelloRegistryFacade hello = (HelloRegistryFacade) registry.lookup("HelloRegistry");
            String response = hello.helloWorld("ZhenJin");
            System.out.println("=======> " + response + " <=======");
        } catch (NotBoundException | RemoteException e) {
            e.printStackTrace();
        }
    }
}
图解:

出处:https://www.tutorialspoint.com/java_rmi/java_rmi_introduction.htm

rmi调用过程
Registry(注册表)是放置所有服务器对象的命名空间。
每次服务端创建一个对象时，它都会使用bind()或rebind()方法注册该对象。
这些是使用称为绑定名称的唯一名称注册的。

要调用远程对象，客户端需要该对象的引用,如(HelloRegistryFacade)。
即通过服务端绑定的名称(HelloRegistry)从注册表中获取对象(lookup()方法)。

(2) 使用Naming方法实现rmi

服务端:
public class NamingService {
    public static void main(String[] args) {
        try {
            // 本地主机上的远程对象注册表Registry的实例
            LocateRegistry.createRegistry(1100);
            // 创建一个远程对象
            HelloNamingFacade hello = new HelloNamingFacadeImpl();
            // 把远程对象注册到RMI注册服务器上，并命名为Hello 
            //绑定的URL标准格式为：rmi://host:port/name
            Naming.bind("rmi://localhost:1100/HelloNaming", hello);
            System.out.println("======= 启动RMI服务成功! =======");
        } catch (RemoteException | MalformedURLException | AlreadyBoundException e) {
            e.printStackTrace();
        }
    }
}

接口和接口实现和Registry的方式一样

客户端:
public class NamingClient {
    public static void main(String[] args) {
        try {
            String remoteAddr="rmi://localhost:1100/HelloNaming";
            HelloNamingFacade hello = (HelloNamingFacade) Naming.lookup(remoteAddr);
            String response = hello.helloWorld("ZhenJin");
            System.out.println("=======> " + response + " <=======");
        } catch (NotBoundException | RemoteException | MalformedURLException e) {
            e.printStackTrace();
        }
    }
}
Naming部分源码:
public static Remote lookup(String name)
    throws NotBoundException,java.net.MalformedURLException,RemoteException{
    ParsedNamingURL parsed = parseURL(name);
    Registry registry = getRegistry(parsed);

    if (parsed.name == null)
        return registry;
    return registry.lookup(parsed.name);
}

Naming其实是对Registry的一个封装

Scala实现rmi

上面说了rmi是通过JVM虚拟机进行一个远程调用的,我们通过Scala,kotlin等jvm语言印证下

服务端:
object ScalaRmiService extends App {
  try {
    val user:UserScalaFacade = new UserScalaFacadeImpl
    LocateRegistry.createRegistry(1103)
    Naming.rebind("rmi://localhost:1103/UserScala", user)
    println("======= 启动RMI服务成功! =======")
  } catch {
    case e: IOException => println(e)
  }
}
接口
trait UserScalaFacade extends Remote {

  /**
    * 通过用户名获取用户信息
    */
  @throws(classOf[RemoteException])
  def getByName(userName: String): User

  /**
    * 通过用户性别获取用户信息
    */
  @throws(classOf[RemoteException])
  def getBySex(userSex: String): List[User]

}
接口实现:
class UserScalaFacadeImpl extends UnicastRemoteObject with UserScalaFacade {

  /**
    * 模拟一个数据库表
    */
  private lazy val userList = List(
    new User("Jane", "女", 16),
    new User("jack", "男", 17),
    new User("ZhenJin", "男", 18)
  )

  override def getByName(userName: String): User = userList.filter(u => userName.equals(u.userName)).head

  override def getBySex(userSex: String): List[User] = userList.filter(u => userSex.equals(u.userSex))

}
实体类:

实体类必须实现序列化(Serializable)才能进行一个远程传输

class User(name: String, sex: String, age: Int) extends Serializable {

  var userName: String = name
  var userSex: String = sex
  var userAge: Int = age
  override def toString = s"User(userName=$userName, userSex=$userSex, userAge=$userAge)"

}
Scala客户端:
object ScalaRmiClient extends App {

  try {

    val remoteAddr="rmi://localhost:1103/UserScala"
    val userFacade = Naming.lookup(remoteAddr).asInstanceOf[UserScalaFacade]

    println(userFacade.getByName("ZhenJin"))
    System.out.println("--------------------------------------")
    for (user <- userFacade.getBySex("男")) println(user)

  } catch {
    case e: NotBoundException => println(e)
    case e: RemoteException => println(e)
    case e: MalformedURLException => println(e)
  }

}
Java客户端:
public class JavaRmiClient {

    public static void main(String[] args) {

        try {
            String remoteAddr="rmi://localhost:1103/UserScala";
            UserScalaFacade userFacade = (UserScalaFacade) Naming.lookup();

            User zhenJin = userFacade.getByName("ZhenJin");
            System.out.println(zhenJin);
            System.out.println("--------------------------------------");
            List<User> userList = userFacade.getBySex("男");
            System.out.println(userList);

        } catch (NotBoundException | RemoteException | MalformedURLException e) {
            e.printStackTrace();
        }
    }
}

上面试验可以证明Scala和Java是可以互通的,Scala本身也是可以直接引用Java类的

序列化简介

​ 序列化(Serialization)是将数据结构或对象状态转换为可以存储（例如，在文件或存储器缓冲区中）或传输（例如，通过网络连接）的格式的过程, 反序列化(Deserialization)则是从一系列字节中提取数据结构的相反操作.

序列化与反序列化

Kotlin实现rmi

服务端:
fun main(args: Array<String>) {
    try {
        val hello: HelloKotlinFacade = HelloKotlinFacadeImpl()
        LocateRegistry.createRegistry(1102)
        Naming.rebind("rmi://localhost:1101/HelloKotlin", hello)
        println("======= 启动RMI服务成功! =======")
    } catch (e: IOException) {
        e.printStackTrace()
    }
}
客户端:
fun main(args: Array<String>) {
    try {
        val hello = Naming.lookup("rmi://localhost:1102/HelloKotlin") as HelloKotlinFacade
        val response = hello.helloWorld("ZhenJin")
        println("=======> $response <=======")
    } catch (e: NotBoundException) {
        e.printStackTrace()
    } catch (e: RemoteException) {
        e.printStackTrace()
    } catch (e: MalformedURLException) {
        e.printStackTrace()
    }
}

实现和接口省略...

SpringBoot实现rmi

StringBoot通过配置就可以简单实现rmi了

服务端:
@Configuration
public class RmiServiceConfig {
    @Bean
    public RmiServiceExporter registerService(UserFacade userFacade) {
        RmiServiceExporter rmiServiceExporter = new RmiServiceExporter();
        rmiServiceExporter.setServiceName("UserInfo");
        rmiServiceExporter.setService(userFacade);
        rmiServiceExporter.setServiceInterface(UserFacade.class);
        rmiServiceExporter.setRegistryPort(1101);
        return rmiServiceExporter;
    }
}
客户端:
@Configuration
public class RmiClientConfig {

    @Bean
    public UserFacade userInfo() {
        RmiProxyFactoryBean rmiProxyFactoryBean = new RmiProxyFactoryBean();
        rmiProxyFactoryBean.setServiceUrl("rmi://localhost:1101/UserInfo");
        rmiProxyFactoryBean.setServiceInterface(UserFacade.class);
        rmiProxyFactoryBean.afterPropertiesSet();
        return (UserFacade) rmiProxyFactoryBean.getObject();
    }

}
客户端测试类:
@Autowired
private UserFacade userFacade;
    
@Test
public void userBySexTest() {
    try {
        List<User> userList = userFacade.getBySex("男");
        userList.forEach(System.out::println);
    } catch (RemoteException e) {
        e.printStackTrace();
    }
}

通过测试类可以看出,这和我们平时的程序调用内部方法没什么区别!

rmi调用过程

大家可以通过下面文章加深了解:

https://stuff.mit.edu/afs/athena/software/java/java_v1.2.2/distrib/sun4x_56/docs/guide/rmi/Factory.html

rmi工厂调用的过程

  • 有两个远程服务接口可供client调用，Factory和Product接口

  • FactoryImpl类实现了Factory接口，ProductImpl类实现了Product接口

    1. FactoryImpl被注册到了rmi-registry中
2. client端请求一个Factory的引用 
3. rmi-registry返回client端一个FactoryImpl的引用 
4. client端调用FactoryImpl的远程方法请求一个ProductImpl的远程引用
5. FactoryImpl返回给client端一个ProductImpl引用 
6. client通过ProductImpl引用调用远程方法

socket工厂文档:
https://docs.oracle.com/javase/8/docs/technotes/guides/rmi/socketfactory/index.html

Zookeeper实现rmi

出处:http://www.importnew.com/20344.html

安装Zookeeper

解压 ZooKeeper

tar -zxvf zookeeper-3.4.12.tar.gz

在 conf 目录新建 zoo.cfg

cd zookeeper-3.4.12/conf
vim zoo.cfg

zoo.cfg 代码如下（自己指定 log 文件目录）：

tickTime=2000
dataDir=/usr/local/zookeeper-3.4.12/data 
dataLogDir=/usr/local/zookeeper-3.4.12/log
clientPort=2181

在 bin 目录下，启动 Zookeeper：

cd zookeeper-3.4.12/bin
./zkServer.sh start

消费者:

public class RmiConsumer {

    // 用于等待 SyncConnected 事件触发后继续执行当前线程
    private CountDownLatch latch = new CountDownLatch(1);

    // 定义一个 volatile 成员变量，用于保存最新的 RMI 地址（考虑到该变量或许会被其它线程所修改，一旦修改后，该变量的值会影响到所有线程）
    private volatile List<String> urlList = new ArrayList<>();

    // 构造器
    public RmiConsumer() {
        ZooKeeper zk = connectServer(); // 连接 ZooKeeper 服务器并获取 ZooKeeper 对象
        if (zk != null) {
            watchNode(zk); // 观察 /registry 节点的所有子节点并更新 urlList 成员变量
        }
    }

    // 查找 RMI 服务
    public <T extends Remote> T lookup() {
        T service = null;
        int size = urlList.size();
        if (size > 0) {
            String url;
            if (size == 1) {
                url = urlList.get(0); // 若 urlList 中只有一个元素，则直接获取该元素
                log.debug("using only url: {}", url);
            } else {
                url = urlList.get(ThreadLocalRandom.current().nextInt(size)); // 若 urlList 中存在多个元素，则随机获取一个元素
                log.debug("using random url: {}", url);
            }
            service = lookupService(url); // 从 JNDI 中查找 RMI 服务
        }
        return service;
    }

    // 连接 ZooKeeper 服务器
    private ZooKeeper connectServer() {
        ZooKeeper zk = null;
        try {
            zk = new ZooKeeper(Constant.ZK_CONNECTION_STRING, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getState() == Event.KeeperState.SyncConnected) {
                        latch.countDown(); // 唤醒当前正在执行的线程
                    }
                }
            });
            latch.await(); // 使当前线程处于等待状态
        } catch (IOException | InterruptedException e) {
            log.error("", e);
        }
        return zk;
    }

    // 观察 /registry 节点下所有子节点是否有变化
    private void watchNode(final ZooKeeper zk) {
        try {
            List<String> nodeList = zk.getChildren(Constant.ZK_REGISTRY_PATH, event -> {
                if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
                    watchNode(zk); // 若子节点有变化，则重新调用该方法（为了获取最新子节点中的数据）
                }
            });
            List<String> dataList = new ArrayList<>(); // 用于存放 /registry 所有子节点中的数据
            for (String node : nodeList) {
                byte[] data = zk.getData(Constant.ZK_REGISTRY_PATH + "/" + node, false, null); // 获取 /registry 的子节点中的数据
                dataList.add(new String(data));
            }
            log.debug("node data: {}", dataList);
            urlList = dataList; // 更新最新的 RMI 地址
        } catch (KeeperException | InterruptedException e) {
            log.error("", e);
        }
    }

    // 在 JNDI 中查找 RMI 远程服务对象
    @SuppressWarnings("unchecked")
    private <T> T lookupService(String url) {
        T remote = null;
        try {
            remote = (T) Naming.lookup(url);
        } catch (NotBoundException | MalformedURLException | RemoteException e) {
            log.error("远程查找出错!", e);
        }
        return remote;
    }
}

生产者:

public class RmiProvider {


    /**
     * 用于等待 SyncConnected 事件触发后继续执行当前线程
     */
    private CountDownLatch latch = new CountDownLatch(1);

    // 发布 RMI 服务并注册 RMI 地址到 ZooKeeper 中
    public void publish(Remote remote, String host, int port) {
        String url = publishService(remote, host, port); // 发布 RMI 服务并返回 RMI 地址
        if (url != null) {
            ZooKeeper zk = connectServer(); // 连接 ZooKeeper 服务器并获取 ZooKeeper 对象
            if (zk != null) {
                createNode(zk, url); // 创建 ZNode 并将 RMI 地址放入 ZNode 上
            }
        }
    }

     /**
      *发布 RMI 服务
      */
    private String publishService(Remote remote, String host, int port) {
        String url = null;
        try {
            url = String.format("rmi://%s:%d/%s", host, port, remote.getClass().getName());
            LocateRegistry.createRegistry(port);
            Naming.rebind(url, remote);
            log.debug("publish rmi service (url: {})", url);
        } catch (RemoteException | MalformedURLException e) {
            log.error("", e);
        }
        return url;
    }

    // 连接 ZooKeeper 服务器
    private ZooKeeper connectServer() {
        ZooKeeper zk = null;
        try {
            zk = new ZooKeeper(Constant.ZK_CONNECTION_STRING, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getState() == Event.KeeperState.SyncConnected) {
                        latch.countDown(); // 唤醒当前正在执行的线程
                    }
                }
            });
            latch.await(); // 使当前线程处于等待状态
        } catch (IOException | InterruptedException e) {
            log.error("", e);
        }
        return zk;
    }

    /**
     * 创建节点
     */
    private void createNode(ZooKeeper zk, String url) {
        try {
            byte[] data = url.getBytes();
            String path = zk.create(Constant.ZK_PROVIDER_PATH, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);    // 创建一个临时性且有序的 ZNode
            log.debug("create zookeeper node ({} => {})", path, url);
        } catch (KeeperException | InterruptedException e) {
            log.error("", e);
        }
    }
}

图解:

zookeeper

代码已上传到GitHub上:https://github.com/jayknoxqu/rmi-example

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,053评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,527评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,779评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,685评论 1赞 276
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,699评论 5赞 366
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,609评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,989评论 3赞 396
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,654评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,890评论 1赞 298
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,634评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,716评论 1赞 330
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,394评论 4赞 319
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,976评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,950评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,191评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 44,849评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,458评论 2赞 342

推荐阅读更多精彩内容

  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 134,591评论 18赞 139
  • 代理模式——远程代理(Java RMI)
    代理模式——远程代理(Java RMI) @(设计模式) 一、远程代理——大使 在日常开发中，我们经常会有本地服务...
    理查德成阅读 5,512评论 0赞 7
  • 【0103 读书清单】蔡康永的说话之道
    1.转达赞美。 不要吝惜赞美，即便是转个三四手的赞美，也会让大家的关系更好。但这并不是奉承，只是为了让本来很讨人喜...
    我是婷玉呀阅读 241评论 2赞 2
  • Putty 终端字体和颜色修改方法
    当你已经保存了一个session时，可以通过以下方式修改终端字体和颜色。 先加载（load）你要修改的sessio...
    靖边候阅读 8,207评论 0赞 1
  • 给班长的信
    昨天007不写就出局，七班长高莹姐，发了一篇文章，关于这么长时间一路来，她所看到的007er们的践行状态的变化和心...
    金金视界阅读 391评论 0赞 1