JMS与WebService负载均衡

众所周知，当多个消息消费者(MessageConsumer)同时监听同一个消息队列(Queue)的时候，JMS提供者会在它们之间提供一种负载均衡机制，从而达到可以同时处理多个消息的目的。图一是一个简单的示意图，当消息生产者发送多个消息时，JMS提供者会把这些消息均匀的分发到不同的消息消费者。　　

　　图一 JMS负载均衡示意图

　　一、WebService负载均衡
　　要在原来的WebService上应用负载均衡，首先应该明确一个前提，就是客户端和服务端应尽可能的不做修改。另外还需要确保可以方便的添加和删除一个Service节点，而不会影响整个应用的运行。庆幸的是WebService调用通常都是无状态的，类似于无状态会话 Bean(Stateless Session Bean)的远程调用，服务端和客户端不需要维持一个会话，也就是说同一个客户端调用多次WebService请求，每个请求可以由不同的Service 为它服务，这样就可以避免Session复制的问题，如果一个Service崩溃了，另一个Service可以继续为客户端服务。
　　二、将JMS负载均衡应用到WebService中
　　接下来就把JMS的负载均衡机制应用到WebService当中去，图二是一个整体的框架图，Proxy作为一个WebService的代理来和客户端交互， 而Listener会去掉用具体的WebService，来完成一次WebService方法的调用。

　　图二 整体框架图

