带弹簧的Hazelcast分布式执行

ExecutorService功能是Java 5附带的，位于java.util.concurrent包中。它扩展了Executor接口，并提供了线程池功能来执行异步简短任务。建议使用Java Executor服务类型来查看基本的ExecutorService实现。

此外，ThreadPoolExecutor是ExecutorService接口的非常有用的实现。它扩展了AbstractExecutorService，提供了ExecutorService执行方法的默认实现。当执行大量异步任务时，它可以提供改进的性能，并维护基本统计信息，例如已完成任务的数量。

还建议如何使用Spring开发和监视线程池服务，以研究如何开发和监视线程池服务。

到目前为止，我们刚刚谈到了“分布式执行器服务”的实现。让我们还调查分布式执行器服务。

Hazelcast Distributed Executor Service功能是java.util.concurrent.ExecutorService的分布式实现。它允许在集群中执行业务逻辑。有四种替代方法可以实现它：

可以在选择的特定群集成员上执行该逻辑。
逻辑可以在拥有所选密钥的成员上执行。
该逻辑可以在Hazelcast将挑选的成员上执行。
逻辑可以在全部或部分集群成员上执行。

本文说明如何通过Hazelcast和Spring开发分布式执行器服务。

二手技术：

JDK 1.7.0_09
Spring3.1.3
Hazelcast 2.4
Maven的3.0.4

步骤1：建立已完成的专案

创建一个Maven项目，如下所示。（可以使用Maven或IDE插件来创建它）。

步骤2：图书馆

首先，将Spring依赖项添加到Maven的pom.xml中

<properties><spring.version>3.1.3.RELEASE</spring.version><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><!-- Spring 3 dependencies --><dependency><groupId>org.springframework</groupId><artifactId>spring-core</artifactId><version>${spring.version}</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>${spring.version}</version></dependency><!-- Hazelcast library --><dependency><groupId>com.hazelcast</groupId><artifactId>hazelcast-all</artifactId><version>2.4</version></dependency><!-- Log4j library --><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.16</version></dependency></dependencies>

maven-compiler-plugin （Maven插件）用于使用JDK 1.7编译项目

<plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.0</version><configuration><source>1.7</source><target>1.7</target></configuration></plugin>

maven-shade-plugin （Maven插件）可用于创建runnable-jar

<plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>2.0</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><transformers><transformerimplementation='org.apache.maven.plugins.shade.resource.
ManifestResourceTransformer'><mainClass>com.onlinetechvision.exe.Application</mainClass></transformer><transformerimplementation='org.apache.maven.plugins.shade.resource.
AppendingTransformer'><resource>META-INF/spring.handlers</resource></transformer><transformerimplementation='org.apache.maven.plugins.shade.resource.
AppendingTransformer'><resource>META-INF/spring.schemas</resource></transformer></transformers></configuration></execution></executions></plugin>

第三步：创建客户豆

创建一个新的Customer bean。该bean将分布在OTV集群中的两个节点之间。在以下示例中，所有定义的属性（id，名称和姓氏）的类型均为String，并且已实现标准java.io.Serializable接口以进行序列化。如果使用自定义或第三方对象类型，则可以实现com.hazelcast.nio.DataSerializable接口以获得更好的序列化性能。

