myJavaBlog

java世界
随笔 - 43, 文章 - 0, 评论 - 2, 引用 - 0
数据加载中……

分享关于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还是运行在多线程下的。
(转载请注明出处:[url=http://www.6rfd.com]澳门博彩[/url]

[url=http://www.9gds.com]易博网[/url]

posted on 2012-04-18 20:30 夏凡 阅读(458) 评论(0)  编辑  收藏


只有注册用户登录后才能发表评论。


网站导航: