這是我的Hazelcast系列的第四部分。 如果一個人沒有看到其他三個人，我建議一個人去看第1 部分 ， 第2 部分和第3部分 。

記錄中

日志記錄是任何應用程序的重要功能，我的示例也是如此。 System.out.println可以用作告訴用戶控制臺應用程序中正在發生什么的有用工具。 但是，讓我們面對現實吧，如果您正在閱讀如何為分布式應用程序使用工具，那么該人確實不是初學者。 看到一系列日志消息不應嚇到任何人。 實際上，對于本文中的示例，他們有必要了解誰在做什么。 畢竟，我們將討論線程編程。

Hazelcast的好伙伴似乎已經同意日志記錄很重要，因此有許多不同的方法來定義要記錄的庫。 日志記錄框架僅取決于JDK日志記錄，并且具有許多適配器，甚至允許自定義日志記錄框架。 選擇的一個日志適配器由屬性hazelcast.logging.type設置為以下設置：

• JDK日志記錄，這是默認設置。
• log4j
• slf4j
• 沒有

我使用了Log4j2，所以我選擇了slf4j并放入了使其工作所需的四個 jar文件。

旋轉分布式線程

與Hazelcast中的許多類一樣，IExecutorService實現了Java庫ExecutorService的接口。 該接口定義什么是線程池。 該接口是java.util.concurrent包的一部分，自Java 1.5開始就存在。 該程序包還具有其實現，可以從java.util.concurrent.Executors中進行訪問。 我希望我在Java 1.4或1.3或1.2或1.1中有類似的東西。 在死鎖發生之前，使線程池變得很有趣。 現在我可以使用Java庫的池了，對我來說已經足夠了。

ExecutorService具有有趣的“功能”。 必須關閉它們，否則服務不會消失。 第一次使用它們時，導致內存泄漏并關閉了JVM。 我在自己的測試中發現了該錯誤，因此客戶不必再看到我的學習經驗。 IExecutorService的皺紋有所不同。 在所有線程完成之前，該服務不會消失。 這導致許多不正常的關機。 你被警告了！

IExecutorServices可以幾種不同方式共享線程。 這里是它們的詳細信息：

任何'Ole實例

這是一個只調用submit(Callable call). 這不僅僅只是將線程隨機設置到集群中。 它使用該線程進行一些負載平衡，因此實例不會被線程破壞。

致特定成員

這是通過submit(Callable call, Member member)方法完成的。 這會將線程發送到群集的特定成員。 這里沒有負載均衡； 只是發送給會員。 小心，一個人很容易使一個構件超載，并在進行任何處理時確實踩剎車。 我可以將其視為創建自定義負載均衡器的一種方式。

致會員集合

是的，可以將一個線程發送給多個成員。 當我進行示例編碼時，所有成員的行為就像他們擁有自己的線程而不共享一個線程。 如果將Callable <T>作為其線程實現實現，則該方法使用成員作為鍵返回“未來地圖”。 如果使用Runnable，則不返回任何內容。

用正確的鑰匙致會員

IMap的條目可以在群集中的任何位置。 如果需要對該條目進行處理，則本地線程將必須通過網絡上拉該條目。 如果條目很大，可能會出現問題。 更好的方法是將希望較小的線程轉移到條目上。 為此，群集需要知道將其發送到哪里。 因此，呼叫submit(Callable call, Object key) 。

致全體會員

這與提交成員集合的方式相同，但是就像集群中的每個成員一樣，都是成員。 如果一個集群中有大量成員，這可能會很“有趣”。 我想我在一個集群中聽到了多達1000名成員。 確保這是人們想要的，然后再調用它。

使用ExecutionCallback

基本上，這是一種發送線程并讓結果異步返回的方法。 如果提交了一個線程，則使用ExecutionCallback。 如果涉及多個成員，則使用MultiExecutionCallback。

范例程式碼

在開始之前，請允許我說我沒有IExecutorService中每個方法的示例。 但是，對于每種討論的類型，我都有一個示例。 關于示例代碼的另一件事。 出于指導目的，我在先前的文章中做了一些復制和粘貼編碼，因此每個示例都可以獨立存在，并且可以了解到什么地方。 我在第3部分中做了很多。 如果沒有注意到，請再次查看。

