Tibco RV - fault tolerance

時間 2019-12-06

標籤 tibco fault tolerance 简体版

原文原文鏈接

有不少系統不允許(在某個特定時段)服務中斷，這類的系統不僅會有備援，一般還會有 fault tolerance 機制，當系統掛掉時，備援的系統會自動啟動服務。Tibco RV 也有提供這樣的機制，底下將做說明。html

這裡舉的例子會有兩支程式，一支命名為 NumberGenerator，很簡單的每秒送出一個累加的數字，另外一支命名為 NumberReceiver，會由 RV 接收 NumberGenerator 送出的數字，並輸出到 console。假設接收數字的這個系統是我們負責的系統，是服務不能中斷的，我們以 RV 提供的 fault tolerance 機制，讓 NumberReceiver 萬一掛掉時，能夠當即啟動另外一個備援的系統，因此，測試時我們會同時啟動兩個 NumberReceiver 先啟動的會成為 active 的服務，後啟動的為備援，先來看一下程式。 java

 1 package idv.steven.rv.ft;
 2 
 3 import java.io.UnsupportedEncodingException;
 4 
 5 import com.tibco.tibrv.Tibrv;
 6 import com.tibco.tibrv.TibrvException;
 7 import com.tibco.tibrv.TibrvMsg;
 8 import com.tibco.tibrv.TibrvRvdTransport;
 9 import com.tibco.tibrv.TibrvTransport;
10 
11 public class NumberGenerator {
12     private String service = "7500";
13     private String network = ";225.1.1.1";
14     private String daemon = "tcp:7500";
15     private String subject = "DEMO.FT.NUM";
16     
17     public void run() {
18         try {
19             Tibrv.open(Tibrv.IMPL_NATIVE);
20             TibrvMsg.setStringEncoding("Big5");
21             TibrvTransport transport = new TibrvRvdTransport(service, network, daemon);
22             
23             for(int i=1; i<=1000; i++) {
24                 TibrvMsg msg = new TibrvMsg();
25                 msg.setSendSubject(subject);
26                 msg.update("number", i);
27                 
28                 transport.send(msg);
29                 
30                 try {
31                     Thread.sleep(1000);
32                 } catch (InterruptedException e) {
33                 }
34             }
35             
36             Tibrv.close();
37         } catch (TibrvException | UnsupportedEncodingException e) {
38             e.printStackTrace();
39         }
40     }
41 
42     public static void main(String[] args) {
43         NumberGenerator gen = new NumberGenerator();
44         gen.run();
45         System.out.println("stop");
46     }
47 }

