现在的位置: 首页 > 综合 > 正文

JMS实战 JMS实战

2018年05月23日 ⁄ 综合 ⁄ 共 10552字 ⁄ 字号 小 中 大 ⁄ 评论关闭
 

JMS实战

.JMS介绍 
    JMS源于企业应用对于消息中间件的需求，使应用程序可以通过消息进行异步处理而互不影响。Sun公司和它的合作伙伴设计的JMS API定义了一组公共的应用程序接口和相应语法，使得Java程序能够和其他消息组件进行通信。JMS有四个组成部分：JMS服务提供者、消息管理对象、消息的生产者消费者和消息本身。 
1)JMS服务提供者实现消息队列和通知，同时实现消息管理的API。JMS已经是J2EE API的一部分，J2EE服务器都提供JMS服务。 
2) 消息管理对象提供对消息进行操作的API。JMS API中有两个消息管理对象：创建jms连接使用的工厂（ConnectionFactory）和目的地（Destination），根据消息的消费方式的不同ConnectionFactory可以分为QueueConnectionFactory和TopicConnectionFactory，目的地（Destination）可以分为队列（Queue）和主题（Topic）两种。 
3)消息的生产者和消费者。消息的产生由JMS的客户端完成，JMS服务提供者负责管理这些消息，消息的消费者可以接收消息。消息的生产者可以分为――点对点消息发布者（P2P）和主题消息发布者（TopicPublisher）。所以，消息的消费者分为两类：主题消息的订阅者（TopicSubscriber)和点对点消息的接收者（queue receiver） 
4)消息。消息是服务提供者和客户端之间传递信息所使用的信息单元。JMS消息由以下三部分组成： 
　　消息头（header）――JMS消息头包含了许多字段，它们是消息发送后由JMS提供者或消息发送者产生，用来表示消息、设置优先权和失效时间等等，并且为消息确定路由。 
　　属性（property）――用来添加删除消息头以外的附加信息。 
　　消息体（body）――JMS中定义了5种消息体：ByteMessage、MapMessage、ObjectMessage、StreamMessage和TextMessage。 

2.Messages 通信方式 
上面提到JMS通信方式分为点对点通信和发布/订阅方式 
1)点对点方式（point-to-point） 
   点对点的消息发送方式主要建立在 Message Queue,Sender,reciever上，Message Queue 存贮消息，Sneder 发送消息，receive接收消息.具体点就是Sender Client发送Message Queue ,而 receiver Cliernt从Queue中接收消息和"发送消息已接受"到Quere,确认消息接收。消息发送客户端与接收客户端没有时间上的依赖，发送客户端可以在任何时刻发送信息到Queue，而不需要知道接收客户端是不是在运行 
2)发布/订阅 方式（publish/subscriber Messaging） 
    发布/订阅方式用于多接收客户端的方式.作为发布订阅的方式，可能存在多个接收客户端，并且接收端客户端与发送客户端存在时间上的依赖。一个接收端只能接收他创建以后发送客户端发送的信息。作为subscriber ,在接收消息时有两种方法，destination的receive方法，和实现message listener 接口的onMessage 方法。 

3.为什么选用ActiveMQ 
   1）ActiveMQ是一个开放源码 
   2）基于Apache 2.0 licenced 发布并实现了JMS 1.1。 
   3）ActiveMQ现在已经和作为很多项目的异步消息通信核心了 
   4）在很多中小型项目中采用ActiveMQ+SPRING+TOMCAT开发模式。 

4.编程模式 
4.1消息产生者向JMS发送消息的步骤 
(1)创建连接使用的工厂类JMS ConnectionFactory 
(2)使用管理对象JMS ConnectionFactory建立连接Connection 
(3)使用连接Connection 建立会话Session 
(4)使用会话Session和管理对象Destination创建消息生产者MessageSender 
(5)使用消息生产者MessageSender发送消息 
4.2消息消费者从JMS接受消息的步骤 
(1)创建连接使用的工厂类JMS ConnectionFactory 
(2)使用管理对象JMS ConnectionFactory建立连接Connection 
(3)使用连接Connection 建立会话Session 
(4)使用会话Session和管理对象Destination创建消息消费者MessageReceiver 
(5)使用消息消费者MessageReceiver接受消息，需要用setMessageListener将MessageListener接口绑定到MessageReceiver 
消息消费者必须实现了MessageListener接口，需要定义onMessage事件方法。 

5.ActiveMQ运行 
ActiveMQ5.0版本默认启动时，启动了内置的jetty服务器，提供一个demo应用和用于监控ActiveMQ的admin应用。运行%activemq_home%bin/目录下的 activemq.bat , 之后你会看见如下一段话表示启动成功。 
打开http://localhost:8161/admin/queues.jsp ，可以查看相应的queue中是否有消息

 

服务端TopicPublisher.java

6 可用的例子

