日期:2014-05-16 浏览次数:20819 次
?
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(