　　与客户端直接交互的是Proxy，它是一个简单的Servlet，它的作用就是拦截客户端发送过来的请求，提取请求中的SOAP包的内容，然后把它封装成一个消息发送到指定的Queue中去，同时它会等待消息的回复，因为回复的消息中包含着WebService调用返回的结果，后面会详细讲到。当Proxy收到回复消息之后，读取其中的内容并把它返回给客户端，从而完成一个WebService的调用。另外需要注意的是，在这里Proxy需要区分出客户端发出的是GET请求，还是POST请求。如果是GET请求，就有可能是客户端正在获取WebService的WSDL。如果是POST请求，那么通常就是客户端在调用一个WebService方法了。下面是一个简单的Proxy实现：
　　1 　　public class ServiceProxy extends HttpServlet {
　　2
　　3 　　private ConnectionFactory factory;
　　4
　　5 　　private Connection connection;
　　6
　　7 　　private Queue queue;
　　8
　　9 　　public void init() throws ServletException {
　　10
　　11 　　super.init();
　　12
　　13 　　try {
　　14
　　15 　　//在这里为了简单起见，采用了ActiveMQ。在实际的J2EE应用中应通过JNDI查找相应的ConnectionFactory和Queue。|||
　　16
　　17 　　factory = new ActiveMQConnectionFactory("vm://localhost");
　　18
　　19 　　connection = factory.createConnection();
　　20
　　21 　　connection.start();
　　22
　　23 　　queue = new ActiveMQQueue("testQueue");
　　24
　　25 　　} catch (JMSException e) {
　　26
　　27 　　e.printStackTrace();
　　28
　　29 　　}
　　30
　　31 　　}
　　32
　　33 　　protected void doGet(HttpServletRequest req, HttpServletResponse resp) {
　　34
　　35 　　if (req.getQueryString().endsWith("wsdl") || req.getQueryString().endsWith("WSDL")) {
　　36
　　37 　　try {
　　38
　　39 　　Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
　　40
　　41 　　Message message = session.createMessage();
　　42
　　43 　　//这个属性用来标示请求是否是WSDL请求，后面的Listener会根据它来调用具体的WebService。
　　44
　　45 　　message.setBooleanProperty("isWsdl", true);
　　46
　　47 　　message.setJMSReplyTo(session.createTemporaryQueue());
　　48
　　49 　　session.createProducer(queue).send(message);
　　50
　　51 　　//等待回复消息，它里面包含Service返回的WSDL，这里使用了一个选择器。
　　52
　　53 　　String selector = String.format("JMSCorrelationID=‘%s‘", message.getJMSMessageID());
　　54
　　55 　　MessageConsumer consumer = session.createConsumer(message.getJMSReplyTo(), selector);
　　56
　　57 　　TextMessage replyMessage = (TextMessage) consumer.receive();
　　58
　　59 　　//将WSDL返回给客户端。
　　60
　　61 　　resp.getWriter().write(replyMessage.getText());
　　62
　　63 　　} catch (Exception e) {
　　64
　　65 　　e.printStackTrace();
　　66
　　67 　　}
　　68
　　69 　　}
　　70
　　71 　　}
　　72
　　73 　　protected void doPost(HttpServletRequest req, HttpServletResponse resp) {
　　74
　　75 　　try {
　　76
　　77 　　//首先从客户端请求中得到SOAP请求部分。
　　78
　　79 　　StringBuffer payLoad = new StringBuffer();
　　80
　　81 　　BufferedReader reader = req.getReader();
　　82
　　83 　　String temp;
　　84
　　85 　　while ((temp = reader.readLine()) != null) {
　　86
　　87 　　payLoad.append(temp);
　　88
　　89 　　}
　　90
　　91 　　//将SOAP请求封装成一个消息，发送到Queue。
　　92
　　93 　　Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
　　94
　　95 　　Message message = session.createTextMessage(payLoad.toString());
　　96
　　97 　　message.setBooleanProperty("isWsdl", false);
　　98
　　99 　　message.setJMSReplyTo(session.createTemporaryQueue());
　　100|||
　　101 　　session.createProducer(queue).send(message);
　　102
　　103 　　//等待回复，回复的消息中包含着服务端返回的SOAP响应。
　　104
　　105 　　String selector = String.format("JMSCorrelationID=‘%s‘", message.getJMSMessageID());
　　106
　　107 　　MessageConsumer consumer = session.createConsumer(message.getJMSReplyTo(), selector);
　　108
　　109 　　TextMessage replyMessage = (TextMessage) consumer.receive();
　　110
　　111 　　//将SOAP响应返回给客户端。
　　112
　　113 　　resp.getWriter().write(replyMessage.getText());
　　114
　　115 　　} catch (Exception e) {
　　116
　　117 　　e.printStackTrace();
　　118
　　119 　　}
　　120
　　121 　　}
　　122
　　123 　　}
　　接下来需要看一下Listener了。对于每一个Service都有一个Listener与之相对应，当一个Listener从Queue中取得一个消息之后，首先应该判断是否是WSDL请求。如果是WSDL请求，则向WebService发送一个GET请求，并将请求的结果，也就是WSDL封装成一个Message回复给Proxy。如果是SOAP请求，则从中取出SOAP请求部分并通过POST方法发送给Service，所得到的返回结果即为Service的SOAP响应，把它回复给Proxy，进而回复给客户端。
　　下面是Listener的一个简单实现，在这里我们使用了HttpClient来发送GET和POST请求。
　　1 　　public class Listener {
　　2
　　3 　　private ConnectionFactory factory;
　　4
　　5 　　private Connection connection;
　　6
　　7 　　private Queue queue;
　　8
　　9 　　private Session session;
　　10
　　11 　　public Listener() throws Exception {
　　12
　　13 　　//为了简单，还是采用了ActiveMQ。
　　14
　　15 　　factory = new ActiveMQConnectionFactory("vm://localhost");
　　16
　　17 　　connection = factory.createConnection();
　　18
　　19 　　connection.start();
　　20
　　21 　　queue = new ActiveMQQueue("testQueue");
　　22
　　23 　　session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
　　24
　　25 　　}
　　26
　　27 　　public void startUp() throws Exception {
　　28
　　29 　　MessageConsumer consumer = session.createConsumer(queue);
　　30
　　31 　　consumer.setMessageListener(new MessageListener() {
　　32
　　33 　　public void onMessage(Message message) {
　　34
　　35 　　try {
　　36
　　37 　　String response;
　　38
　　39 　　//通过判断isWSDL这个属性来判断是否是WSDL请求。
　　40
　　41 　　if (message.getBooleanProperty("isWsdl")) {
　　42
　　43 　　//如果是WSDL请求，则向服务器发送一个GET请求，并等待WSDL响应结果。
　　44
　　45 　　HttpClient client = new HttpClient();|||
　　46
　　47 　　//具体Service的地址，可以使用properties文件等。
　　48
　　49 　　GetMethod m = new GetMethod("http://localhost:8081/hello?wsdl");
　　50
　　51 　　client.executeMethod(m);
　　52
　　53 　　response = m.getResponseBodyAsString();
　　54
　　55 　　} else {
　　56
　　57 　　//如果是SOAP请求，则将SOAP包发送给服务器，并等待SOAP响应。
　　58
　　59 　　HttpClient client = new HttpClient();
　　60
　　61 　　PostMethod m = new PostMethod("http://localhost:8081/hello");
　　62
　　63 　　m.setRequestBody(((TextMessage) message).getText());
　　64
　　65 　　client.executeMethod(m);
　　66
　　67 　　response = m.getResponseBodyAsString();
　　68
　　69 　　}
　　70
　　71 　　//将WSDL响应或SOAP响应封装成一个消息，回复给Proxy。
　　72
　　73 　　MessageProducer producer = session.createProducer(message.getJMSReplyTo());
　　74
　　75 　　TextMessage replyMessage = session.createTextMessage(response);
　　76
　　77 　　//设置JMSCorrelationID，Proxy通过它来得到该回复消息。
　　78
　　79 　　replyMessage.setJMSCorrelationID(message.getJMSCorrelationID());
　　80
　　81 　　producer.send(replyMessage);
　　82
　　83 　　} catch (Exception ex) {
　　84
　　85 　　ex.printStackTrace();
　　86
　　87 　　}
　　88
　　89 　　}
　　90
　　91 　　});
　　92
　　93 　　}
　　94
　　95 　　}
　　三、还需要考虑的
　　1，如果其中一个Service崩溃，我们需要停止相应的Listener，这一点可以在Listener中做到，例如当Listener访问 Service时出错，它可以等待一段时间以后再去监听Queue，同时把已经取得的消息重新发回到Queue中去，以便让其他的Listener处理。当然也可以将Listener放置到Service端，这样Service崩溃，Listener也就跟着不可用了。
　　2，如果某一时刻，所有的Service都不可用，Proxy也不应该无限等待下去，而应该设置一个等待回复消息的超时期限，如果这个期限内没有收到回复消息，则表明Service都不可用，同时通知客户端。
　　3，过期消息的处理，如同上一点，当Service不可用时，Proxy通知客户端出错，但是此时Proxy已经向Queue中发送了一条消息，当Service恢复时，这条消息会被重新处理。我们应该避免这种情况发生，因为这条消息已经过期了。可以设置消息的过期期限小于Proxy等待回复消息的期限，以确保它不会被重新处理。
　　4，最糟糕的情况就是，Proxy或者Queue崩溃，这样就算Service正常，客户端也会调用失败。
　　5，效率，既然我们选择了WebService和负载均衡，那就表示我们应该接受它效率相对低下的弱点，在这里需要说明的是JMS并不是效率的瓶颈，因为Listener只负责拿到消息并发送给Service，这时Service还是运行在多线程下的。