客户端：TopicListener.java

[java] view
plaincopy

  1. public class TopicListener implements MessageListener {  
  2.   
  3.     private Connection connection;  
  4.     private MessageProducer producer;  
  5.     private Session session;  
  6.     private int count;  
  7.     private long start;  
  8.     private Topic topic;  
  9.     private Topic control;  
  10.   
  11.     private String url = "tcp://localhost:61616";  
  12.       
  13.     private static MyConsumerFrame frame= new MyConsumerFrame();;  
  14.       
  15.     private static Long countMsg = 1L;  
  16.   
  17.     public static void main(String[] argv) throws Exception {  
  18.   
  19.         TopicListener l = new TopicListener();  
  20.         String[] unknown = CommandLineSupport.setOptions(l, argv);  
  21.         if (unknown.length > 0) {  
  22.             System.out.println("Unknown options: " + Arrays.toString(unknown));  
  23.             System.exit(-1);  
  24.         }  
  25.         l.run();  
  26.     }  
  27.   
  28.     public void run() throws JMSException {  
  29.         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);  
  30.         connection = factory.createConnection();  
  31.         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
  32.         topic = session.createTopic("topictest.messages");  
  33.         control = session.createTopic("topictest.control");  
  34.   
  35.         MessageConsumer consumer = session.createConsumer(topic);  
  36.         consumer.setMessageListener(this);  
  37.   
  38.         connection.start();  
  39.   
  40.         producer = session.createProducer(control);  
  41.         System.out.println("Waiting for messages...");  
  42.     }  
  43.   
  44.     private static boolean checkText(Message m, String s) {  
  45.         try {  
  46.             return m instanceof TextMessage && ((TextMessage)m).getText().equals(s);  
  47.         } catch (JMSException e) {  
  48.             e.printStackTrace(System.out);  
  49.             return false;  
  50.         }  
  51.     }  
  52.   
  53.     public void onMessage(Message message) {  
  54.         if (message instanceof ExitRecord) {  
  55.             System.out.println("___________________Object Message!");  
  56.             ExitRecord record = (ExitRecord)message;  
  57.             Long msgCpuId = record.getCpucardId();  
  58.             frame.getTableModle().addRow(new String[]{String.valueOf(countMsg++), String.valueOf(msgCpuId)});  
  59.         }  
  60.           
  61.         if (checkText(message, "SHUTDOWN")) {  
  62.   
  63.             try {  
  64.                 connection.close();  
  65.             } catch (Exception e) {  
  66.                 e.printStackTrace(System.out);  
  67.             }  
  68.   
  69.         } else if (checkText(message, "REPORT")) {  
  70.             // send a report:  
  71.             try {  
  72.                 long time = System.currentTimeMillis() - start;  
  73.                 String msg = "Received " + count + " in " + time + "ms";  
  74.                 producer.send(session.createTextMessage(msg));  
  75.             } catch (Exception e) {  
  76.                 e.printStackTrace(System.out);  
  77.             }  
  78.             count = 0;  
  79.   
  80.         } else {  
  81.   
  82.             if (count == 0) {  
  83.                 start = System.currentTimeMillis();  
  84.             }  
  85.   
  86.             if (++count % 1000 == 0) {  
  87.                 System.out.println("Received " + count + " messages.");  
  88.             }  
  89.         }  
  90.     }  
  91.   
  92.     public void setUrl(String url) {  
  93.         this.url = url;  
  94.     }  
  95.   
  96. }  