這次我沒有這樣做，因為將會復制很多代碼，并且結果會非常難看。 我使用了枚舉，我認為結果非常好。 由于示例數量有限，我認為枚舉是一個不錯的選擇，并且使我能夠以塊的形式顯示代碼，而如果首先顯示框架，則可以理解。

有了這個解釋，讓我們繼續前進！

構架

這是主要的位。 它由主類和線程類組成。 注意主類如何顯示調用線程可以提交的每種方式。

主要

package hazelcastservice;import com.hazelcast.core.Hazelcast; import com.hazelcast.core.HazelcastInstance; import com.hazelcast.core.IExecutorService; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.RejectedExecutionException; import org.slf4j.Logger; import org.slf4j.LoggerFactory;/**** @author Daryl*/ public class Main {private static final Logger logger = LoggerFactory.getLogger(Main.class);public static final String SERVICE_NAME = "spinnerella";public static final int NUM_INSTANCES = 5;/*** @param args the command line arguments*/public static void main(String[] args) {System.setProperty("hazelcast.logging.type", "slf4j");List<HazelcastInstance> instances = new ArrayList<>(NUM_INSTANCES);for(int i = 0; i < NUM_INSTANCES; i++) {instances.add(Hazelcast.newHazelcastInstance());logger.info("instance {} up", i);}IExecutorService spinner = instances.get(0).getExecutorService(SERVICE_NAME);try {HazelcastIExecutorServiceExamples.TO_SOME_MEMBER.example(instances, spinner);HazelcastIExecutorServiceExamples.TO_PARTICULAR_MEMBER.example(instances, spinner);HazelcastIExecutorServiceExamples.ON_THE_KEY_OWNER.example(instances, spinner);HazelcastIExecutorServiceExamples.ON_A_SET_OF_MEMBERS.example(instances, spinner);HazelcastIExecutorServiceExamples.ON_ALL_MEMBERS.example(instances, spinner);HazelcastIExecutorServiceExamples.CALLBACK.example(instances, spinner);HazelcastIExecutorServiceExamples.MULTIPLE_MEMBERS_WITH_CALLBACK.example(instances, spinner);//Lets setup a loop to make sure they are all done (Especially the callback ones)for(HazelcastIExecutorServiceExamples example: HazelcastIExecutorServiceExamples.values()) {while(!example.isDone()) {Thread.sleep(1000);}}} catch(ExecutionException ee) {logger.warn("Can't finish the job", ee);} catch(InterruptedException ie) {logger.warn("Everybody out of the pool", ie);} finally {// time to clean up my toysboolean allClear = false;while(!allClear) {try {Thread.sleep(1000);Hazelcast.shutdownAll();allClear = true;} catch(InterruptedException ie) {//got interrupted. try again} catch(RejectedExecutionException ree) {logger.debug("caught a RejectedExecutionException");allClear = false;}}logger.info("All done");}} }

線

package hazelcastservice;import java.io.Serializable; import java.util.Random; import java.util.concurrent.Callable; import org.slf4j.Logger; import org.slf4j.LoggerFactory;/*** This class was inspired by the song "I Like to Move it" from the movie * Madagascar by Dreamworks. I offer NO apologies for using it. * * To those software developers who like consistent results, I used java.util.Random to* make it loop inconsistently each time call is called. * * Sometimes you need to make your own entertainment.* @author Daryl*/ public class MoveItMoveIt implements Callable<Integer>, Serializable {private static final Logger logger = LoggerFactory.getLogger(MoveItMoveIt.class);private static final int UPPER_BOUND = 15;@Overridepublic Integer call() throws Exception {Random random = new Random();int howMany = random.nextInt(UPPER_BOUND); // int howMany = 2;for(int i = 0; i < howMany; i++) {logger.info("I like to Move it Move it!");}logger.info("Move it!");return howMany;} }

細節

在這里，我將展示所討論的不同類型的呼叫。 請記住，這些是Enum類的塊。 done是一個受保護的變量，并且需要實現public void example(List<HazelcastInstance> instances, IExecutorService spinner) 。

任何'Ole實例

TO_SOME_MEMBER() {@Overridepublic void example(List<HazelcastInstance> instances, IExecutorService spinner)throws ExecutionException, InterruptedException {logger.info("Submit to some member.");Future<Integer> howMany = spinner.submit(new MoveItMoveIt());logger.info("It moved it {} times", howMany.get());done = true;}}

致特定成員

TO_PARTICULAR_MEMBER {@Overridepublic void example(List<HazelcastInstance> instances, IExecutorService spinner)throws ExecutionException, InterruptedException {logger.info("Submit to a particular member.");Member member = getRandomMember(instances);logger.debug("member is {}", member);Future<Integer> howMany = spinner.submitToMember(new MoveItMoveIt(), member);logger.info("It moved it {} times.", howMany.get());done = true;}private Member getRandomMember(List<HazelcastInstance> instances) {Set<Member> members = instances.get(0).getCluster().getMembers();int i = 0;int max = new Random().nextInt(instances.size());Iterator<Member> iterator = members.iterator();Member member = iterator.next();while(iterator.hasNext() && (i < max)) {member = iterator.next();i++;}return member;}}

致會員集合

ON_A_SET_OF_MEMBERS {@Overridepublic void example(List<HazelcastInstance> instances, IExecutorService spinner)throws ExecutionException, InterruptedException {logger.info("Send to some of the members");Set<Member> randomMembers = getRandomMembers(instances);Map<Member, Future<Integer>> results = spinner.submitToMembers(new MoveItMoveIt(), randomMembers);for(Future<Integer> howMany: results.values()) {logger.info("It moved {} times", howMany.get());}done = true;}private Set<Member> getRandomMembers(List<HazelcastInstance> instances) {int max = new Random().nextInt(instances.size());Set<Member> newSet = new HashSet<>(instances.size());int k = 0;Iterator<Member> i = instances.get(0).getCluster().getMembers().iterator();while(i.hasNext() && k < max) {newSet.add(i.next());k++;}return newSet;}}

用正確的鑰匙致會員

ON_THE_KEY_OWNER {@Overridepublic void example(List<HazelcastInstance> instances, IExecutorService spinner)throws ExecutionException, InterruptedException {logger.info("Send to the one owning the key");HazelcastInstance randomInstance = getRandomInstance(instances);IMap<Long, Boolean> map = randomInstance.getMap("default");Long one = 1L;map.put(one, Boolean.TRUE);Future<Integer> howMany = spinner.submitToKeyOwner(new MoveItMoveIt(), one);logger.info("It moved it {} times.", howMany.get());done = true;}private HazelcastInstance getRandomInstance(List<HazelcastInstance> instances) {return instances.get(new Random().nextInt(instances.size()));}}

致全體會員

ON_ALL_MEMBERS {@Overridepublic void example(List<HazelcastInstance> instances, IExecutorService spinner)throws ExecutionException, InterruptedException {logger.info("Send to all members");Map<Member, Future<Integer>> results = spinner.submitToAllMembers(new MoveItMoveIt());for(Future<Integer> howMany: results.values()) {logger.info("It moved {} times", howMany.get());}done = true;}}

使用ExecutionCallback

此示例代碼包含兩段代碼，分別顯示一個回調和多個回調。

CALLBACK {@Overridepublic void example(List<HazelcastInstance> instances, IExecutorService spinner)throws ExecutionException, InterruptedException {logger.info("example with a callback");spinner.submit(new MoveItMoveIt(), new ExecutionCallback<Integer>() {@Overridepublic void onResponse(Integer response) {logger.info("It moved {} times", response);done = true;}@Overridepublic void onFailure(Throwable thrwbl) {logger.error("trouble in the callback", thrwbl);done = true;}});} },MULTIPLE_MEMBERS_WITH_CALLBACK {@Overridepublic void example(List<HazelcastInstance> instances, IExecutorService spinner)throws ExecutionException, InterruptedException {logger.info("running on multiple members with callback");spinner.submitToAllMembers(new MoveItMoveIt(), new MultiExecutionCallback() {@Overridepublic void onResponse(Member member, Object o) {logger.info("member finished with {} moves", o);}@Overridepublic void onComplete(Map<Member, Object> map) {logger.info("All members completed");for(Object value: map.values()) {logger.info("It moved {} times", value);}done = true;}});}

結論

再次在自己的博客上發布自己的代碼/想法非常好。 我快速瀏覽了Hazelcast的IExecutorService的功能。 我的示例代碼遵循DRY原理。 完整的代碼可以在這里找到。

參考文獻

與我的Hazelcast指南一樣，我的信息來自Hazelcast文檔，可在此處找到。

翻譯自: https://www.javacodegeeks.com/2014/10/beginners-guide-to-hazelcast-part-4.html

總結

以上是生活随笔為你收集整理的Hazelcast入门指南第4部分的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。