引言
隨著信息系統的結構的日益復雜和規模的不斷擴大，交易中間件在復雜系統的應用也越來越廣。交易中間件 作為一個中間層的系統，在接收客戶端請求時，通常需要做一些負載控制和用戶緩存的一些功能。對於軟件測試人員來說， 測試交易中間件時，避免不了模擬客戶端在高負載情況下的一些有規律或隨機的行為。這些測試有時是功能性驗證測試 (Functional Verification Test)，有時也涉及到性能測試 (Performance Test)。
本文將介紹如何使用 Java 語言 編寫多線程的自動化測試腳本，並且使用線程池模擬一些特殊的有規律的測試場景。
本文首先會簡單的介紹交易中 間件及 Java 多線程編程的概念。接著提出項目中遇到的問題。然後就碰到的問題，使用 Java 多線程技術模擬測試場景。 解決這個問題後，就類似的問題提出推廣的思路。
本示例的必備條件
Java 的多線程應用范圍很廣，交易中 間件的種類也有許多。本文 JDK 的版本是 JDK5，原因是 JDK5 中加入了比較豐富的多線程並發工具。目前，JDK 的最新版 本是 JDK7，其中又增加了許多工具包括 Phaser、ThreadLocalRandom、ForkJoinPool、以及 TransferQueue 等，但是如果 掌握了 JDK5 的多線程工具，對 JDK7 的工具也一定不會陌生了。
本文的交易中間件是以 IBM Information Management System（IMS）的 TM 為示例，這個示例是為您創建一個學習的場景，當然 Java 的多線程應用的范圍非常廣泛 ，不限於這一種交易中間件。在本文的推廣部分，也另外假設了一個場景，並加以實現。
如果您也需要以 IMS TM 來測試，請確保提供以下的測試環境。
需要在 Windows 上安裝的軟件：
JDK 1.5（或者更新的版本）
Rational Functional Tester V7.0（或者更新的版本）
需要在 IBM z/OS 上安裝的環境：
IMS Version 9 （或者更新的版本）
IMS Connect Version 9 （或者 更新的版本）
OTMA
TCP/IP
概念介紹
交易中間件
中間件的產品種類很多，根據中間件 在系統中所起的作用和采用的技術不同，大致劃分為五大類：數據庫中間件 (Database Middleware，DM)、遠程過程調用中 間件 (Remote Procedure Call, RPC)、基於對象請求代理 (Object Request Broker, ORB)、中間件與交易中間件 ( Transaction Processing Monitor, TPM, 也稱事務處理中間件 )。
交易中間件是一種復雜的中間件產品，通常是在 負載的環境下運用分布式應用的速度和可靠性來實現的。交易中間件向用戶提供一系列的服務，包括通信服務、日志服務、 系統服務和交易服務等。
交易中間件的通信主要是基於 TCP/IP 的 socket 技術和基於消息傳遞與排隊機制實現的 ，其通信的過程如圖 1 所示：
圖 1. 交易中間件通信過程
交易中間件端通常有偵聽方法在監聽客戶端 的連接請求，返回一個連接後並生成一個相應的客戶服務進程。在接收到客戶端的數據後，對數據進行分割、加密、封裝成 消息包。然後做分發、入隊、發送等操作。
Java 多線程編程
從交易中間件的概念中可以看出，如果要模擬 客戶端使用交易中間件，首先要模擬建立 Socket 連接。接下來，如果需要模擬多個用戶的連接，就需運用用 Java 的多線 程機制了。
Java 的多線程機制使應用程序能夠並發的執行，並且 Java 能夠運用同步機制使並發的線程能夠互斥的 使用共享資源，從而達到線程安全的作用。Java 的多線程的實現方法有兩種：1) 通過 Thread 繼承。為創建一個線程，最 簡單的方法就是從 Thread 類繼承。2) 通過 Runnable 接口。該方式主要是用來實現特殊功能，如復雜的多繼承功能等。 Runnable 接口的應用雖然提高了軟件開發的靈活度，但是同時也是造成 BUG 的根源之一，所以應根據不同的需求，合適的 選取兩種方法，兩種實現的方法的代碼可見代碼清單一。
清單 1.Java 多線程實現的兩種方式
// 方式一
public class MyThread extends Thread {
public void run() {
System.out.println("MyThread.run()");
}
}
// 在合適的地方啟動線程如下：
MyThread myThread1 = new MyThread();
MyThread myThread2 = new MyThread();
myThread1.start();
myThread2.start();
// 方式二
public class MyThread extends OtherClass implements Runnable {
public void run() {
System.out.println("MyThread.run()");
}
}
// 為了啟動 MyThread，需要首先實例化一個 Thread，並傳入自己的 MyThread 實例
MyThread myThread = new MyThread();
Thread thread = new Thread(myThread);
thread.start();
在 Java 中，要實現多線程流程控制方法有以下幾個：
通過 sleep() 來實現 Java 多 線程流程控制，可以用於多線程流程控制演示。
通過 interrupt() 函數對 Java 多線程進行流程控制。這種控制方法比 sleep() 函數要精確得多，但在用法上與 sleep() 函數不太相同。
通過 wait() 和 notify() 來進行流程控制。wait() 方式與 sleep() 方式有相同之處，都是從線程內部對線程進行中 斷的。不同的是，它沒有提供中斷時間，其喚醒方式就是 notify() 和 notifyA ll()。
通過 join() 來進行流程控制，join()方式結合了 interrupt 和 sleep 方式的優點。在實際設計當中應注意使用 join 方式，因為不恰當的使用可能會打亂線程的流程。
項目面臨的問題
項目背景
IBM 的 IMS 和 CICS （1968）是在 IBM 大型機 mainframe 時代最早的交易中間件，它們都采用請求隊列管理、檢查點機制和批處理的啟動機制 。本文所要測試的交易中間件正是 IBM 公司的大型數據庫產品 IMS 的交易中間件。它提供了交易處理的通信、並發訪問控 制、事務控制、資源管理和必要的事務監聽功能。IMS 的控制及數據流程可見圖 2。
圖 2. IMS 控制及數據流程
項目需 求
在 IMS TM 中，負責 TCP/IP 通信的模塊是 IMS Connect，這也是本文測試的核心連接模塊。該模塊的新增功能 包括 TCPIPQ 和 Health Report 功能。TCPIPQ 是一個 socket 連接的緩沖隊列，那些暫時沒有被接受的連接請求會進入這 個隊列，它的大小是可控的。Health Report 功能是定時的檢查目前的 socket 連接是否達到最大連接數，並且將已連接的 比例匯報給相應的控制模塊，客戶也可以通過命令查看當前已建立 socket 的占有率。
面臨的問題
在測試這兩個新增功能時，由於 IMS Connect 能夠建立的最大 Socket 連接數可以達到 65535 個，而最少的數目也有 100。這麼多的 socket 數目，如果一個一個的開啟客戶端程序來發起交易是不現實的。
在測試 Health Report 時，當需要測試占有率達到百分之百，即模擬當前的 socket 連接已經達到上限的場景，需要 同時建立並發的請求，並且數目要足夠多，達到連接上限。
在測試 TCPIPQ 時，只有讓現有的 socket 連接達到最大連接數，新的連接才能進入 TCPIPQ 隊列。如果需要測試隊列 的出隊和入隊操作是否正確，需要多個線程有規律的簡歷 socket 連接。
解決方法及實現細節
為了解決項目面臨的測試難題，本文采用 Java 的多線程機制來模擬這三個問題所涉及的 場景。由於在本項目中客戶端發出的請求需要復雜的處理，所以本文采用實現 Runnable 接口的方法來實現多線程，這樣發 送請求的類還可以繼承其它的工具類來實現復雜的處理。
由於 IMS Connect 所支持的 socket 的數目 (MAXSOC) 可 以從 100 到 65535，而 TCPIPQ 隊列的大小可以從 50 到 65535，本文在實現客戶端多線程請求前，設置 IMS Connect 的 MAXSOC=100 並且設置 TCPIPQ=50。
在這個前提下，本文實現的線程類在演示時，如果創建 100 個線程的客戶端連 接，就會達到 socket 連接的上限，使 IMS Connect 的 Health 為 0，意味著不能再監聽多余的 socket 請求，如果繼續 發送 50 個請求，這些請求會進入 TCPIPQ 的隊列，使這個緩存隊列也達到上限。再繼續發送請求就會被拒絕，並得到拒絕 的提示信息。
如果使用最簡單的實現方法，可以參考如下的代碼清單 2。
清單 2. 多客戶端並發連接的簡單 實現
class Utility { // 工具類
public byte[] getHeadData(String...parms){
.....
....
}
public byte[] getBodyData(String...parms){
.....
....
}
public byte[] getFootData(String...parms){
.....
....
}
}
class SocketClient extends Utility implements Runnable,Test{ // 實現 Runnable 接口
String hostName;
int port;
public SocketClient(String hostName,int port) {
this.hostName=hostName;
this.port=port;
}
// 線程啟動後的連接，並發送數據的
public void run() {
try {
InetAddress inetAddress = InetAddress.getByName(hostName);
Socket socket = new Socket(inetAddress,port);
OutputStream os = socket.getOutputStream();
BufferedOutputStream bos = new BufferedOutputStream(os);
byte[] sendHeadData = getHeadData("aaa","bbb","ccc");
byte[] sendBodyData = getBodyData("ddd","eee","fff");
byte[] sendFootData = getFootData("ggg","hhh","iii");
bos.write(sendHeadData);
bos.write(sendBodyData);
bos.write(sendFootData);
bos.flush();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
// 測試的主線程，演示測試的一些場景
public class SocketConnectionDemo {
public int connectionNumb; // 連接數
public String hostName;// 連接的 IMS Connect 的域名
public int port;// 連接的端口
public SocketConnectionDemo(int connectionNumb,String hostName, int port){
this.connectionNumb=connectionNumb;
this.hostName=hostName;
this.port=port;
}
public static void main(String[] args) {
int connectionNumb = 50;
String hostName = "ec32181.vmec.svl.com";
int port = 9999;
SocketConnectionDemo scd = new SocketConnectionDemo(connectionNumb,hostName,port);
}
// 最簡單的測試場景
public void testScenario1(){
for(int i=1; i<=connectionNumb; i++){
SocketClient sc = new SocketClient(hostName, port);
Thread clientT = new Thread(sc,"Client"+i);
clientT.start();
}
// 測試的具體邏輯
… ..
… .
}
}
從這個代碼清單可以看出，testScenario1 是具體的測試場景調用的方法。這個測試場景使用最簡單的方法實 現線程的啟動，並根據具體的 connectionNumb 建立相應的連接數，在這些連接都建立之後，可以執行具體的測試邏輯，測 試在峰值情況下的 IMS Connect 的表現，並查看具體的 Health Value。
這個實現也存在一些問題，這些多線程的 客戶端雖然是並行的執行，但是它們並不是在同一起跑線開始的，因為在主線程中，它們都是一個一個建立的線程，一旦建 立就 start 了。為了實現所有線程同時開始，快速使得 IMS Connect 達到峰值，就需要使用線程池技術，這就是 JDK1.5 中新增並發工具類。在此使用了 CountDownLatch 來實現連接數的控制，並且通過一個值為 1 的 CountDownLatch 對象來 實現同一起跑線開始的閘門功能。詳細代碼請查看代碼清單 3。
清單 3. 使用 CountDownLatch 實現並發連接
class ThreadTest implements Runnable{
private static CountDownLatch startCdl; // 用於啟動所有連接線程的閘門
private static CountDownLatch doneCdl;// 所有連接工作都結束的控制器
public ThreadTest(CountDownLatch startCdl,CountDownLatch doneCdl) {
this.startCdl=startCdl;
this.doneCdl=doneCdl;
}
public void run() {
try {
startCdl.await();
System.out.println(
Thread.currentThread().getName()+" has been working!!!!");
// 此處需要代碼清單一的那些連接操作
…… ..
……
… ..
doneCdl.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public class CountDownLatchDemo1 {
public static void main(String[] args) {
CountDownLatch startCdl = new CountDownLatch(1);// 啟動的閘門值為 1
CountDownLatch doneCdl = new CountDownLatch(100);// 連接的總數為 100
for(int i=1; i <=100; i ++){
ThreadTest tt = new ThreadTest(startCdl,doneCdl);
new Thread(tt,"Thread"+i).start();
}
// 記錄所有連接線程的開始時間
long start = System.nanoTime();
// 所有線程雖然都已建立，並 start。但只有等閘門打開才都開始運行。
startCdl.countDown();
try {
doneCdl.await();// 主線程等待所有連接結束
// 連接達到峰值後，執行一些測試邏輯代碼
......
......
......
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
// 記錄所有連接線程的結束時間
long end = System.nanoTime();
System.out.println("The task takes time(ms): "+(end-start)/100000);
}
}
應用及驗證
在上一節，介紹了整個多線程的實現細節。在本節中，將運行這個多線程的應用，並結合項 目的 IMS Connect 服務器，演示連接數是否能像預期一樣達到上限，進行峰值的測試。
IMS Connect 目前的版本是 V12，正在開發的版本是 V13。在 V13 中，可以通過一些命令來查看當前的 Health 值，即連接數與上限的比例情況。本文 中，設置 IMS Connect 的上限連接數為 100，開放端口為 5，那麼在 IMS Connect 開啟時，可以通過命令查看得到 Health 值為 95，這意味著目前連接數占有率為 95%。
接下來，開啟多線程的連接，並發的連接 95 個連接。IMS Connect 後台會顯示連接在不斷的增長，當達到峰值時會出現相應的提示信息。
此時，再查看 Health 的值，會發 現已經變成了 000，這意味著 IMS Connect 已經達到了峰值。
推廣
本文為了實現多個連接線程同時開啟並 發拼搶連接交易中間件服務器的過程，使用了 JDK1.5 的並發類 CountDownLatch。但是，JDK1.5 以後的版本引入了許多高 級的並發特性，充分利用了現代多處理器和多核心系統的功能以編寫大規模並發應用程序。主要包含原子量、並發集合、同 步器、可重入鎖，並對線程池的構造提供了強力的支持。
除了本文使用的 CountDownLatch，還有許多同步器。例如 ，Semaphore、Barrier、Exchanger、Future 等。為了實現項目後期的連接緩存隊列 TCPIPQ 的測試，Exchanger 可能會被 使用。Exchanger 可以實現兩組線程互相交換一些共享資源的功能。
為了說明 Exchanger 的使用，本文假設一種場 景。假設有一個連接緩存隊列，有一個連接器負責創建連接，創建的連接會存儲在隊列裡。另外一個釋放連接器可以釋放連 接，從隊列裡移除連接。連接器每次會隨機地創建 1 或 2 個連接。釋放連接器只能每次釋放 1 個連接。最後肯定會有連 接隊列滿的時候，那時就可以進行連接隊列的測試了。
要實現這一種場景，可以使用 Exchanger 來實現。可以查看 代碼清單 4。
清單 4. 使用 Exchanger 實現連接緩存隊列的測試
// 假設建立連接的類是 Connection
class Connection{
private String connName;
private String ipAddress;
public Connection(String connName,String ipAddString) {
this.connName=connName;
this.ipAddress=ipAddString;
}
.....
...
}
public class ConnectionQueueDemo {
// 使用交換器實現連接器與釋放連接器之間資源的共享
private static Exchanger<LinkedList<Connection>> exconn =
new Exchanger<LinkedList<Connection>>();
// 連接器
public class Connector implements Runnable{
private LinkedList<Connection> connQueue;
private String ipAddress;
public Connector(LinkedList<Connection> connQueue,String ipAddress) {
this.connQueue=connQueue;
this.ipAddress=ipAddress;
}
public void run() {
boolean flag=true;
while(flag){
// 每次連接隨機的 1~2 個連接。
Random random = new Random();
int connNumb = (random.nextInt())%2 + 1; // 得到隨機的 1~2 個連接數
if(connNumb > 1){
System.out.println("Connector creates 2 connection!");
}else{
System.out.println("Connector creates 1 connection!");
}
for(int i=0; i<connNumb; i++){
Connection conn = new Connection("Connector", getIpAddress());
connQueue.add(conn);
}
// 休息 1 秒
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
try {
// 交換給釋放連接器，讓釋放連接器工作！
connQueue =(LinkedList<Connection>) exconn.exchange(connQueue);
} catch (InterruptedException e) {
e.printStackTrace();
}
if(connQueue.size()==10){
System.out.println("The connection queue is full!! The programme is end!");
// 當隊列滿時，可以加入一些測試邏輯代碼
....
...
flag=false;
System.exit(0);
}else{
System.out.println("After Disconnector, the size of the queue is "+connQueue.size());
}
}
}
public String getIpAddress(){
return ipAddress;
}
}
// 釋放連接器
public class Disconnector implements Runnable {
private LinkedList<Connection> connQueue;
public Disconnector(LinkedList<Connection> connQueue) {
this.connQueue=connQueue;
}
public void run() {
boolean flag=true;
while(flag){
System.out.println("Disconnector disconnects 1 connection!");
if(!connQueue.isEmpty())
connQueue.remove(0);
// 休息 1 秒
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
try {
// 交換給連接器，讓連接器工作！
connQueue =(LinkedList<Connection>) exconn.exchange(connQueue);
} catch (InterruptedException e) {
e.printStackTrace();
}
if(connQueue.size()==0){
System.out.println("There is no connection in the queue!");
}else{
System.out.println(
"After Connector, the size of the queue is "+connQueue.size());
}
}
}
}
public static void main(String[] args) {
LinkedList<Connection> connQueue = new LinkedList<Connection>();
ConnectionQueueDemo connectionQueueDemo = new ConnectionQueueDemoEx();
new Thread(connectionQueueDemo.new Connector(connQueue,"192.168.1.1")).start();
new Thread(connectionQueueDemo.new Disconnector(connQueue)).start();
}
}
在 main 函數裡是具體的 Demo 實現。新建了連接器和釋放連接器兩個線程，它們共享一個連接緩存隊列。由於 ，連接器每次隨機的連接的連接數要大於釋放連機器釋放的連接數，所以最後，連接隊列會滿。後台打印的輸出截屏如下：
圖 3.Exchanger 實現的多線程 Demo 的後台輸出
總結
在交易中間件的測試中，伴隨著 網絡連接的復雜情況，程序員需要模擬不同的場景來測試不同的邊界情況。不僅如此，還需要模擬大量客戶端連接來測試達 到峰值的情況。本文以 IMS 的交易管理器的一個組件 IMS Connect 為例，闡述了如何使用 Java 並發編程的技術來模擬多 個客戶端發出連接請求的方法。
本文也對 Java 多線程編程做出了一些推廣應用。Java 的多線程技術在 JDK1.5 之 後有了許多新的特性和擴展，在推廣的章節，本文利用了 Exchanger 類來實現了客戶端連接和釋放的測試場景。
最 後，希望通過本文的介紹能對會面臨同樣問題或類似問題的程序員們提供一些思路，能有利於測試人員更好的測試交易中間 件，達到峰值測試的效果。