[java] view
plaincopy

  1. public class TopicPublisher implements MessageListener {  
  2.   
  3.     private static final char[] DATA = "abcdefghijklmnopqrstuvwxyz".toCharArray();  
  4.   
  5.     private final Object mutex = new Object();  
  6.     private Connection connection;  
  7.     private Session session;  
  8.     private MessageProducer publisher;  
  9.     private Topic topic;  
  10.     private Topic control;  
  11.   
  12.     private String url = "tcp://localhost:61616";  
  13.     private int size = 256;  
  14.     private int subscribers = 1;  
  15.     private int remaining;  
  16.     private int messages = 10000;  
  17.     private long delay;  
  18.     private int batch = 2000;  
  19.   
  20.     private byte[] payload;  
  21.   
  22.     public static void main(String[] argv) throws Exception {  
  23.         TopicPublisher p = new TopicPublisher();  
  24.         String[] unknown = CommandLineSupport.setOptions(p, argv);  
  25.         if (unknown.length > 0) {  
  26.             System.out.println("Unknown options: " + Arrays.toString(unknown));  
  27.             System.exit(-1);  
  28.         }  
  29.         p.run();  
  30.     }  
  31.   
  32.     private void run() throws Exception {  
  33.         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);  
  34.         connection = factory.createConnection();  
  35.         session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
  36.         topic = session.createTopic("topictest.messages");  
  37.         control = session.createTopic("topictest.control");  
  38.   
  39.         publisher = session.createProducer(topic);  
  40.         publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);  
  41.   
  42.         payload = new byte[size];  
  43.         for (int i = 0; i < size; i++) {  
  44.             payload[i] = (byte)DATA[i % DATA.length];  
  45.         }  
  46.   
  47.         session.createConsumer(control).setMessageListener(this);  
  48.         connection.start();  
  49.   
  50.         long[] times = new long[batch];  
  51.         for (int i = 0; i < batch; i++) {  
  52.             if (i > 0) {  
  53.                 Thread.sleep(delay * 1000);  
  54.             }  
  55.             times[i] = batch(messages);  
  56.             System.out.println("Batch " + (i + 1) + " of " + batch + " completed in " + times[i] + " ms.");  
  57.         }  
  58.   
  59.         long min = min(times);  
  60.         long max = max(times);  
  61.         System.out.println("min: " + min + ", max: " + max + " avg: " + avg(times, min, max));  
  62.   
  63.         // request shutdown  
  64.         publisher.send(session.createTextMessage("SHUTDOWN"));  
  65.   
  66.         connection.stop();  
  67.         connection.close();  
  68.     }  
  69.   
  70.     private long batch(int msgCount) throws Exception {  
  71.         long start = System.currentTimeMillis();  
  72.         remaining = subscribers;  
  73.         publish();  
  74.         waitForCompletion();  
  75.         return System.currentTimeMillis() - start;  
  76.     }  
  77.   
  78.     private void publish() throws Exception {  
  79.   
  80.         // send events  
  81.         BytesMessage msg = session.createBytesMessage();  
  82.         msg.writeBytes(payload);  
  83.         for (int i = 0; i < messages; i++) {  
  84.             publisher.send(msg);  
  85.             if ((i + 1) % 1000 == 0) {  
  86.                 System.out.println("Sent " + (i + 1) + " messages");  
  87.             }  
  88.         }  
  89.   
  90.         // request report  
  91.         publisher.send(session.createTextMessage("REPORT"));  
  92.     }  
  93.   
  94.     private void waitForCompletion() throws Exception {  
  95.         System.out.println("Waiting for completion...");  
  96.         synchronized (mutex) {  
  97.             while (remaining > 0) {  
  98.                 mutex.wait();  
  99.             }  
  100.         }  
  101.     }  
  102.   
  103.     public void onMessage(Message message) {  
  104.         synchronized (mutex) {  
  105.             System.out.println("Received report " + getReport(message) + " " + --remaining + " remaining");  
  106.             if (remaining == 0) {  
  107.                 mutex.notify();  
  108.             }  
  109.         }  
  110.     }  
  111.   
  112.     Object getReport(Message m) {  
  113.         try {  
  114.             return ((TextMessage)m).getText();  
  115.         } catch (JMSException e) {  
  116.             e.printStackTrace(System.out);  
  117.             return e.toString();  
  118.         }  
  119.     }  
  120.   
  121.     static long min(long[] times) {  
  122.         long min = times.length > 0 ? times[0] : 0;  
  123.         for (int i = 0; i < times.length; i++) {  
  124.             min = Math.min(min, times[i]);  
  125.         }  
  126.         return min;  
  127.     }  
  128.   
  129.     static long max(long[] times) {  
  130.         long max = times.length > 0 ? times[0] : 0;  
  131.         for (int i = 0; i < times.length; i++) {  
  132.             max = Math.max(max, times[i]);  
  133.         }  
  134.         return max;  
  135.     }  
  136.   
  137.     static long avg(long[] times, long min, long max) {  
  138.         long sum = 0;  
  139.         for (int i = 0; i < times.length; i++) {  
  140.             sum += times[i];  
  141.         }  
  142.         sum -= min;  
  143.         sum -= max;  
  144.         return sum / times.length - 2;  
  145.     }  
  146.   
  147.     public void setBatch(int batch) {  
  148.         this.batch = batch;  
  149.     }  
  150.   
  151.     public void setDelay(long delay) {  
  152.         this.delay = delay;  
  153.     }  
  154.   
  155.     public void setMessages(int messages) {  
  156.         this.messages = messages;  
  157.     }  
  158.   
  159.     public void setSize(int size) {  
  160.         this.size = size;  
  161.     }  
  162.   
  163.     public void setSubscribers(int subscribers) {  
  164.         this.subscribers = subscribers;  
  165.     }  
  166.   
  167.     public void setUrl(String url) {  
  168.         this.url = url;  
  169.     }  
  170. }  
返回
【上篇】
【下篇】

作者:

抱歉!评论已关闭.

返回首页

Copyright © 2013-2018 学步园  保留所有权利.
软文销售 QQ客服：2265327166 点击这里给我发消息（其他合作也可洽谈）
必威体育
必威电竞