基于Java的2D mmorpg开源引擎Threerings系列之二（分布式对象）

通过我们上一篇的介绍，大家应该对Threerings这个引擎有了一个初步的认识。在引擎的核心框架之一的Narya中，主要包括了presents，crowd和bureau三个package，而presents则包括了今天我们要介绍的DObject部分。

 

presents这个包或者说框架是对底层网络通讯的一层封装，将底层的网络通讯的实现细节抽象成对象与事件（object和event），以供建造在其上的游戏应用能够方便的调用。在深入讨论presents框架对构造在其上的应用所提供的服务之前，我们不妨先来看一下下面所列的几点，对于一个mmorpg游戏来说需要一种什么样的网络通讯机制。

 

1）单一的服务器上连接有大量的客户端。

2）通讯不仅发生在服务端与客户端，还发生在客户端与客户端之间，但客户端与客户端的通讯也必须通过服务器。

3）前面已经讨论过，客户端无权直接修改共享的敏感数据，而只有向服务端发出请求，服务端验证后才会修改该数据。

 

而presents框架正是通过分布式对象（distributed objects）即我们所谓的DObject机制来满足上述需求的。presents框架所提供的服务允许构造在其上的游戏应用通过这种DObject机制来访问需要共享的信息。DObject对象由服务端来维护，客户端通过订阅DObject对象获得一个该对象的本地代理，当状态发生变化时客户端通过接收来自服务端的“事件”来更新本地的代理对象。

 

客户端无法直接修改本地的代理对象，而是通过将请求封装在“事件”中，通过调用本地代理对象的setter方法将该请求发送到服务端供进一步处理。服务端在验证了该请求的合法性之后会把封装了该请求的“事件”应用到维护在服务端的主DObject中，随后会把该“事件”分发给所有订阅了该DObject的所有客户端。包括发起请求的客户端之内的所有客户端再将该“事件”应用到本地的代理对象上。正是通过这样的机制所有的客户端可以维护到最新的对象数据。

 

      

 

 

定义DObject对象

 

定义一个DObject对象就如同定义一个普通的java对象一样，只是在定义完之后需要通过ant运行一个该引擎自带的代码生成工具，该工具会自动生成并插入运行DObject系统所必须的一些方法和常量。一个刚定义好的DObject对象看起来是这样的：

  

    public class CageObject extends DObject
    {
        /** The number of monkeys in the cage. */
        public int monkeys;

        /** The name of the owner of this cage. */
        public String owner;
    }

 

 

 在DObject中所有的非transient并且是public修饰符的属性都会被该引擎自带的代码生成工具捕获，并生成用于DObject系统相应的方法和常量。所有非public或者有transient修饰符的属性会被忽略，也就是说当订阅者通过网络从服务器接收一个该DObject对象的本地代理时这些属性并不会通过网络被接收到。

 

当我们运行完代码生成工具后，我们定义的对象会变成这样： 

 

    public class CageObject extends DObject
    {
        // AUTO-GENERATED: FIELDS START
        /** The field name of the monkeys field. */                                      
        public static final String MONKEYS = "monkeys";

        /** The field name of the owner field. */
        public static final String OWNER = "owner";
        // AUTO-GENERATED: FIELDS END

 

        /** The number of monkeys in the cage. */
        public int monkeys;

 

        /** The name of the owner of this cage. */
        public String owner;

 

        // AUTO-GENERATED: METHODS START
        /**
         * Requests that the monkeys field be set to the
         * specified value. The local value will be updated immediately and an
         * event will be propagated through the system to notify all listeners
         * that the attribute did change. Proxied copies of this object (on
         * clients) will apply the value change when they received the
         * attribute changed notification.
         */
        public void setMonkeys (int value)
        {
            int ovalue = this.monkeys;
            requestAttributeChange(
                EVEN_BASE, new Integer(value), new Integer(ovalue));
            this.monkeys = value;
        }

        /**
         * Requests that the owner field be set to the
         * specified value. The local value will be updated immediately and an
         * event will be propagated through the system to notify all listeners
         * that the attribute did change. Proxied copies of this object (on
         * clients) will apply the value change when they received the
         * attribute changed notification.
         */
        public void setOwner (String value)
        {
            String ovalue = this.owner;
            requestAttributeChange(
                ODD_BASE, value, ovalue);
            this.owner = value;
        }
        // AUTO-GENERATED: METHODS END
    }

 

