Java中间件入门教程

[TOC]

##前言## 本博客介绍Java中间件的一些知识，仅仅是一些知识储备。 ##中间件## ###中间件概念### 中间件：中间件是一种介于操作系统和应用软件之间的一种软件，它使用系统软件所提供的基础服务（功能），衔接网络上应用系统的各个部分或不同的应用，能够达到资源共享、功能共享的目的。 若是以新一代的中间件系列产品来组合应用，同时配合以可复用的商务对象构件，则应用开发费用可节省至80%。 ###中间件分类###

1. 消息中间件 消息中间件适用与进行网络通讯的系统，建立网络通讯的通道，进行数据和文件的传送 产品：ActiveMQ、ZeroMQ、RabbitMQ、IBM webSphere MQ...
2. 交易中间件 交易中间件管理分布与不同操作系统的数据，实现数据一致性，保证系统的负载均衡 产品：IBM CICS,Bea tuxedo...
3. 对象中间件 保证不同厂家的软件之间的交互访问 产品：IBM componentbroker, iona orbix,borland visibroker...
4. 应用服务器 用来构造internet/intranet应用和其它分布式构件应用 产品：IBM Websphere,Bea weblogic...
5. 安全中间件 以公钥基础设施（pki）为核心的、建立在一系列相关国际安全标准之上的一个开放式应用开发平台 产品：entrust entrust...
6. 应用集成服务器 把工作流和应用开发技术如消息及分布式构件结合在一起，使处理能方便自动地和构件、script 应用、工作流行为结合在一起，同时集成文档和电子邮件 产品：lss flowman、ibm flowmark、vitria businessagiliti

##ESB## ESB，即企业服务总线 松散耦合一直是企业软件开发中的一个很重要的内容，而面向服务的SOA编程在随着ESB的应用得到了进一步的发展，ESB就像服务提供者和服务使用者之间的中间层 ##JMS## JMS，即Java Message Service ESB仅仅是作为一个中间层，所以应用程序之间的消息通讯必须借助JMS，即通过JMS从服务使用者接收消息，并将其转发到相应的服务提供者。 而且，JMS 还定义了可发送的若干不同类型的消息。例如，Text 消息包含消息的字符串表示形式；Object 消息包含序列化的 Java 对象；Map 消息包含键/值对的映射，等等。

附录： MQ DEMO：