package com.onlinetechvision.customer;import java.io.Serializable;/*** Customer Bean.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class Customer implements Serializable {private static final long serialVersionUID = 1856862670651243395L;private String id;private String name;private String surname;public String getId() {return id;}public void setId(String id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public String getSurname() {return surname;}public void setSurname(String surname) {this.surname = surname;}@Overridepublic int hashCode() {final int prime = 31;int result = 1;result = prime * result + ((id == null) ? 0 : id.hashCode());result = prime * result + ((name == null) ? 0 : name.hashCode());result = prime * result + ((surname == null) ? 0 : surname.hashCode());return result;}@Overridepublic boolean equals(Object obj) {if (this == obj)return true;if (obj == null)return false;if (getClass() != obj.getClass())return false;Customer other = (Customer) obj;if (id == null) {if (other.id != null)return false;} else if (!id.equals(other.id))return false;if (name == null) {if (other.name != null)return false;} else if (!name.equals(other.name))return false;if (surname == null) {if (other.surname != null)return false;} else if (!surname.equals(other.surname))return false;return true;}@Overridepublic String toString() {return 'Customer [id=' + id + ', name=' + name + ', surname=' + surname	+ ']';}}

步骤4：建立ICacheService介面

为服务层创建了一个新的ICacheService接口，以公开缓存功能。

package com.onlinetechvision.cache.srv;import com.hazelcast.core.IMap;
import com.onlinetechvision.customer.Customer;/*** A new ICacheService Interface is created for service layer to expose cache functionality.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public interface ICacheService {/*** Adds Customer entries to cache** @param String key* @param Customer customer**/void addToCache(String key, Customer customer);/*** Deletes Customer entries from cache** @param String key**/void deleteFromCache(String key);/*** Gets Customer cache** @return IMap Coherence named cache*/IMap<String, Customer> getCache();
}

步骤5：创建CacheService实现

CacheService是ICacheService接口的实现。

package com.onlinetechvision.cache.srv;import com.hazelcast.core.IMap;
import com.onlinetechvision.customer.Customer;
import com.onlinetechvision.test.listener.CustomerEntryListener;/*** CacheService Class is implementation of ICacheService Interface.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class CacheService implements ICacheService {private IMap<String, Customer> customerMap;/*** Constructor of CacheService** @param IMap customerMap**/@SuppressWarnings('unchecked')public CacheService(IMap<String, Customer> customerMap) {setCustomerMap(customerMap);getCustomerMap().addEntryListener(new CustomerEntryListener(), true);}/*** Adds Customer entries to cache** @param String key* @param Customer customer**/@Overridepublic void addToCache(String key, Customer customer) {getCustomerMap().put(key, customer);}/*** Deletes Customer entries from cache** @param String key**/@Overridepublic void deleteFromCache(String key) {getCustomerMap().remove(key);}/*** Gets Customer cache** @return IMap Coherence named cache*/@Overridepublic IMap<String, Customer> getCache() {return getCustomerMap();}public IMap<String, Customer> getCustomerMap() {return customerMap;}public void setCustomerMap(IMap<String, Customer> customerMap) {this.customerMap = customerMap;}}

步骤6：建立IDistributedExecutorService接口

为服务层创建了一个新的IDistributedExecutorService接口，以公开分布式执行功能。

package com.onlinetechvision.executor.srv;import java.util.Collection;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;import com.hazelcast.core.Member;/*** A new IDistributedExecutorService Interface is created for service layer to expose distributed execution functionality.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public interface IDistributedExecutorService {/*** Executes the callable object on stated member** @param Callable callable* @param Member member* @throws InterruptedException* @throws ExecutionException**/String executeOnStatedMember(Callable<String> callable, Member member) throws InterruptedException, ExecutionException;/*** Executes the callable object on member owning the key** @param Callable callable* @param Object key* @throws InterruptedException* @throws ExecutionException**/String executeOnTheMemberOwningTheKey(Callable<String> callable, Object key) throws InterruptedException, ExecutionException;/*** Executes the callable object on any member** @param Callable callable* @throws InterruptedException* @throws ExecutionException**/String executeOnAnyMember(Callable<String> callable) throws InterruptedException, ExecutionException;/*** Executes the callable object on all members** @param Callable callable* @param Set all members* @throws InterruptedException* @throws ExecutionException**/Collection<String> executeOnMembers(Callable<String> callable, Set<Member> members) throws InterruptedException, ExecutionException;
}

步骤7：创建DistributedExecutorService实现

DistributedExecutorService是IDistributedExecutorService接口的实现。

