理解ActiveMQ中的REST API和AjaxAPI

 

Sending Message as XML Payload，参考代码：

import java.io.StringWriter;
import java.util.Hashtable;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.xml.stream.XMLOutputFactory;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.XMLStreamWriter;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQTextMessage;

public class Publisher {

    protected int MAX_DELTA_PERCENT = 1;
    protected Map<String, Double> LAST_PRICES = new Hashtable<String, Double>();
    protected static int count = 10;
    protected static int total;
    
    protected static String brokerURL = "tcp://localhost:61616";
    protected static transient ConnectionFactory factory;
    protected transient Connection connection;
    protected transient Session session;
    protected transient MessageProducer producer;
    
    public Publisher() throws JMSException {
    	factory = new ActiveMQConnectionFactory(brokerURL);
    	connection = factory.createConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = session.createProducer(null);
    }
    
    public void close() throws JMSException {
        if (connection != null) {
            connection.close();
        }
    }
    
    public static void main(String[] args) throws JMSException, XMLStreamException {
    	Publisher publisher = new Publisher();
        while (total < 1000) {
            for (int i = 0; i < count; i++) {
                publisher.sendMessage(args);
            }
            total += count;
            System.out.println("Published '" + count + "' of '" + total + "' price messages");
            try {
              Thread.sleep(1000);
            } catch (InterruptedException x) {
            }
          }
        publisher.close();
    }

    protected void sendMessage(String[] stocks) throws JMSException, XMLStreamException {
        int idx = 0;
        while (true) {
            idx = (int)Math.round(stocks.length * Math.random());
            if (idx < stocks.length) {
                break;
            }
        }
        String stock = stocks[idx];
        Destination destination = session.createTopic("STOCKS." + stock);
        Message message = createStockMessage(stock, session);
        
        System.out.println("Sending: " + ((ActiveMQTextMessage)message).getText() + " on destination: " + destination);
        producer.send(destination, message);
    }

    protected Message createStockMessage(String stock, Session session) throws JMSException, XMLStreamException {
        Double value = LAST_PRICES.get(stock);
        if (value == null) {
            value = new Double(Math.random() * 100);
        }

        // lets mutate the value by some percentage
        double oldPrice = value.doubleValue();
        value = new Double(mutatePrice(oldPrice));
        LAST_PRICES.put(stock, value);
        double price = value.doubleValue();

        double offer = price * 1.001;

        boolean up = (price > oldPrice);
        
        StringWriter res = new StringWriter();
        XMLStreamWriter writer = XMLOutputFactory.newInstance().createXMLStreamWriter(res);
        
        writer.writeStartDocument();
        writer.writeStartElement("stock");
        writer.writeAttribute("name", stock);
    	writer.writeStartElement("price");
    	writer.writeCharacters(String.valueOf(price));
    	writer.writeEndElement();
    	
    	writer.writeStartElement("offer");
    	writer.writeCharacters(String.valueOf(offer));
    	writer.writeEndElement();
    	
    	writer.writeStartElement("up");
    	writer.writeCharacters(String.valueOf(up));
    	writer.writeEndElement();
        writer.writeEndElement();
        writer.writeEndDocument();
        
		TextMessage message = session.createTextMessage();
		message.setText(res.toString());
		return message;
    }

    protected double mutatePrice(double price) {
        double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT;

        return price * (100 + percentChange) / 100;
    }
}

     输出结果：

Sending: <?xml version='1.0' encoding='UTF-8'?><stock name="JAVA"><price>18.570994459452137</price><offer>18.589565453911586</offer><up>false</up></stock> on destination: topic://STOCKS.JAVA
Sending: <?xml version='1.0' encoding='UTF-8'?><stock name="JAVA"><price>18.608655053355005</price><offer>18.627263708408357</offer><up>true</up></stock> on destination: topic://STOCKS.JAVA
Sending: <?xml version='1.0' encoding='UTF-8'?><stock name="IONA"><price>20.295460590285106</price><offer>20.31575605087539</offer><up>false</up></stock> on destination: topic://STOCKS.IONA
......

 

1. ActiveMQ REST API

    ActiveMQ有一个内嵌的Web Server，与Broker一起启动。该Web Server用于提供ActiveMQ所需要的所有Web

    基础功能，包括REST API.

    REST API由org.apache.activemq.web.MessageServlet实现，需要在web.xml配置该Servlet来进行Expose

    ActiveMQ REST API.例如：

<servlet>
		<servlet-name>MessageServlet</servlet-name>
		<servlet-class>org.apache.activemq.web.MessageServlet</servlet-class>
		<load-on-startup>1</load-on-startup>
	</servlet>
	<servlet-mapping>
		<servlet-name>MessageServlet</servlet-name>
		<url-pattern>/message/*</url-pattern>
	</servlet-mapping>

     通过上述配置，Broker Destination被Expose成工程的相对路径，例如：STOCKS.JAVA Topic被映射的URI：

     http://localhost:8161/demo/message/STOCKS/JAVA?type=topic

     通过该路径可以使用GET、POST进行接受或发送消息至Destination:STOCKS.JAVA.

2. AJAX API