黑体部分的代码就是工具为我们所生成的，其中包括属性的setter方法和常量的定义，当我们从服务器接收到属性更新的事件后，就是通过这些常量来区别具体是哪个属性的状态被更新。而且只要AUTO-GENERATED块中的内容不要去手工修改的话，你可以重复添加或修改对象的属性，代码生成工具会自动帮你生成该属性所对应的setter方法和常量定义，而所有在AUTO-GENERATED块之外的内容都会保持不变。

 

看的仔细的读者也许会发现，在setter方法中，在发出属性更新的请求后，新的值会被立即写入到该DObject对象的本地副本中。这是因为有网络延迟，客户端往往还没有等到服务器传来的属性更新“事件”就假设属性已经被成功修改，从而导致运行时产生错误。Threerings小组在多次经历这样的问题之后不得不对框架做出了修改，通过将新的值立即写入到DObject对象的本地副本的属性中来解决这些问题。但是反过来也说明，在一般情况下，DObject对象属性的更新并不是直接由客户端触发，而是由服务端在接收到从客户端发来的可以导致DObject对象属性发生变化的请求时触发的。

 

创建DObject对象

 

你可以象创建一个普通的java对象一样创建一个DObject对象，然后通过DObject Manager来注册它。但要注意的是你必须在server端使用RootDObjectManager对象来注册它，因为它的父类 DObjectManager中并没有registerObject这个方法，如果在客户端的话，你是无法创建DObject对象的（准确的说你也可以创建，但它仅仅只是一个普通的java对象而已，并不具备框架所赋予它的DObject对象的机制），而只能通过调用服务端的服务来实现（通过 InvocationService，会在下一篇中详细介绍）。

 

 

public class ServerEntity {
    public void init (RootDObjectManager omgr) {
        _object = omgr.registerObject(new CageObject());
    }
    protected CageObject _object;
}

 

 

订阅DObject对象

 

客户端通过订阅来获取一个DObject对象的本地代理

 

 

    public class ObjectUser implements Subscriber {
        public void init (Client client, int objectId) {
            client.getDObjectManager().subscribeToObject(objectId, this);
        }

        // inherited from interface Subscriber
        public void objectAvailable (DObject object) {
            // yay! we got our object
            _object = (CageObject)object;
        }

        // inherited from interface Subscriber
        public void requestFailed (int oid, ObjectAccessException cause) {
            // oh the humanity, we failed to subscribe
        }

        protected CageObject _object;
    }

 

 

之后可以通过类似的机制来解除对该DObject对象的订阅

 

 

    public class ObjectUser implements Subscriber {
        // ...
        public void shutdown (Client client) {
            client.getDObjectManager().unsubscribeFromObject(
                _object.getOid(), this);
            _object = null;
        }
        // ...
    }

 

 

既然说到这里，我们也不妨再多说几句，即在一个异步的分布式环境中，并不能保证在ObjectUser中，对DObject对象的订阅请求一定会在你调用shutdown之前被处理，如果出现这种情况，那么在前面的例子中，你会得到一个null pointer异常，更糟糕的情况是当你以为你的DObject对象已经被解除订阅了，而实际上却没有。为了解决这个问题，框架引入了一个SafeSubscriber类。

 

 

 

    public class ObjectUser implements Subscriber {
        public void init (Client client, int objectId) {
            _safesub = new SafeSubscriber(objectId, this);
            _safesub.subcribe(client.getDObjectManager());
        }

        // inherited from interface Subscriber
        public void objectAvailable (DObject object) {
            // yay! we got our object
            _object = (CageObject)object;
        }

        // inherited from interface Subscriber
        public void requestFailed (int oid, ObjectAccessException cause) {
            // oh the humanity, we failed to subscribe
        }

        public void shutdown (Client client) {
            _safesub.unsubscribe(client.getDObjectManager());
            _object = null;
        }

        protected SafeSubscriber _safesub;
        protected CageObject _object;
    }

 

SafeSubscriber类中的subscribe以及unsubscribe方法是对DObjectManager中的subscribeToObject以及unsubscribeFromObject两个方法的封装，使用SafeSubscriber类，它能保证在解除订阅之前订阅请求一定会先行被处理，甚至在先行的订阅请求失败等复杂情况下仍然能保证请求能够被正确处理。

 

事件侦听

 

当DObject对象被成功订阅后，所有有关于此DObject对象的事件都会被派送到位于客户端的本地代理对象上。如果需要动态的对不同事件作出相应的反应，那么可以使用侦听器（listener）。在客户端上可以注册任意数目的侦听器，当DObject对象被解除订阅后，所有注册的侦听器也即随之而去。

 