package com.onlinetechvision.executor.srv;import java.util.Collection;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;import org.apache.log4j.Logger;import com.hazelcast.core.DistributedTask;
import com.hazelcast.core.Member;
import com.hazelcast.core.MultiTask;/*** DistributedExecutorService Class is implementation of IDistributedExecutorService Interface.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class DistributedExecutorService implements IDistributedExecutorService {private static final Logger logger = Logger.getLogger(DistributedExecutorService.class);private ExecutorService hazelcastDistributedExecutorService;/*** Executes the callable object on stated member** @param Callable callable* @param Member member* @throws InterruptedException* @throws ExecutionException**/@SuppressWarnings('unchecked')public String executeOnStatedMember(Callable<String> callable, Member member) throws InterruptedException, ExecutionException {logger.debug('Method executeOnStatedMember is called...');ExecutorService executorService = getHazelcastDistributedExecutorService();FutureTask<String> task = (FutureTask<String>) executorService.submit( new DistributedTask<String>(callable, member));String result = task.get();logger.debug('Result of method executeOnStatedMember is : ' + result);return result;}/*** Executes the callable object on member owning the key** @param Callable callable* @param Object key* @throws InterruptedException* @throws ExecutionException**/@SuppressWarnings('unchecked')public String executeOnTheMemberOwningTheKey(Callable<String> callable, Object key) throws InterruptedException, ExecutionException {logger.debug('Method executeOnTheMemberOwningTheKey is called...');ExecutorService executorService = getHazelcastDistributedExecutorService();FutureTask<String> task = (FutureTask<String>) executorService.submit(new DistributedTask<String>(callable, key));String result = task.get();logger.debug('Result of method executeOnTheMemberOwningTheKey is : ' + result);return result;}/*** Executes the callable object on any member** @param Callable callable* @throws InterruptedException* @throws ExecutionException**/public String executeOnAnyMember(Callable<String> callable) throws InterruptedException, ExecutionException {logger.debug('Method executeOnAnyMember is called...');ExecutorService executorService = getHazelcastDistributedExecutorService();Future<String> task = executorService.submit(callable);String result = task.get();logger.debug('Result of method executeOnAnyMember is : ' + result);return result;}/*** Executes the callable object on all members** @param Callable callable* @param Set all members* @throws InterruptedException* @throws ExecutionException**/public Collection<String> executeOnMembers(Callable<String> callable, Set<Member> members) throws ExecutionException, InterruptedException {logger.debug('Method executeOnMembers is called...');MultiTask<String> task = new MultiTask<String>(callable, members);ExecutorService executorService = getHazelcastDistributedExecutorService();executorService.execute(task);Collection<String> results = task.get();logger.debug('Result of method executeOnMembers is : ' + results.toString());return results;}public ExecutorService getHazelcastDistributedExecutorService() {return hazelcastDistributedExecutorService;}public void setHazelcastDistributedExecutorService(ExecutorService hazelcastDistributedExecutorService) {this.hazelcastDistributedExecutorService = hazelcastDistributedExecutorService;}}

步骤8：建立TestCallable类别

TestCallable类显示要执行的业务逻辑。

集群的第一个成员的TestCallable任务：

