第三章 ActiveMQ模拟股票价格涨跌(发布/订阅模式)
2017-10-13 10:51
411 查看
概述:模拟某个用户订阅某些股票,然后向对应用户推送股票的价格
这个例子来自于ActiveMQ In Action源码中
运行Consumer 类
运行Publisher类
![](http://img.blog.csdn.net/20171013105717199?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbTBfMzczNTU5NTE=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center)
![](http://img.blog.csdn.net/20171013110000104?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbTBfMzczNTU5NTE=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center)
注意:不要关闭这个窗口
2、运行Comsumer类
这里需要一个监听类,监听发布者的消息
import java.text.DecimalFormat;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
public class Listener implements MessageListener {
public void onMessage(Message message) { //消费监听器只有一个方法onMessage
try {
MapMessage map = (MapMessage)message; //表示Message本身是map
String stock = map.getString("stock");
double price = map.getDouble("price");
double offer = map.getDouble("offer");
boolean up = map.getBoolean("up");
DecimalFormat df = new DecimalFormat( "#,###,###,##0.00" ); //表示格式数据#表示没有为空,0表示没有赋值为0,例如54.97944 最后格式化54.97
System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t" + (up?"up":"down"));
} catch (Exception e) {
e.printStackTrace();
}
}
}
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Consumer {
private static String brokerURL = "tcp://localhost:61616";
private static transient ConnectionFactory factory; //transient表示序列化时候不会包含在对象中
private transient Connection connection;
private transient Session session;
public Consumer() throws JMSException {
factory = new ActiveMQConnectionFactory(brokerURL); //创建activeMq连接工厂类
connection = factory.createConnection(); //获取连接实例
connection.start(); //开始连接
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //为连接创建一个会话
}
public void close() throws JMSException { //创建关闭动作
if (connection != null) {
connection.close();
}
}
public static void main(String[] args) throws JMSException {
Consumer consumer = new Consumer();
for (String stock : args) {
Destination destination = consumer.getSession().createTopic("STOCKS." + stock); //创建一个感兴趣的话题
MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination); //根据根据感兴趣话题获取消息消费对象
messageConsumer.setMessageListener(new Listener()); //为消费绑定对应监听器
}
}
public Session getSession() {
return session;
}
}
![](http://img.blog.csdn.net/20171013110649898?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbTBfMzczNTU5NTE=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center)
控制台没有任何输出
import java.util.Map;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQMapMessage;
public class Publisher {
protected int MAX_DELTA_PERCENT = 1;
protected Map<String, Double> LAST_PRICES = new Hashtable<String, Double>(); //并发map
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 {
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 {
int idx = 0;
while (true) {
idx = (int)Math.round(stocks.length * Math.random()); // 0 1 2
if (idx < stocks.length) {
break;
}
}
String stock = stocks[idx];
Destination destination = session.createTopic("STOCKS." + stock);
Message message = createStockMessage(stock, session);
System.out.println("Sending: " + ((ActiveMQMapMessage)message).getContentMap() + " on destination: " + destination);
producer.send(destination, message);
}
protected Message createStockMessage(String stock, Session session) throws JMSException {
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);
MapMessage message = session.createMapMessage();
message.setString("stock", stock);
message.setDouble("price", price);
message.setDouble("offer", offer);
message.setBoolean("up", up);
return message;
}
protected double mutatePrice(double price) { //随机改变价格
double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT;
return price * (100 + percentChange) / 100;
}
}
这个也要重复Consumer操作,为main方法传递同样的值(其实这些值就是生成topic原材料,发布者会将消息发送这个topic上,消费者会监听这个topic的信息)
![](http://img.blog.csdn.net/20171013111407429?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbTBfMzczNTU5NTE=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center)
Publisher部分日志
![](http://img.blog.csdn.net/20171013111544137?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbTBfMzczNTU5NTE=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center)
Comsumer部分日志:
![](http://img.blog.csdn.net/20171013111624812?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvbTBfMzczNTU5NTE=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center)
参考《ActiveMQ In Action》书
这个例子来自于ActiveMQ In Action源码中
1、先决条件:
本来只需要maven就可以,但为了方便采用IDE工具Eclipse2、执行顺序
启动ActiveMQ运行Consumer 类
运行Publisher类
3、具体步骤
3.1、启动ActiveMQ
注意:不要关闭这个窗口
2、运行Comsumer类
这里需要一个监听类,监听发布者的消息
import java.text.DecimalFormat;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
public class Listener implements MessageListener {
public void onMessage(Message message) { //消费监听器只有一个方法onMessage
try {
MapMessage map = (MapMessage)message; //表示Message本身是map
String stock = map.getString("stock");
double price = map.getDouble("price");
double offer = map.getDouble("offer");
boolean up = map.getBoolean("up");
DecimalFormat df = new DecimalFormat( "#,###,###,##0.00" ); //表示格式数据#表示没有为空,0表示没有赋值为0,例如54.97944 最后格式化54.97
System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t" + (up?"up":"down"));
} catch (Exception e) {
e.printStackTrace();
}
}
}
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Consumer {
private static String brokerURL = "tcp://localhost:61616";
private static transient ConnectionFactory factory; //transient表示序列化时候不会包含在对象中
private transient Connection connection;
private transient Session session;
public Consumer() throws JMSException {
factory = new ActiveMQConnectionFactory(brokerURL); //创建activeMq连接工厂类
connection = factory.createConnection(); //获取连接实例
connection.start(); //开始连接
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //为连接创建一个会话
}
public void close() throws JMSException { //创建关闭动作
if (connection != null) {
connection.close();
}
}
public static void main(String[] args) throws JMSException {
Consumer consumer = new Consumer();
for (String stock : args) {
Destination destination = consumer.getSession().createTopic("STOCKS." + stock); //创建一个感兴趣的话题
MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination); //根据根据感兴趣话题获取消息消费对象
messageConsumer.setMessageListener(new Listener()); //为消费绑定对应监听器
}
}
public Session getSession() {
return session;
}
}
控制台没有任何输出
3.2运行Publisher类
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.MapMessage;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQMapMessage;
public class Publisher {
protected int MAX_DELTA_PERCENT = 1;
protected Map<String, Double> LAST_PRICES = new Hashtable<String, Double>(); //并发map
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 {
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 {
int idx = 0;
while (true) {
idx = (int)Math.round(stocks.length * Math.random()); // 0 1 2
if (idx < stocks.length) {
break;
}
}
String stock = stocks[idx];
Destination destination = session.createTopic("STOCKS." + stock);
Message message = createStockMessage(stock, session);
System.out.println("Sending: " + ((ActiveMQMapMessage)message).getContentMap() + " on destination: " + destination);
producer.send(destination, message);
}
protected Message createStockMessage(String stock, Session session) throws JMSException {
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);
MapMessage message = session.createMapMessage();
message.setString("stock", stock);
message.setDouble("price", price);
message.setDouble("offer", offer);
message.setBoolean("up", up);
return message;
}
protected double mutatePrice(double price) { //随机改变价格
double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT) - MAX_DELTA_PERCENT;
return price * (100 + percentChange) / 100;
}
}
这个也要重复Consumer操作,为main方法传递同样的值(其实这些值就是生成topic原材料,发布者会将消息发送这个topic上,消费者会监听这个topic的信息)
Publisher部分日志
Comsumer部分日志:
参考《ActiveMQ In Action》书
相关文章推荐
- 【java-activemq】 activemq发布订阅模式
- 分布式消息中间件(二)——ActiveMQ发布-订阅消息模式
- activeMq的发布订阅模式
- (三)ActiveMQ之发布- 订阅消息模式实现
- Spring整合activeMq(二):发布订阅模式
- 3,ActiveMQ-Topic订阅发布模式
- ActiveMQ发布订阅模式
- ActiveMQ发布订阅模式
- 使用ActiveMQ和Spring配置订阅发布模式
- spring整合activemq发布订阅消息模式
- activemq 发布订阅模式
- ActiveMQ发布-订阅消息模式
- ActiveMQ发布订阅模式
- activeMQ发布订阅模式中中常用工具类
- MQ系列-ActiveMQ发布/订阅模式
- Spring Redis与ActiveMQ发布订阅模式源码分析
- 使用Spring配置ActiveMQ的发布订阅模式
- 分布式服务框架学习笔记8 ActiveMQ入门2 发布/订阅模式
- 使用Spring配置ActiveMQ的发布订阅模式
- ActiveMQ的发布订阅模式