1package com.wms.batchMsg;
2
3import java.io.File;
4import java.io.IOException;
5import java.sql.Timestamp;
6import java.text.ParseException;
7import java.util.Date;
8
9import org.apache.log4j.Logger;
10
11import com.ibm.mq.MQEnvironment;
12import com.ibm.mq.MQException;
13import com.ibm.mq.MQGetMessageOptions;
14import com.ibm.mq.MQMessage;
15import com.ibm.mq.MQPutMessageOptions;
16import com.ibm.mq.MQQueue;
17import com.ibm.mq.MQQueueManager;
18import com.ibm.mq.constants.MQConstants;
19
20public class MQUtil {
21	
22	private static String qmName;      
23    private static MQQueueManager qMgr;
24    
25    private static Logger logger = Logger.getLogger(MQUtil.class);
26    
27    static{
28    	try{
29	        MQEnvironment.hostname=ConfigManager.getValue("MQ_MQHost");
30	        MQEnvironment.channel=ConfigManager.getValue("MQ_Server_Channel");
31	        MQEnvironment.CCSID=Integer.parseInt(ConfigManager.getValue("MQ_CCSID"));
32	        MQEnvironment.port=Integer.parseInt(ConfigManager.getValue("MQ_port"));
33	        //MQEnvironment.userID = ConfigManager.getValue("MQ_UserId");
34	        //MQEnvironment.password = ConfigManager.getValue("MQ_pass");
35	        qmName = ConfigManager.getValue("MQ_QMname");
36	        MQEnvironment.properties.put(MQConstants.TRANSPORT_PROPERTY,MQConstants.TRANSPORT_MQSERIES_CLIENT);
37	        qMgr = new MQQueueManager(qmName);
38		}catch(MQException e){
39			e.printStackTrace();
40			logger.info("qManager failed: Completion code " + e.completionCode + " Reason Code is "
41					+ e.reasonCode);
42		}
43    }
44    
45    public static MQQueue getSendQueue(String queueName) {
46		MQQueue sQueue;
47        int openSendOptions = MQConstants.MQOO_OUTPUT | MQConstants.MQOO_FAIL_IF_QUIESCING
48                | MQConstants.MQOO_SET_IDENTITY_CONTEXT;
49        try {
50            sQueue = qMgr.accessQueue(queueName, openSendOptions);
51        } catch (MQException e) {
52            e.printStackTrace();
53            return null;
54        }
55        return sQueue;
56    }
57    
58    public static MQQueue getReceiveQueue(String revQueueName){
59		MQQueue rQueue ;
60		int openRcvOptions = MQConstants.MQOO_INPUT_AS_Q_DEF | MQConstants.MQOO_FAIL_IF_QUIESCING;
61		try{
62			rQueue = qMgr.accessQueue(revQueueName, openRcvOptions);
63		}catch(MQException e){
64			e.printStackTrace();
65			return null;
66		}
67		return rQueue;
68	}
69    
70    public static void sendMsg(MQMsgEntity entity,String queueName) {
71		MQQueue sendQ = null;
72        try {
73            MQMessage qMsg = new MQMessage();
74            byte[] qByte = entity.getMsgStr().getBytes("UTF-8");
75//            String message = entity.getMsgStr();
76            qMsg.messageId = MQConstants.MQMI_NONE;
77            //TODO send and receive
78            if(entity.getCorrelId()!=null){
79            	qMsg.correlationId = entity.getCorrelId();
80            }
81            qMsg.format = MQConstants.MQFMT_STRING;
82            qMsg.write(qByte);
83            MQPutMessageOptions pmo = new MQPutMessageOptions();
84            pmo.options = pmo.options + MQConstants.MQPMO_NEW_MSG_ID;
85            pmo.options = pmo.options + MQConstants.MQPMO_NO_SYNCPOINT;
86            pmo.options = pmo.options + MQConstants.MQPMO_SET_IDENTITY_CONTEXT;
87            sendQ = getSendQueue(queueName);
88            sendQ.put(qMsg, pmo);
89            qMgr.commit();
90            //logger.info("The send message is: " +new String(qByte,"UTF-8"));
91        } catch (MQException e) {
92            logger.info("A WebSphere MQ error occurred : Completion code "
93                            + e.completionCode + " Reason Code is "
94                            + e.reasonCode);
95        } catch (java.io.IOException e) {
96        	logger.info("An error occurred whilst to the message buffer "
97                            + e);
98        }finally{
99        	try{
100        		if(sendQ!=null){
101        			sendQ.close();
102        		}
103			}catch(MQException e){
104				// TODO Auto-generated catch block
105				e.printStackTrace();
106				logger.info("Error for MQ connection:"+e.getMessage());
107			}
108        }
109
110    }
111    
112//    public static void messageHandlerByQueueName(MQMsgEntity entity,String queueName) {
113//    	try {
114//    		if(queueName.equalsIgnoreCase("sap_OrdersQueue")){
115//        		ECOrder order = new ECOrder();
116//        		order.CallOrderCURFC(entity, "ZECI001");
117//        	}else if(queueName.equalsIgnoreCase("sap_OrderPendReqQueue")){
118//        		ECOrderPending orderPending = new ECOrderPending();
119//        		orderPending.CallOrderPendRFC(entity, "ZECI005");
120//        	}else if(queueName.equalsIgnoreCase("sap_OrderPendCancelQueue")){
121//        		ECOrderPending orderPending = new ECOrderPending();
122//        		orderPending.CallCancelOrderPendRFC(entity, "ZECI006");
123//        	}else if(queueName.equalsIgnoreCase("sap_ECReturnsQueue")){
124//        		ECOrder order = new ECOrder();
125//        		order.callOrderCancelRFC(entity, "ZECI001");
126//        	}else if(queueName.equalsIgnoreCase("sap_downpaymentQueue")){
127//        		ECDownPayment downPayment = new ECDownPayment();
128//        		downPayment.callDownPaymentRFC(entity, "ZECI007");
129//        	}else if(queueName.equalsIgnoreCase("sap_360LBPQueue")){
130//        		EC360LBP lbp = new EC360LBP();
131//        		lbp.generateHtmlFromQueue(entity.getMsgStr());
132//        	}
133//		} catch (Exception e) {
134//			e.printStackTrace();
135//			logger.error(e.getMessage());
136//		}
137//    	
138//    }
139    
140    public MQQueueManager generateNewMQQM(){
141		MQQueueManager qMgr = null;
142		try{
143			  
144	        MQEnvironment.hostname=ConfigManager.getValue("MQ_MQHost");
145	        MQEnvironment.channel=ConfigManager.getValue("MQ_Server_Channel");
146	        MQEnvironment.CCSID=Integer.parseInt(ConfigManager.getValue("MQ_CCSID"));
147	        MQEnvironment.port=Integer.parseInt(ConfigManager.getValue("MQ_port"));   
148	        String qmName = ConfigManager.getValue("MQ_QMname");
149	        MQEnvironment.properties.put(MQConstants.TRANSPORT_PROPERTY,MQConstants.TRANSPORT_MQSERIES_CLIENT);
150	        qMgr = new MQQueueManager(qmName);
151	        
152		}catch(MQException e){
153			e.printStackTrace();
154			logger.info("qManager failed: Completion code " + e.completionCode + " Reason Code is "
155					+ e.reasonCode);
156		}
157		return qMgr;
158	}
159    
160    public void MultiThreadGetMqMessage(MQQueueManager qMgr,String queueName){
161    	MQQueue revQ = null;
162        String mqString = null;
163        MQMsgEntity entity = new MQMsgEntity();
164        
165        int openRcvOptions = MQConstants.MQOO_INPUT_AS_Q_DEF | MQConstants.MQOO_FAIL_IF_QUIESCING;
166        try {
167        	MQMessage retrievedMessage = new MQMessage();
168            MQGetMessageOptions gmo = new MQGetMessageOptions();
169            gmo.options += MQConstants.MQPMO_NO_SYNCPOINT;// 
170            gmo.options = gmo.options + MQConstants.MQGMO_WAIT;//
171            gmo.options = gmo.options + MQConstants.MQGMO_FAIL_IF_QUIESCING;// 
172            gmo.waitInterval = MQConstants.MQWI_UNLIMITED;// 
173            gmo.matchOptions = MQConstants.MQMO_MATCH_MSG_ID;
174            retrievedMessage.format=MQConstants.MQFMT_STRING;
175            //  MQC.MQWI_UNLIMITED;
176            revQ = qMgr.accessQueue(queueName, openRcvOptions);
177            revQ.get(retrievedMessage, gmo);
178            qMgr.commit();
179            int length = retrievedMessage.getDataLength();
180            if(length >0){
181            	long startTime = System.currentTimeMillis();
182	            byte[] msg = new byte[length];
183	            retrievedMessage.readFully(msg);
184	            mqString = new String(msg, "UTF-8");
185	            if(queueName.equalsIgnoreCase("sap_360LBPQueue")){
186	            	mqString = mqString.replace("'", "\"");
187	            }
188	            long timeuse = System.currentTimeMillis() - startTime;
189	            Date currentDate = new Date();
190    			Timestamp receiveTimestamp = new Timestamp(currentDate.getTime());
191	            logger.info("=========mqString from "+queueName+" :"+mqString);
192	            DBUtil.insertIntoMQLog("Receive",queueName, mqString, timeuse, "success", "", null, receiveTimestamp);
193	            entity.setMsgStr(mqString);
194	            //messageHandlerByQueueName(entity,queueName);
195	            
196    			
197            }else{
198            	logger.info("Error MQ string Sent!");
199            }
200        }
201        catch (MQException e) {
202        	e.printStackTrace();
203            if (e.reasonCode != 2033) 
204            {
205            	logger.info(e.getMessage());
206                logger.info("Completion code "
207                        + e.completionCode + " Reason Code is " + e.reasonCode);
208            }
209        } catch (IOException e) {
210            logger.info("IO error:" + e.getMessage());
211        } finally{
212        	try{
213        		if(revQ!=null){
214        			revQ.close();
215        		}
216			}catch(MQException mqEx){
217				int rc = mqEx.reasonCode;
218				if (rc != MQException.MQRC_NO_MSG_AVAILABLE)
219				{
220					logger.info(" PUT Message failed with rc = " 
221				+ rc);
222				}
223
224			}
225        }
226    }
227    
228    public static String getMQMessage(String queueName) throws ParseException {
229        MQQueue revQ = null;
230        String mqString = null;
231        MQMsgEntity entity = new MQMsgEntity();
232        try {
233        	MQMessage retrievedMessage = new MQMessage();
234            MQGetMessageOptions gmo = new MQGetMessageOptions();
235            gmo.options += MQConstants.MQPMO_NO_SYNCPOINT;// 
236            gmo.options = gmo.options + MQConstants.MQGMO_WAIT;//
237            gmo.options = gmo.options + MQConstants.MQGMO_FAIL_IF_QUIESCING;// 
238            gmo.waitInterval = MQConstants.MQWI_UNLIMITED;// 
239            gmo.matchOptions = MQConstants.MQMO_MATCH_MSG_ID;
240            retrievedMessage.format=MQConstants.MQFMT_STRING;
241            //  MQC.MQWI_UNLIMITED;
242            revQ = getReceiveQueue(queueName);
243            revQ.get(retrievedMessage, gmo);
244            qMgr.commit();
245            int length = retrievedMessage.getDataLength();
246            if(length >0){
247	            byte[] msg = new byte[length];
248	            retrievedMessage.readFully(msg);
249	            mqString = new String(msg, "UTF-8");
250	            logger.info("=========getMQMessage===mqString from "+queueName+" :"+mqString);
251	            entity.setMsgStr(mqString);
252	            //messageHandlerByQueueName(entity,queueName);
253            }else{
254            	logger.info("Error MQ string Sent!");
255            }
256        }
257        catch (MQException e) {
258        	e.printStackTrace();
259            if (e.reasonCode != 2033) 
260            {
261                e.printStackTrace();
262                logger.info("Completion code "
263                        + e.completionCode + " Reason Code is " + e.reasonCode);
264            }
265        } catch (java.io.IOException e) {
266            System.out.println("error" + e.getMessage());
267        }finally{
268        	try{
269        		if(revQ!=null){
270        			revQ.close();
271        		}
272			}catch(MQException mqEx){
273				int rc = mqEx.reasonCode;
274				if (rc != MQException.MQRC_NO_MSG_AVAILABLE)
275				{
276				System.out.println(" PUT Message failed with rc = " 
277				+ rc);
278				}
279
280			}
281        }
282        return mqString;
283    }
284
285	public void revAndSend(MQMsgEntity entity,String queueName){
286		//
287		sendMsg(entity,queueName);
288	}
289	
290	public void subscribeMessage() throws ParseException{
291		while(true){
292			logger.info("waiting to get message.....");
293			getMQMessage("sap_OrdersQueue");
294			
295		}
296	}
297	
298	public void subscribeOrderPendMessage() throws ParseException{
299		while(true){
300			logger.info("waiting to get message.....");
301			getMQMessage("sap_ECReturnsQueue");
302			
303		}
304	}
305	
306	
307	
308	public  static void main(String[] args) throws IOException, ParseException {
309		MQMsgEntity entity = new MQMsgEntity();
310		String sendMsg = XMLBeanUtil.readFileToString(new File("D://batchXML0108.txt"));
311		int intPktCtlNbr = 1;
312		String StrPkt = null;
313		String newPktCtlNbr =null;
314		for (int i = 0; i < 20000; i++) {
315			newPktCtlNbr = String.format("%09d", intPktCtlNbr+i);  
316			StrPkt="<PktCtlNbr>"+"V"+newPktCtlNbr+"</PktCtlNbr>";
317			String changeSendMsg = sendMsg.replaceAll("<PktCtlNbr>6001996171</PktCtlNbr>", StrPkt);
318			System.out.println(StrPkt);
319			entity.setMsgStr(changeSendMsg);
320			sendMsg(entity,"wms_SAPOrderQueue");
321		}
322//		MQUtil util = new MQUtil();
323//		util.subscribeMessage();
324//		util.subscribeOrderPendMessage();
325//		util.messageHandlerByQueueName(entity, "sap_360LBPQueue");
326//		getMQMessage("sap_OrderPendCancelQueue");
327//		System.out.println("rev message is:"+message);
328	}
329
330}