mqthread.java

针对IBM的消息中间件MQ编写的程序

package com.face;

import com.mq.MQOperator;

import java.util.*;

public class MQThread extends Thread {
	//Define const for call function
	public static final int IDLE = 0;
	public static final int WAITRETURN = -1;
	public static final int PUT_MESSAGE = 1;
	public static final int GET_MESSAGE = 2;
	public static final int CONN_QMANAGER = 3;
	public static final int LIST_QUEUE = 4;
	
	private IMQDisplay mywindow; //Display message interface class
	private int runflag = 0; //函数是否运行的标志
                             // Should be locked!
	private String MQI_NAME; //MQI的名称
	private int completion_code; //完成码
	private int reason_code; //原因码
	private String return_message; //返回信息
	private boolean suspendFlag; //线程是否运行的标志
	private String[] queue;
	private String QM_NAME; //队列管理器名称
	private String Q_NAME; //队列名称
	private String outMessage; //要发送的消息
	private String inMessage = null; //读取的消息
	private MQOperator mqOp; 
	
	/**
	 * 构造函数
	 */
	public MQThread() {
		this.suspendFlag = false;
	}
	
	/**
	 * 构造函数
	 */
	public MQThread(IMQDisplay listener) {
		this.mywindow = listener;
		this.suspendFlag = false;
		runflag = IDLE;
	}
	
	/**
	 * 设置运行标志位
	 * @param flag 运行标志位
	 * @return boolean
	 */
	public synchronized boolean setrunflag(int flag) {
		/*如果线程还在运行，则等待，不能做任何工作*/
		if(runflag != IDLE) {
			return false;
		}
		runflag = flag;
		return true;
	}
	
	/**
	 * 获取运行标志位
	 * @return int
	 */
	public int getrunflag() {
		return runflag;
	}
	
	/**
	 * 设置队列管理器的名称
	 * @param QMgrName 队列管理器的名称
	 */
	public void setQM_NAME(String QMgrName) {
		QM_NAME = QMgrName;
	}
	
	/**
	 * 取得队列管理器的名称
	 * @return String 队列管理器名称
	 */
	public String getQM_NAME() {
		return QM_NAME;
	}
	
	/**
	 * 设置队列的名称
	 * @param QName 队列的名称
	 */
	public void setQ_NAME(String QName) {
		Q_NAME = QName;
	}
	
	/**
	 * 取得队列的名称
	 * @return String 队列的名称
	 */
	public String getQ_NAME() {
		return Q_NAME;
	}
	
	/**
	 * 取得队列名称数组
	 * @return String[]
	 */
	public String[] getQ_NAMES() {
		return queue;
	}
	
	/**
	 * 设置队列名称数组
	 * @param q
	 */
	public void setQ_NAMES(String[] q) {
		queue = q;
	}
	
	/**
	 * 取得完成码
	 * @return int 完成码
	 */
	public int getComCode() {
		return completion_code;
	}
	
	/**
	 * 设置完成码
	 * @param code 完成码
	 */
	public void setComCode(int code) {
		completion_code = code;
	}
	
	/**
	 * 取得原因码
	 * @return int 原因码
	 */
	public int getReaCode() {
		return reason_code;
	}
	
	/**
	 * 设置原因码
	 * @param code 原因码
	 */
	public void setReaCode(int code) {
		reason_code = code;
	}
	
	/**
	 * 设置发送消息
	 * @param message 要发送的消息
	 */
	public void setOutMessage(String message) {
		outMessage = message;
	}
	
	/**
	 * 获取要发送的消息
	 * @return String 要发送的消息
	 */
	public String getOutMessage() {
		return outMessage;
	}
	
	/**
	 * 设置接收消息
	 * @param message
	 */
	public void setInMessage(String message) {
		inMessage = message;
	}
	
	/**
	 * 获取接收到的消息
	 * @return
	 */
	public String getInMessage() {
		return inMessage;
	}
	
	/**
	 * 获取MQI的名称
	 * @return String MQI名称
	 */
	public String getMQIName() {
		return MQI_NAME;
	}
	
	/**
	 * 设置MQI的名称
	 * @param mqiName MQI名称
	 */
	public void setMQIName(String mqiName) {
		MQI_NAME = mqiName;
	}
	
	/**
	 * 获得返回信息
	 * @return String
	 */
	public String getreturn_message() {
		runflag = IDLE;
		return return_message;
	}	
	
	/**
	 * 屏幕打印
	 * @param ch
	 * @param msg
	 */
	private void myprintln(int ch, String msg) {
		mywindow.println(ch, msg);
		//System.out.println(msg);
	}
	