上面的 NumberGenerator.java 很簡單的每秒送出一個累加的數字，由 1 累加到 1000 為止。對於上面程式有不瞭解的，能夠先參考「Tibco RV request/reply 的同步與非同步」，之後再回頭來看 fault tolerance。接下來要進入主題，看一下 fault tolerance 程式。tcp

  1 package idv.steven.rv.ft;
  2 
  3 import com.tibco.tibrv.Tibrv;
  4 import com.tibco.tibrv.TibrvException;
  5 import com.tibco.tibrv.TibrvFtMember;
  6 import com.tibco.tibrv.TibrvFtMemberCallback;
  7 import com.tibco.tibrv.TibrvListener;
  8 import com.tibco.tibrv.TibrvMsg;
  9 import com.tibco.tibrv.TibrvMsgCallback;
 10 import com.tibco.tibrv.TibrvRvdTransport;
 11 
 12 public class NumberReceiver implements TibrvMsgCallback, TibrvFtMemberCallback, Runnable {
 13     private String service = "7500";
 14     private String network = ";225.1.1.1";
 15     private String daemon = "tcp:7500";
 16     private String subject = "DEMO.FT.NUM";
 17     
 18     private String ftservice = "7504";
 19     private String ftnetwork = ";225.1.10.1";
 20     private String ftdaemon = "tcp:7504";
 21     
 22     private String ftgroupName = "DEMO.FT.GROUP";
 23     private int     ftweight = 50;
 24     private int     activeGoalNum = 1;
 25     private double  hbInterval = 1.5;
 26     private double  prepareInterval = 3; 
 27     private double  activateInterval = 4.8;
 28     
 29     private TibrvRvdTransport transport = null;
 30     private TibrvListener listener = null;
 31     
 32     private boolean active = false;
 33     
 34     @Override
 35     public void run() {
 36         try {
 37             Tibrv.open(Tibrv.IMPL_NATIVE);
 38             transport = new TibrvRvdTransport(service, network, daemon);
 39             TibrvRvdTransport fttransport = new TibrvRvdTransport(ftservice, ftnetwork, ftdaemon);
 40             fttransport.setDescription("fault tolerance");
 41             
 42             new TibrvFtMember(Tibrv.defaultQueue(), // TibrvQueue 
 43                       this,                 // TibrvFtMemberCallback
 44                       fttransport,          // TibrvTransport    
 45                       ftgroupName,          // groupName
 46                       ftweight,             // weight
 47                       activeGoalNum,        // activeGoal
 48                       hbInterval,           // heartbeatInterval
 49                       prepareInterval,      // preparationInterval,
 50                                             // Zero is a special value,
 51                                             // indicating that the member does 
 52                                             // not need advance warning to activate 
 53                       activateInterval,     // activationInterval
 54                       null);                // closure
 55             
 56             
 57             while(true) {
 58                 try {
 59                       Tibrv.defaultQueue().dispatch();
 60                 }
 61                 catch (TibrvException e) {
 62                    System.err.println("Exception dispatching default queue:");
 63                    System.exit(0);
 64                 }
 65                 catch(InterruptedException ie) {
 66                    System.exit(0);
 67                 }
 68             }
 69         } catch (TibrvException e) {
 70             e.printStackTrace();
 71         }
 72     }
 73     
 74     void enableListener() {
 75         try {
 76           // Subscribe to subject        
 77             listener = new TibrvListener(Tibrv.defaultQueue(),
 78                                          this, 
 79                                          transport, 
 80                                          subject, 
 81                                          null);
 82             System.out.println("Start Listening on: " + subject);    
 83         }
 84         catch (TibrvException e) {
 85            System.err.println("Failed to create subject listener:");
 86            System.exit(0);
 87         }
 88     }
 89     
 90     void disableListener() {
 91        listener.destroy();
 92        System.out.println("Destroy Listener on Subject: " + subject);    
 93     }
 94 
 95     @Override
 96     public void onFtAction(TibrvFtMember member, String ftgroupName, int action) {
 97         if (action == TibrvFtMember.PREPARE_TO_ACTIVATE) {
 98             System.out.println("TibrvFtMember.PREPARE_TO_ACTIVATE invoked...");
 99             System.out.println("*** PREPARE TO ACTIVATE: " + ftgroupName);
100         }
101         else if (action == TibrvFtMember.ACTIVATE) {
102             System.out.println("TibrvFtMember.ACTIVATE invoked...");
103             System.out.println("*** ACTIVATE: " + ftgroupName);
104             enableListener();
105             active = true;
106         }
107         else if (action == TibrvFtMember.DEACTIVATE) {
108             System.out.println("TibrvFtMember.DEACTIVATE invoked...");
109             System.out.println("*** DEACTIVATE: " + ftgroupName);
110             disableListener();
111             active = false;
112         }
113     }
114 
115     @Override
116     public void onMsg(TibrvListener listener, TibrvMsg msg) {
117         if (subject.equals(listener.getSubject())) {
118             try {
119                 int num = msg.getAsInt("number", 0);
120                 System.out.println("number: " + num);
121             } catch (TibrvException e) {
122                 e.printStackTrace();
123             }
124         }
125     }
126 
127     public static void main(String[] args) throws InterruptedException {
128         NumberReceiver rcv = new NumberReceiver();
129         Thread tRcv = new Thread(rcv);
130         tRcv.start();
131         tRcv.join();
132         System.out.println("stop");
133     }
134 }