package com.onlinetechvision.task;import java.io.Serializable;
import java.util.concurrent.Callable;/*** TestCallable Class shows business logic to be executed.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class TestCallable implements Callable<String>, Serializable{private static final long serialVersionUID = -1839169907337151877L;/*** Computes a result, or throws an exception if unable to do so.** @return String computed result* @throws Exception if unable to compute a result*/public String call() throws Exception {return 'First Member' s TestCallable Task is called...';}}

集群第二个成员的TestCallable任务：

package com.onlinetechvision.task;import java.io.Serializable;
import java.util.concurrent.Callable;/*** TestCallable Class shows business logic to be executed.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class TestCallable implements Callable<String>, Serializable{private static final long serialVersionUID = -1839169907337151877L;/*** Computes a result, or throws an exception if unable to do so.** @return String computed result* @throws Exception if unable to compute a result*/public String call() throws Exception {return 'Second Member' s TestCallable Task is called...';}}

步骤9：创建AnotherAvailableMemberNotFoundException类

当找不到另一个可用成员时，将引发AnotherAvailableMemberNotFoundException 。为避免此异常，应在第二个节点之前启动第一个节点。

package com.onlinetechvision.exception;/*** AnotherAvailableMemberNotFoundException is thrown when another available member is not found.* To avoid this exception, first node should be started before the second node.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class AnotherAvailableMemberNotFoundException extends Exception {private static final long serialVersionUID = -3954360266393077645L;/*** Constructor of AnotherAvailableMemberNotFoundException** @param  String Exception message**/public AnotherAvailableMemberNotFoundException(String message) {super(message);}}

步骤10：建立CustomerEntryListener类别

CustomerEntryListener类侦听命名的缓存对象上的条目更改。

package com.onlinetechvision.test.listener;import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.EntryListener;/*** CustomerEntryListener Class listens entry changes on named cache object.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
@SuppressWarnings('rawtypes')
public class CustomerEntryListener implements EntryListener {/*** Invoked when an entry is added.** @param EntryEvent**/public void entryAdded(EntryEvent ee) {System.out.println('EntryAdded... Member : ' + ee.getMember() + ', Key : '+ee.getKey()+', OldValue : '+ee.getOldValue()+', NewValue : '+ee.getValue());}/*** Invoked when an entry is removed.** @param EntryEvent**/public void entryRemoved(EntryEvent ee) {System.out.println('EntryRemoved... Member : ' + ee.getMember() + ', Key : '+ee.getKey()+', OldValue : '+ee.getOldValue()+', NewValue : '+ee.getValue());}/*** Invoked when an entry is evicted.** @param EntryEvent**/public void entryEvicted(EntryEvent ee) {}	/*** Invoked when an entry is updated.** @param EntryEvent**/public void entryUpdated(EntryEvent ee) {}}

步骤11：创建入门类

入门类会加载客户以缓存和执行分布式任务。

集群的第一个成员的入门类：

package com.onlinetechvision.exe;import com.onlinetechvision.cache.srv.ICacheService;
import com.onlinetechvision.customer.Customer;/*** Starter Class loads Customers to cache and executes distributed tasks.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class Starter {private	ICacheService cacheService;/*** Loads cache and executes the tasks**/public void start() {loadCacheForFirstMember();}/*** Loads Customers to cache**/public void loadCacheForFirstMember() {Customer firstCustomer = new Customer();firstCustomer.setId('1');firstCustomer.setName('Jodie');firstCustomer.setSurname('Foster');Customer secondCustomer = new Customer();secondCustomer.setId('2');secondCustomer.setName('Kate');secondCustomer.setSurname('Winslet');getCacheService().addToCache(firstCustomer.getId(), firstCustomer);getCacheService().addToCache(secondCustomer.getId(), secondCustomer);}public ICacheService getCacheService() {return cacheService;}public void setCacheService(ICacheService cacheService) {this.cacheService = cacheService;}}

集群第二个成员的入门类：