	public void run() {
		int i;
		//completion_code = 0;
		return_message = "--";
		MQOperator mqOperator = new MQOperator();
		for(i = 0; ;i ++) {
			//myprintln(3, "runflag=" + runflag);
			switch(runflag) {
				case 1:
					MQI_NAME = "PutMessage";
					put(mqOperator);
							
					//return_message = new Date() + "Complete!\n";
					if(completion_code == 0) {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_OK" + "(" 
						                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					} else if(completion_code == 1) {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_WARNING" + "(" 
		                                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					} else {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_FAILED" + "(" 
		                                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					}					
					runflag = WAITRETURN;
					
					
					
					break;
				case 2:
					MQI_NAME = "GetMessage";
					get(mqOperator);
					
					//return_message = new Date() + "Complete!\n";
					if(completion_code == 0) {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_OK" + "(" 
						                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					} else if(completion_code == 1) {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_WARNING" + "(" 
		                                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					} else {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_FAILED" + "(" 
		                                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					}					
					runflag = WAITRETURN;
															
					break;
				case 3:
					MQI_NAME = "MQQueueManager";
					connQMgr(mqOperator);			
					//System.out.println(completion_code);
					//System.out.println(reason_code);
					//return_message = new Date() + "Complete!\n";
					if(completion_code == 0) {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_OK" + "(" 
						                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					} else if(completion_code == 1) {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_WARNING" + "(" 
		                                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					} else {
						return_message = "[MQI]:" + MQI_NAME + "[完成码]:MQCC_FAILED" + "(" 
		                                 + completion_code + ")" + "[原因码]:" + "(" + reason_code + ")";
					}					
					runflag = WAITRETURN;					
					break;
				case 4:
					MQI_NAME = "ListQueues";
					getQueueName(mqOperator);				
					//return_message = new Date() + "Complete!\n";
					return_message = "Connect Success!\n";			
					runflag = WAITRETURN;
					
					break;
				default:
					try {
						//myprintln(4, "sleep");
						sleep(2000);
					} catch(Exception e) {
						
					}
					break;										
			}
		}
	}
	
	/**
	 * 控制线程为被占用
	 */
	public synchronized void mysuspend() {
		suspendFlag = true;
	}
	
	/**
	 * 控制线程重新被使用
	 */
	public synchronized void myresume() {
		suspendFlag = false;
		try {
			notify(); //Wakes up a single thread that is waiting on this object's monitor
		} catch(Exception e) {
			System.out.println("notify error:" + e.getMessage());
		}
		
	}
	
	/**
	 * 发送消息
	 * @param op
	 * @return boolean
	 */
	public boolean put(MQOperator op) {
		mqOp = op;
		if(mqOp.connQManager(QM_NAME) == null) {
			myprintln(3, "输入无效!");
			
			return false;
		} else if(mqOp.openQueue(Q_NAME) == null) {
			//myprintln(3, "选择有误!");
			
			return false;
		} else if(mqOp.getCurQueueDepth() >= mqOp.getMaxQueueDepth() - 100) {
			myprintln(2, "请注意!队列可放消息空间不足!");
			return false;
		} else {
			
			if(mqOp.putMessage(outMessage) != 1) {
				myprintln(2, "发送失败!");
				
			} else {
				myprintln(2, "发送成功!");
				
			}
		}
		completion_code = mqOp.getCompletionCode();
		reason_code = mqOp.getReasonCode();	
		
		mqOp.closeQueue();
		mqOp.disConnection();
		return true;
	}
	
	/**
	 * 读取消息
	 * @param op
	 * @return boolean
	 */
	public boolean get(MQOperator op) {
		mqOp = op;
		if(mqOp.connQManager(QM_NAME) == null) {
			return false;
		} else if(mqOp.openQueue(Q_NAME) == null) {
			return false;
		} else {
			if(mqOp.getCurQueueDepth() == 0) {
				myprintln(2, "队列为空!");			
			} else {
				myprintln(2, "[获取的消息为]:" + mqOp.getMessage());
				//setInMessage(mqOp.getMessage());
			}
		}
		completion_code = mqOp.getCompletionCode();
		reason_code = mqOp.getReasonCode();	
		
		mqOp.closeQueue();
		mqOp.disConnection();
		return true;
	}
	
	/**
	 * 连接队列管理器
	 * @param op
	 * @return boolean
	 */
	public boolean connQMgr(MQOperator op) {
		mqOp = op;
		if(mqOp.connQManager(QM_NAME) == null) {
			myprintln(3, "输入无效,请重新输入队列管理器!");
			
			completion_code = mqOp.getCompletionCode();
			reason_code = mqOp.getReasonCode();	
			
			return false;
		} else {
			myprintln(3, "输入有效,请选择队列!");
			return true;
		}
	}
	
	/**
	 * 获取队列名称
	 * @param op
	 * @return boolean
	 */
	public boolean getQueueName(MQOperator op) {
		mqOp = op;
		queue = mqOp.listQueueName(QM_NAME);
		
		completion_code = mqOp.getCompletionCode();
		reason_code = mqOp.getReasonCode();	
		
		return true;
		
	}

	
}