AttributeChangeListener最常见的侦听器之一，当我们所侦听的DObject属性发生变化的时候即会被通知到，还是来看一下下面的例子。

 

    public class ObjectUser
        implements Subscriber, AttributeChangeListener {
        // ...
        public void init (Client client, int objectId) {
            _safesub = new SafeSubscriber(_subscriber, objectId);
            _safesub.subcribe(client.getDObjectManager());
        }

        // inherited from interface Subscriber
        public void objectAvailable (DObject object) {
            // yay! we got our object
            _object = (CageObject)object;
            _object.addListener(this);
        }

        // inherited from interface Subscriber
        public void requestFailed (int oid, ObjectAccessException cause) {
            // oh the humanity, we failed to subscribe
        }

        // inherited from interface AttributeChangeListener
        public void attributeChanged (AttributeChangedEvent event)
        {
            System.out.println("Wow! The " + event.getName() +
                               " field changed to " + event.getValue() + ".");
        }

        public void shutdown (Client client) {
            _safesub.unsubscribe(client.getDObjectManager());
            if (_object != null) {
                 // removing our listener not necessary as we are
                 // unsubscribing, but it's a good habit to develop as
                 // frequently listeners will come and go during the
                 // lifetime of an object subscription
                _object.removeListener(this);
                _object = null;
            }
        }

        protected SafeSubscriber _safesub;
        protected CageObject _object;
    }

 

在当前的分布式系统中，当有任何一方调用了CageObject的setter方法后都会生成一个attributeChange事件并发送到服务器，服务器处理完后会重新派发此事件到所有的CageObject对象的订阅者，若客户端注册了attributeChage事件的侦听器后，此事件就会被该侦听器所捕获，并调用attributeChanged()方法。在这里大家不要把subscriber和listener两个概念搞混，最关键的区别是不管有没有注册某个DObject侦听器，只要订阅了该DObject那么所有的事件都会通过Presents系统被派送并应用到该DObject的代理对象上。另外还需要注意的一点是不仅在客户端上可以注册事件侦听器，在服务端同样可以，一旦事件通过网络被派发就会被立即通知到。

 

其次还需要知道的一点是侦听器是在事件被应用到对象之后才会被通知到。之前的属性值可以通过AttributeChangedEvent.getOldValue()方法来得到，不过在实际当中好像很少需要知道这个oldValue。

 

分布式集合属性

 

在前面的例子中我们一直使用primitive类型作为DObject的分布式属性，但在某些情况下我们还是需要引入更复杂的数据结构来作为DObject的分布式属性。接下来我们会介绍Presents框架所支持的两种集合类型，即sets和arrays来作为我们的DObject对象的分布式属性，通过框架提供的机制，使用起来就如同是primitive属性一般。

 

分布式数组

 

在DObject中使用元素为primitive类型的数组，在使用代码生成工具时会被侦测到，而其自动生成的代码提供了一种既可以更新整个数组也可以一次只更新数组中单个元素的机制。

 

 

    public class ChessObject extends DObject
    {
        // AUTO-GENERATED: FIELDS START
        /** The field name of the state field. */
        public static final String STATE = "state";
        // AUTO-GENERATED: FIELDS END

        /** Used to track our board state. */
        public int[] state;

        // AUTO-GENERATED: METHODS START
        /**
         * Requests that the state field be set to the
         * specified value. The local value will be updated immediately and an
         * event will be propagated through the system to notify all listeners
         * that the attribute did change. Proxied copies of this object (on
         * clients) will apply the value change when they received the
         * attribute changed notification.
         */
        public void setState (int[] value)
        {
            int[] ovalue = this.state;
            requestAttributeChange(
                STATE, value, ovalue);
            this.state = (value == null) ? null : (int[])value.clone();
        }

        /**
         * Requests that the indexth element of
         * state field be set to the specified value.
         * The local value will be updated immediately and an event will be
         * propagated through the system to notify all listeners that the
         * attribute did change. Proxied copies of this object (on clients)
         * will apply the value change when they received the attribute
         * changed notification.
         */
        public void setStateAt (int value, int index)
        {
            int ovalue = this.state[index];
            requestElementUpdate(
                STATE, index, new Integer(value), new Integer(ovalue));
            this.state[index] = value;
        }
        // AUTO-GENERATED: METHODS END
    }

 

针对数组单个元素的更新，我们可以使用ElementUpdateListener侦听器来侦听。当数组中的单个元素更新后，实现了这个接口的侦听器会自动被通知。不过当调用setState()方法更新整个数组后，我们还是要使用普通的AttributeChangeListener侦听器来侦听。

 