package com.onlinetechvision.exe;import java.util.Set;
import java.util.concurrent.ExecutionException;import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.Member;
import com.onlinetechvision.cache.srv.ICacheService;
import com.onlinetechvision.customer.Customer;
import com.onlinetechvision.exception.AnotherAvailableMemberNotFoundException;
import com.onlinetechvision.executor.srv.IDistributedExecutorService;
import com.onlinetechvision.task.TestCallable;/*** Starter Class loads Customers to cache and executes distributed tasks.** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class Starter {private String hazelcastInstanceName;private Hazelcast hazelcast;private IDistributedExecutorService distributedExecutorService;private	ICacheService cacheService;/*** Loads cache and executes the tasks**/public void start() {loadCache();executeTasks();}/*** Loads Customers to cache**/public void loadCache() {Customer firstCustomer = new Customer();firstCustomer.setId('3');firstCustomer.setName('Bruce');firstCustomer.setSurname('Willis');Customer secondCustomer = new Customer();secondCustomer.setId('4');secondCustomer.setName('Colin');secondCustomer.setSurname('Farrell');getCacheService().addToCache(firstCustomer.getId(), firstCustomer);getCacheService().addToCache(secondCustomer.getId(), secondCustomer);}/*** Executes Tasks**/public void executeTasks() {try {getDistributedExecutorService().executeOnStatedMember(new TestCallable(), getAnotherMember());getDistributedExecutorService().executeOnTheMemberOwningTheKey(new TestCallable(), '3');getDistributedExecutorService().executeOnAnyMember(new TestCallable());getDistributedExecutorService().executeOnMembers(new TestCallable(), getAllMembers());} catch (InterruptedException | ExecutionException | AnotherAvailableMemberNotFoundException e) {e.printStackTrace();}}/*** Gets cluster members** @return Set<Member> Set of Cluster Members**/private Set<Member> getAllMembers() {Set<Member> members = getHazelcastLocalInstance().getCluster().getMembers();return members;}/*** Gets an another member of cluster** @return Member Another Member of Cluster* @throws AnotherAvailableMemberNotFoundException An Another Available Member can not found exception*/private Member getAnotherMember() throws AnotherAvailableMemberNotFoundException {Set<Member> members = getAllMembers();for(Member member : members) {if(!member.localMember()) {return member;}}throw new AnotherAvailableMemberNotFoundException('No Other Available Member on the cluster. Please be aware that all members are active on the cluster');}/*** Gets Hazelcast local instance** @return HazelcastInstance Hazelcast local instance*/@SuppressWarnings('static-access')private HazelcastInstance getHazelcastLocalInstance() {HazelcastInstance instance = getHazelcast().getHazelcastInstanceByName(getHazelcastInstanceName());return instance;}public String getHazelcastInstanceName() {return hazelcastInstanceName;}public void setHazelcastInstanceName(String hazelcastInstanceName) {this.hazelcastInstanceName = hazelcastInstanceName;}public Hazelcast getHazelcast() {return hazelcast;}public void setHazelcast(Hazelcast hazelcast) {this.hazelcast = hazelcast;}public IDistributedExecutorService getDistributedExecutorService() {return distributedExecutorService;}public void setDistributedExecutorService(IDistributedExecutorService distributedExecutorService) {this.distributedExecutorService = distributedExecutorService;}public ICacheService getCacheService() {return cacheService;}public void setCacheService(ICacheService cacheService) {this.cacheService = cacheService;}}

步骤12：创建hazelcast-config.properties文件

hazelcast-config.properties文件显示集群成员的属性。

第一会员属性：

hz.instance.name = OTVInstance1hz.group.name = dev
hz.group.password = devhz.management.center.enabled = true
hz.management.center.url = http://localhost:8080/mancenterhz.network.port = 5701
hz.network.port.auto.increment = falsehz.tcp.ip.enabled = truehz.members = 192.168.1.32hz.executor.service.core.pool.size = 2
hz.executor.service.max.pool.size = 30
hz.executor.service.keep.alive.seconds = 30hz.map.backup.count=2
hz.map.max.size=0
hz.map.eviction.percentage=30
hz.map.read.backup.data=true
hz.map.cache.value=true
hz.map.eviction.policy=NONE
hz.map.merge.policy=hz.ADD_NEW_ENTRY

第二成员属性：

hz.instance.name = OTVInstance2hz.group.name = dev
hz.group.password = devhz.management.center.enabled = true
hz.management.center.url = http://localhost:8080/mancenterhz.network.port = 5702
hz.network.port.auto.increment = falsehz.tcp.ip.enabled = truehz.members = 192.168.1.32hz.executor.service.core.pool.size = 2
hz.executor.service.max.pool.size = 30
hz.executor.service.keep.alive.seconds = 30hz.map.backup.count=2
hz.map.max.size=0
hz.map.eviction.percentage=30
hz.map.read.backup.data=true
hz.map.cache.value=true
hz.map.eviction.policy=NONE
hz.map.merge.policy=hz.ADD_NEW_ENTRY

