在这个例子中我们会设计和编写个分布式JXTA应用可以解决并行计算问题我们将用个反复使用形式建立这个应用在每步中扩大它能力和 使用到API集这些原代码只能覆盖最重要部分需要全部代码可以到网站WebSite上查询

些大型计算问题子集可以用并行思路方法解决个工作并行执行意味着你可以将个问题分解为几个小子问题 这些子问题又可以被同时被执行当个子问题结束后它将返回自己结果到主主将这些返回小结果组合为更大答案

举个例子考虑将任何2个整数间质数列出来工作质数就是那些只能被它自己和1整除自然数 那些可以被1和其他比自己小数整除叫合数因此最简单思路方法产生个质数列表就是在自然数列表中消除所有合数剩余就是质数了

这个思路方法反复消除系列自然数中合数它将列表中每个数都用2到它平方根去除如果任何个数被这样除了的后没有余数那么这个数就是合数并且对它作上标记 当这些反复(工作)结束了的后所有被标记数将被消除剩下就是质数了

 

但是如果是个很长列表有数百万长度那么我们将它分为几个小点列表然后对每个列表执行上述思路方法每个子计算被分发到网络上区别机器去执行充分利用分布计算资源质数查询是这些可以被并行分解大型问题集中个现在流行P2P风格软件Software有不少比如SETI@HOME工程它目标是对来自外太空信号解码寻找外星智能生物类似工程还有让用户共享出他们空闲CPU资源去模拟蛋白质折叠或者解码DNA串

在这个应用中主将向用户请求2个数然后产生个在2数的间包含所以质数数列这个主将首先在JXTA网络上尝试找到提供质数查询服务其他peer然后将列表片段分发给它们去执行当个peer完成了它那部分它将返回这个片段中质数

为了这样分配能够进行我们要让所有peer能够在网络上通告(advertise)它具备质数查询能力以让其他peer可以找到并连接它

 

JXTA Application Design

可能这个应用最不平常个方面就是每个peer既扮演主(master process)角色和个从属角色(slave)也有可能个从属还要决定是否进步分解这个问题到子任务这种服务模式/客户模式操作是P2P设计实质我们将它定义为SM/CM操作

 

消息定义

当我们设计个JXTA时候我们必须忍受JXTA是个基于消息系统:2个peer中主要协议(contract)主要是通过消息这样设计第个工作就是定义消息传递在这个质数查询应用中个peer传递条包含2个边界数消息给另外个peer接受方计算在这2个边界数中质数返回到以前peer

net.jxta.endpo.Message 类 概括出个消息感念它允许赋予任何个消息集个键(KEY)我们将在下表中用键-值 来表示这个类例子

Table 16.1. Request Message
 
Key
 Value
 
ServiceConstants.LOW_INT
 Lower boundary of the (sub)list
 
ServiceConstants.HIGH_INT
 Upper boundary of the (sub)list
 


 

Table 16.2. Response Message
 
Key
 Value
 
ServiceConstants.LOW_INT
 Lower boundary of the (sub)list
 
ServiceConstants.HIGH_INT
 Upper boundary of the (sub)list
 
ServiceConstants.PRIMELIST
 A containing all primes between the bounds of the list. The primes are separated by ; characters.
 


服务定义和发现

下步我们必须定义个让主找到从属思路方法换句话说我们比如让peer预先了解到提供这个服务其他peer

就象早先提及那样个JXTA服务是用它module类和specication定义因此 我们将为这个质数查询服务module和specication定义个通告. 然后让个peer提供个让这些通告在JXTA网络中传播服务这个质数查询module类将采用JXTACLASS:com.sams.p2p.primecruncher这个名字modulespec将采用the name JXTASPEC:com.sams.p2p.primecruncher这个名字

主将用这个名字去发现通告module介绍说明

因此 除了消息定义 服务名字串也是peer在设计阶段应该获取信息所有 peer交互信息将在运行时候被发现

服务实现

当个质数查询服务开始时候它将化JXTA平台以得到去World and Net Peer Group通道当化只后peer将创造并发布它通告包括它module类和module介绍说明通告

模板介绍说明通告将包含个管道通告 那些发现有此服务个模板介绍说明通告客户端必须得到这个管道通告并且通过这个管道连接到那个服务

发布了通告的后我们服务打开个输入通道并且对进来消息进行监听当个消息到达的后这个服务尝试从这个消息中获得high和low这2个边界数字然后将它们传递到个仅产生质数链表组件中 当那个组件返回了结果(个含质数)这个质数查讯服务将产生个包含结果消息并将此消息送回客户端在最先反复中服务将简单地打印出它接受到消息接下来提炼中它将打开个管道将结果送回客户端客户端将把这些从各个peer中得到结果组合起来并将最后数列存入文件中.

Listing 16.2 Outline of PrimePeer and Initialization of a JXTA Peer

package primecruncher;

import net.jxta.peergroup.PeerGroup;

import net.jxta.peergroup.PeerGroupFactory;

import net.jxta.peergroup.PeerGroupID;

import net.jxta.discovery.DiscoveryService;

import net.jxta.pipe.PipeService;

import net.jxta.pipe.InputPipe;

import net.jxta.pipe.PipeID;

import net.jxta.exception.PeerGroupException;

import net.jxta.protocol.ModuleClassAdvertisement;

import net.jxta.protocol.ModuleSpecAdvertisement;

import net.jxta.protocol.PipeAdvertisement;

import net.jxta.document.*;

import net.jxta.platform.ModuleClassID;

import net.jxta.platform.ModuleSpecID;

import net.jxta.id.IDFactory;

import net.jxta.endpo.Message;

import java.io.FileInputStream;

import java.io.IOException;

import java.io.FileOutputStream;

import java.io.StringWriter;

public PrimePeer {

        private PeerGroup group;

        private DiscoveryService discoSvc;

        private PipeService pipeSvc;

        private InputPipe inputPipe;

        private final String PIPE_ADV_FILE = "primeserver_pipe.adv";

        public void (String argv) {

               PrimePeer pp = PrimePeer;

               pp.startJxta;

               pp.doAdvertise;

               pp.startService;

        }

        public PrimePeer {

        }

        private void startJxta {

               try {

                       group = PeerGroupFactory.NetPeerGroup;

                       discoSvc = group.getDiscoveryService;

                       pipeSvc = group.getPipeService;

               } catch (PeerGroupException e) {

                       .out.prln("Cannot create Net Peer Group: " + e.getMessage(

));

                       .exit(-1);

               }

        }

 

      /**

       * Create and propagate advertisements

       */

        private void doAdvertise {

           ...

        }

        /*

         * Start up the service, listen for incoming messages on the service's input pipe.

         */

        private void startService {

           ...

        }

    /**

         * Compute the requested list of prime numbers.

          */

        private void processInput(String high, String low) {

           ...

        }

}

在这个startJxta服务化思路方法中我们首先获得个通往World Peer Group引用(reference):这是通过个静态PeerGroupFacrory实现这个思路方法将为JXTA运行作好准备下步中我们将获得个到2个Net Peer Group提供peer group服务引用:DiscoveryService和the PipeService我们将在创建服务通告时候用到它们2个