在使用数组作为分布式对象还要注意下数组的界标，比如当发出请求更新数组索引为9的元素的时候，请确保你的数组至少有10个元素，否则会抛出一个数组下标越界的异常。实际上在使用数组的时候并非只能使用primitive作为元素类型，而可以是任何实现Streamable接口的对象，对于Streamable接口我们在接下来的一节中会详细介绍。

 

Streamable接口和SimpleStreamableObject对象

 

Streamable接口是用来标记那些可以通过网络来传输的对象，并且可以使用在DObject中作为分布式数组中的元素。同java中的Serializable接口类似，底层的对象序列化是通过反射来实现的。在对象序列化的时候，只要没有标记成transient的属性都会被序列化。请看下面的例子:

 

 

    public class Player implements Streamable
    {
        /** This player's name. */
        public String name;

        /** This player's rating. */
        public int rating;
    }

    public class ChessObject extends DObject
    {
        /** A record for each player in the game. */
        public Player[] players;
    }

 

自动生成的代码在这里就省略了，不过你可以想象一下象这样两个方法setPlayers(Player[] value)和setPlayersAt(Player value, int index)会被生成并包含该方法中应有的代码。在这里需要指出的是实现了streamable接口的对象在网络上传输的时候是整个的对象在传输，而不只是更新了的单个属性，估计是因为这样做太复杂了而且用处也不大。如果带宽是首要考虑目标，你可以自己从DEvent类继承并定制一个专门的event类来控制什么需要被传输，什么不需要，这里就不展开讨论了。SimpleStreamableObject类是Streamable接口的一个简单实现，它使用反射默认实现了toString()方法，可以打印出属性的实际值（在调试和日志输出的时候比较有用）。

 

Distributed Sets

 

如果是开发一个分布式系统的话，经常碰到的一个情况是需要一个可以自由添加元素的分布式的对象集合，而这个集合中元素的排列顺序通常却并不重要。为了满足这种需求，框架为我们提供了分布式的集合类DSet。一个DSet对象往往包含了多个元素，我们把它叫做entry，每个entry必须要实现DSet.Entry接口，而一旦实现这个接口也即自动实现了Streamable接口；实现DSet.Entry接口的元素还必须提供一个Comparable key用来区别DSet中的其他元素（通过key还可以使用效率比较高的二分查询算法）。若在DObject中使用了DSet的话，除了set之外自动生成的代码还会包括addTo，update以及removeFrom三个方法，比如我们来看下面的这个例子：

 

 

    public class Monkey implements DSet.Entry
    {
        /** The monkey's name. */
        public String name;

        /** The monkey's age. */
        public int age;

        // documentation inherited from interface DSet.Entry
        public Comparable getKey ()
        {
            return name;
        }
    }

    public class CageObject extends DObject
    {
        // AUTO-GENERATED: FIELDS START
        /** The field name of the monkeys field. */
        public static final String MONKEYS = "monkeys";
        // AUTO-GENERATED: FIELDS END

        /** A collection of monkeys. */
        public DSet monkeys;

        // AUTO-GENERATED: METHODS START
        /**
         * Requests that the specified entry be added to the
         * monkeys set.
         */
        public void addToMonkeys (DSet.Entry elem)
        {
            requestEntryAdd(MONKEYS, monkeys, elem);
        }

        /**
         * Requests that the entry matching the supplied key be removed from
         * the monkeys set.
         */
        public void removeFromMonkeys (Comparable key)
        {
            requestEntryRemove(MONKEYS, monkeys, key);
        }

        /**
         * Requests that the specified entry be updated in the
         * monkeys set.
         */
        public void updateMonkeys (DSet.Entry elem)
        {
            requestEntryUpdate(MONKEYS, monkeys, elem);
        }

        /**
         * Requests that the monkeys field be set to the
         * specified value.
         */
        public void setMonkeys (DSet value)
        {
            requestAttributeChange(MONKEYS, value, this.monkeys);
            this.monkeys = (value == null) ? null : (DSet)value.clone();
        }
        // AUTO-GENERATED: METHODS END
    }

 

当然了，我们可以直接更新整个set的值，但更多的仅仅只是往set中增加新的entry，更新set中entry的值，或者删除某个entry。与DSet配套的自然有它相应的侦听器（SetListener），使用这个侦听器的话，当一个distributed set被更改的时候会自动被通知到。这个侦听器具体使用与前面介绍其他两个侦听器非常类似，这里就不再举例说明了。