步骤13：创建applicationContext-hazelcast.xml

创建Spring Hazelcast配置文件applicationContext-hazelcast.xml ，并配置Hazelcast分布式执行器服务和Hazelcast实例。

<beans xmlns='http://www.springframework.org/schema/beans'xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'xmlns:hz='http://www.hazelcast.com/schema/spring'xsi:schemaLocation='http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsdhttp://www.hazelcast.com/schema/springhttp://www.hazelcast.com/schema/spring/hazelcast-spring-2.4.xsd'><hz:map id='customerMap' name='customerMap' instance-ref='instance'/><!-- Hazelcast Distributed Executor Service definition --><hz:executorService id='hazelcastDistributedExecutorService' instance-ref='instance' name='hazelcastDistributedExecutorService' /><!-- Hazelcast Instance configuration --><hz:hazelcast id='instance'><hz:config><!-- Hazelcast Instance Name --><hz:instance-name>${hz.instance.name}</hz:instance-name><!-- Hazelcast Group Name and Password --><hz:group name='${hz.group.name}' password='${hz.group.password}'/> <!-- Hazelcast Management Center URL --><hz:management-center  enabled='${hz.management.center.enabled}' url='${hz.management.center.url}'/><!-- Hazelcast Tcp based network configuration --><hz:network port='${hz.network.port}' port-auto-increment='${hz.network.port.auto.increment}'><hz:join><hz:tcp-ip enabled='${hz.tcp.ip.enabled}'><hz:members>${hz.members}</hz:members></hz:tcp-ip></hz:join></hz:network><!-- Hazelcast Distributed Executor Service configuration --><hz:executor-service name='executorService'core-pool-size='${hz.executor.service.core.pool.size}'max-pool-size='${hz.executor.service.max.pool.size}'keep-alive-seconds='${hz.executor.service.keep.alive.seconds}'/><!-- Hazelcast Distributed Map configuration --><hz:map name='map'backup-count='${hz.map.backup.count}'max-size='${hz.map.max.size}'eviction-percentage='${hz.map.eviction.percentage}'read-backup-data='${hz.map.read.backup.data}'cache-value='${hz.map.cache.value}'eviction-policy='${hz.map.eviction.policy}'merge-policy='${hz.map.merge.policy}'  /></hz:config></hz:hazelcast>	</beans>

步骤14：创建applicationContext.xml

Spring配置文件applicationContext.xml已创建。

<beans xmlns='http://www.springframework.org/schema/beans'xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'xmlns:hz='http://www.hazelcast.com/schema/spring'xsi:schemaLocation='http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsd'><import resource='classpath:applicationContext-hazelcast.xml' /><!-- Beans Declaration --><bean id='propertyConfigurer' class='org.springframework.beans.factory.config.PropertyPlaceholderConfigurer'><property name='locations'><list><value>classpath:/hazelcast-config.properties</value></list></property></bean><bean id='cacheService' class='com.onlinetechvision.cache.srv.CacheService'><constructor-arg ref='customerMap'/></bean><bean id='distributedExecutorService' class='com.onlinetechvision.executor.srv.DistributedExecutorService'><property name='hazelcastDistributedExecutorService' ref='hazelcastDistributedExecutorService' /></bean><bean id='hazelcast' class='com.hazelcast.core.Hazelcast'/><bean id='starter' class='com.onlinetechvision.exe.Starter'><property name='hazelcastInstanceName' value='${hz.instance.name}' /><property name='hazelcast' ref='hazelcast' /><property name='distributedExecutorService' ref='distributedExecutorService' /><property name='cacheService' ref='cacheService' /></bean>
</beans>

步骤15：创建应用程序类

创建应用程序类以运行应用程序。