上面的程式與 fault tolerance 有關的是第 39~54 行及第 96~113 行，在解釋程式以前，我先對 RV 提供的 fault tolerance 做簡單的說明。ide

group name

RV 的 fault tolerance 以 group name 來區分，也就是同一個 group name 的程式會成為一個羣組，相互備援。this

active & inactive

RV 將 fault tolerance 的系統狀態分紅三個階段，以下:spa

1. DEACTIVATE: 顧名思義，就是非屬於提供服務的狀態，這時候系統當然就別作什麼事，只要一直傾聽 RV 的訊息，當傾聽到要切換狀態時再進行相關工做。
2. PREPARE_TO_ACTIVATE: 當 RV 送來這個狀態，表示本來正提供服務的系統可能出了狀況，備援的系統要開始準備啟動，一般收到這個 event 時，會進行一些系統資源初始化的工做，以便當程式真的要啟動提供服務時，能夠儘快啟動，以使服務中斷的時間儘可能的縮短。
3. ACTIVATE: 收到這個 event，就真的要接手服務了!

fault tolerance callback function

上述三個狀態的切換，RV 會透過 onFtAction 這個 callback function 通知備援的程式，因此程式要實做 TibrvFtMemberCallback 這個介面。code

active goal

在我們將測試的例子裡，只會啟動兩個 reciver 程式，一次只會有一個程式提供服務 (接收 generator 傳來的數字並輸出到 console)，可是，實務上要有幾支程式處於 ACTIVATE 狀態是能夠設定的，這個數字稱為 active goal。htm

weight

每支程式創建 fault tolerance 時，在建構子的參數裡會有個 weight 的參數，這個值由 1 到整數的最大值，數字越大優先權越大，因此，若是有三支程式 A、B、C，其權重 (weight) 分別為 50、30、100，當 RV 要從這三支程式裡挑一支啟動來提供服務時，就會挑權重最大的 C。blog

heart beats

每個處於 ACTIVATE 狀態的程式，會不停的送出 heart beats (心跳) 訊息給其它同羣組的程式，以表示它還活著，還在正常提供服務。ip

現在說明 NumberReceiver.java 程式，以下:

Line 12: 全部參與 fault tolerance 的程式都要實做 TibrvFtMemberCallback。
Line 39: 創建一個 fault tolerance 的通道。
Line 42: 使這支程式成為羣組中的一個成員 (member)，這裡面有好幾個和時間有關的參數 - hbInterval、prepareInterval、activeInterval，單位是秒，指的是 heart beat 的時間、進入 PREPARE_TO_ACTIVATE 的時間及切換到 ACTIVATE 的時間，在這裡設定為 1.五、三、4.8，表示每三秒會有一次的心跳訊息，當 3 秒鐘都沒有收到訊息即進入 PREPARE_TO_ACTIVATE 狀態，當 4.8 秒沒有收到心跳訊息，就切換到 ACTIVATE 狀態。
Line 74~90: 這兩個 method (enableListener、disableListener) 是用來啟動服務及中止服務。
Line 96~113: onFtAction 是 fault tolerance 的 callback function，這裡能夠接收到狀態改變的 event。
Line 116~125: 收到 NumberGenerator 送來的訊息，將它輸出到 console。

現在進行測試 …

首先啟動 NumberGenerator …
開啟兩個「命令提示字元」的視窗，並同時啟動 NumberReceiver，能夠看到其中一個開始提供服務，以下圖:

在左邊本來正在提供服務的程式按 Ctrl-C 以中斷服務，能夠看到右邊的程式接手了服務。

再次啟動左邊的程式，左邊的程式將處於 DEACTIVATE 的狀態。

在右邊視窗中按 Ctrl-C 中斷右邊的程式，能夠看到左邊的程式又接手服務。

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。