package com.onlinetechvision.exe;import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;/*** Application class starts the application** @author onlinetechvision.com* @since 27 Nov 2012* @version 1.0.0**/
public class Application {/*** Starts the application** @param  String[] args**/public static void main(String[] args) {ApplicationContext context = new ClassPathXmlApplicationContext('applicationContext.xml');Starter starter = (Starter) context.getBean('starter');starter.start();}}

步骤16：建立专案

生成OTV_Spring_Hazelcast_DistributedExecution项目后，将创建OTV_Spring_Hazelcast_DistributedExecution-0.0.1-SNAPSHOT.jar 。

重要说明：集群的成员具有不同的Coherence配置，因此应为每个成员分别构建项目。

步骤17：与HAZELCAST管理中心集成

使用Hazelcast Management Center，可以监视和管理群集中的节点。

customerMap拥有的实体和备份计数可以通过“映射内存数据表”查看。我们通过customerMap分发了4个条目，如下所示：

可以通过地图浏览器查看示例键和值：

添加了第一个条目：

添加了第三项：

hazelcastDistributedExecutor服务详细信息可以通过“ Executors”选项卡查看。我们已经对第一个成员执行了3个任务，对第二个成员执行了2个任务，如下所示：

步骤18：通过启动集群成员来运行项目

在集群的成员上运行OTV_Spring_Hazelcast_DistributedExecution-0.0.1-SNAPSHOT.jar文件之后，将显示以下控制台输出日志：

第一成员控制台输出：

Kas 25, 2012 4:07:20 PM com.hazelcast.impl.AddressPicker
INFO: Interfaces is disabled, trying to pick one address from TCP-IP config addresses: [x.y.z.t]
Kas 25, 2012 4:07:20 PM com.hazelcast.impl.AddressPicker
INFO: Prefer IPv4 stack is true.
Kas 25, 2012 4:07:20 PM com.hazelcast.impl.AddressPicker
INFO: Picked Address[x.y.z.t]:5701, using socket ServerSocket[addr=/0:0:0:0:0:0:0:0,localport=5701], bind any local is true
Kas 25, 2012 4:07:21 PM com.hazelcast.system
INFO: [x.y.z.t]:5701 [dev] Hazelcast Community Edition 2.4 (20121017) starting at Address[x.y.z.t]:5701
Kas 25, 2012 4:07:21 PM com.hazelcast.system
INFO: [x.y.z.t]:5701 [dev] Copyright (C) 2008-2012 Hazelcast.com
Kas 25, 2012 4:07:21 PM com.hazelcast.impl.LifecycleServiceImpl
INFO: [x.y.z.t]:5701 [dev] Address[x.y.z.t]:5701 is STARTING
Kas 25, 2012 4:07:24 PM com.hazelcast.impl.TcpIpJoiner
INFO: [x.y.z.t]:5701 [dev]
--A new cluster is created and First Member joins the cluster.
Members [1] {Member [x.y.z.t]:5701 this
}Kas 25, 2012 4:07:24 PM com.hazelcast.impl.MulticastJoiner
INFO: [x.y.z.t]:5701 [dev] Members [1] {Member [x.y.z.t]:5701 this
}...
-- First member adds two new entries to the cache...
EntryAdded... Member : Member [x.y.z.t]:5701 this, Key : 1, OldValue : null, NewValue : Customer [id=1, name=Jodie, surname=Foster]
EntryAdded... Member : Member [x.y.z.t]:5701 this, Key : 2, OldValue : null, NewValue : Customer [id=2, name=Kate, surname=Winslet]...
--Second Member joins the cluster.
Members [2] {Member [x.y.z.t]:5701 thisMember [x.y.z.t]:5702
}...
-- Second member adds two new entries to the cache...
EntryAdded... Member : Member [x.y.z.t]:5702, Key : 4, OldValue : null, NewValue : Customer [id=4, name=Colin, surname=Farrell]
EntryAdded... Member : Member [x.y.z.t]:5702, Key : 3, OldValue : null, NewValue : Customer [id=3, name=Bruce, surname=Willis]

第二个成员控制台输出：

Kas 25, 2012 4:07:48 PM com.hazelcast.impl.AddressPicker
INFO: Interfaces is disabled, trying to pick one address from TCP-IP config addresses: [x.y.z.t]
Kas 25, 2012 4:07:48 PM com.hazelcast.impl.AddressPicker
INFO: Prefer IPv4 stack is true.
Kas 25, 2012 4:07:48 PM com.hazelcast.impl.AddressPicker
INFO: Picked Address[x.y.z.t]:5702, using socket ServerSocket[addr=/0:0:0:0:0:0:0:0,localport=5702], bind any local is true
Kas 25, 2012 4:07:49 PM com.hazelcast.system
INFO: [x.y.z.t]:5702 [dev] Hazelcast Community Edition 2.4 (20121017) starting at Address[x.y.z.t]:5702
Kas 25, 2012 4:07:49 PM com.hazelcast.system
INFO: [x.y.z.t]:5702 [dev] Copyright (C) 2008-2012 Hazelcast.com
Kas 25, 2012 4:07:49 PM com.hazelcast.impl.LifecycleServiceImpl
INFO: [x.y.z.t]:5702 [dev] Address[x.y.z.t]:5702 is STARTING
Kas 25, 2012 4:07:49 PM com.hazelcast.impl.Node
INFO: [x.y.z.t]:5702 [dev] ** setting master address to Address[x.y.z.t]:5701
Kas 25, 2012 4:07:49 PM com.hazelcast.impl.MulticastJoiner
INFO: [x.y.z.t]:5702 [dev] Connecting to master node: Address[x.y.z.t]:5701
Kas 25, 2012 4:07:49 PM com.hazelcast.nio.ConnectionManager
INFO: [x.y.z.t]:5702 [dev] 55715 accepted socket connection from /x.y.z.t:5701
Kas 25, 2012 4:07:55 PM com.hazelcast.cluster.ClusterManager
INFO: [x.y.z.t]:5702 [dev]
--Second Member joins the cluster.
Members [2] {Member [x.y.z.t]:5701Member [x.y.z.t]:5702 this
}Kas 25, 2012 4:07:56 PM com.hazelcast.impl.LifecycleServiceImpl
INFO: [x.y.z.t]:5702 [dev] Address[x.y.z.t]:5702 is STARTED
-- Second member adds two new entries to the cache...
EntryAdded... Member : Member [x.y.z.t]:5702 this, Key : 3, OldValue : null, NewValue : Customer [id=3, name=Bruce, surname=Willis]
EntryAdded... Member : Member [x.y.z.t]:5702 this, Key : 4, OldValue : null, NewValue : Customer [id=4, name=Colin, surname=Farrell]25.11.2012 16:07:56 DEBUG (DistributedExecutorService.java:42) - Method executeOnStatedMember is called...
25.11.2012 16:07:56 DEBUG (DistributedExecutorService.java:46) - Result of method executeOnStatedMember is : First Member' s TestCallable Task is called...25.11.2012 16:07:56 DEBUG (DistributedExecutorService.java:61) - Method executeOnTheMemberOwningTheKey is called...
25.11.2012 16:07:56 DEBUG (DistributedExecutorService.java:65) - Result of method executeOnTheMemberOwningTheKey is : First Member' s TestCallable Task is called...25.11.2012 16:07:56 DEBUG (DistributedExecutorService.java:78) - Method executeOnAnyMember is called...
25.11.2012 16:07:57 DEBUG (DistributedExecutorService.java:82) - Result of method executeOnAnyMember is : Second Member' s TestCallable Task is called...25.11.2012 16:07:57 DEBUG (DistributedExecutorService.java:96) - Method executeOnMembers is called...
25.11.2012 16:07:57 DEBUG (DistributedExecutorService.java:101) - Result of method executeOnMembers is : [First Member' s TestCallable Task is called..., Second Member' s TestCallable Task is called...]