ActiveMQ Topic使用（三）转

本篇主要演示ActiveMQ中的Topic使用

 

在开始演示前先引入一些概念：

Q:什么是Topic？

A:Topic就是SUB/HUB即发布订阅模式，是SUB/hub就是一拖N的USB分线器的意思。意思就是一个来源分到N个出口。值得注意的是，此模式下，仅对有效的出口即时发送消息，如publisher发布消息Message A到A,B服务器，此时B服务器服务未开启，则仅有A服务器收到消息Message A，事后B服务器开启但乃收不到消息Message A。此时，publisher再次发布消息Message B时，A,B服务器都能收到Message B。

Q:ActiveMQ 支持哪些消息确认模式？

A:下面的表格来自《ActiveMQ in Action》原版CHAPTER 13 Tuning ActiveMQ for performance 内的节选，本人翻译的如有偏差请指出，括号内的是我个人的一些理解仅供参考。

确认模式	发送确认	描述
Session.AUTO_ACKNOWLEDGE	每条消息消耗时自动发送一条应答消息到ActiveMQ代理。	很慢，但常常作为消息消费者的默认的处理机制。
Session.DUPS_OK_ACKNOWLEDGE	允许消费者发送一条消息消耗应答消息到ActiveMQ代理	当达到预设限制的50%，一条确认消息将会被发回，最快的消费消息的标准方式。（最快往往意味着你要处理更多的东西，比如侦测和丢弃重发的消息）
ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE	为每条消息消费发送一条确认消息。	准许主控制单独确认已授权的消息，但这会很慢。
optimizeAcknowledge	准许消费者发送一条范围内的确认消息到ActiveMQ代理来确认消息消费。	结合Session.AUTO_ACKNOWLEDGE 一条消费消息将会在目标缓冲到65%时发送确认信息到ActiveMQ代理。这是最快的消息消费方式。（虽然作者很推崇，但我认为和上面一样存在一些问题需要处理。）

 

 

下面首先是Publisher（发布者）的例子：

 

publicclass Publisher {

 publicstaticint count = 10;// 每次生成量
 privatestaticint total = 0;// 总发送次数

 privatestatic String brokerURL = "tcp://localhost:61616";// MQ 接口地址
 privatestatictransient ConnectionFactory factory;// JMX连接工厂
 privatetransient Connection connection;// JMX连接
 privatetransient Session session;// JMX会话
 privatetransient MessageProducer producer;// 消息生产这

 /**
 * 发布者构造方法
 *
 * @throws JMSException
 */
 public Publisher() throws JMSException {
 factory = new ActiveMQConnectionFactory(brokerURL);
 connection = factory.createConnection();
 try {
 connection.start();
 } catch (JMSException jmse) {
 connection.close();
 throw jmse;
 }
 // 送连接获得会话，获得的会话消息确认模式详见上面介绍
 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 // 从会话中获得生产者,由于是Topic，仅用于发布消息并未指定目标，所以为null
 producer = session.createProducer(null);
 }

 /**
 * 关闭连接
 *
 * @throws JMSException
 */
 publicvoid close() throws JMSException {
 if (connection != null) {
 connection.close();
 }
 }

 publicstaticvoid main(String[] args) throws JMSException {
 Publisher publisher = new Publisher();
 String[] a = new String[] { "test" };//生产出来的消息，放到哪个库存里
 while (total < 100) {
 for (int i = 0; i < count; i++) {
 publisher.sendMessage(a);
 }
 total += count;
 System.out.println("生产了：" + count + "，总共生产了：" + total);
 try {
 Thread.sleep(1000);
 } catch (InterruptedException x) {
 }
 }
 publisher.close();
 }

 /**
 * 消息发送
 *
 * @param stocks
 * @throws JMSException
 */
 protectedvoid sendMessage(String[] stocks) throws JMSException {
 for (String string : stocks) {
 Destination destination = session.createTopic("STOCKS." + string);
 Message message = createStockMessage(string, session);//构造消息
 System.out.println("发送: " + ((ActiveMQMapMessage) message).getContentMap() + " ，目标为: " + destination);
 producer.send(destination, message);//将消息发送给目标
 }
 }

 /**
 * 消息发送模版
 *
 * @param stock
 * @param session
 * @return
 * @throws JMSException
 */
 protected Message createStockMessage(String stock, Session session) throws JMSException {
 MapMessage message = session.createMapMessage();
 message.setString("stock", stock);
 message.setString("name", "商品");
 message.setDouble("price", 100.00);
 message.setDouble("offer", 50.00);
 message.setBoolean("promotion", false);
 return message;
 }
}

public class Publisher {

	public static int count = 10;// 每次生成量
	private static int total = 0;// 总发送次数

	private static String brokerURL = "tcp://localhost:61616";// MQ 接口地址
	private static transient ConnectionFactory factory;// JMX连接工厂
	private transient Connection connection;// JMX连接
	private transient Session session;// JMX会话
	private transient MessageProducer producer;// 消息生产这

	/**
	 * 发布者构造方法
	 * 
	 * @throws JMSException
	 */
	public Publisher() throws JMSException {
		factory = new ActiveMQConnectionFactory(brokerURL);
		connection = factory.createConnection();
		try {
			connection.start();
		} catch (JMSException jmse) {
			connection.close();
			throw jmse;
		}
		// 送连接获得会话，获得的会话消息确认模式详见上面介绍
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 从会话中获得生产者,由于是Topic，仅用于发布消息并未指定目标，所以为null
		producer = session.createProducer(null);
	}

	/**
	 * 关闭连接
	 * 
	 * @throws JMSException
	 */
	public void close() throws JMSException {
		if (connection != null) {
			connection.close();
		}
	}

	public static void main(String[] args) throws JMSException {
		Publisher publisher = new Publisher();
		String[] a = new String[] { "test" };//生产出来的消息，放到哪个库存里
		while (total < 100) {
			for (int i = 0; i < count; i++) {
				publisher.sendMessage(a);
			}
			total += count;
			System.out.println("生产了：" + count + "，总共生产了：" + total);
			try {
				Thread.sleep(1000);
			} catch (InterruptedException x) {
			}
		}
		publisher.close();
	}

	/**
	 * 消息发送
	 * 
	 * @param stocks
	 * @throws JMSException
	 */
	protected void sendMessage(String[] stocks) throws JMSException {
		for (String string : stocks) {
			Destination destination = session.createTopic("STOCKS." + string);
			Message message = createStockMessage(string, session);//构造消息
			System.out.println("发送: " + ((ActiveMQMapMessage) message).getContentMap() + " ，目标为: " + destination);
			producer.send(destination, message);//将消息发送给目标
		}
	}

	/**
	 * 消息发送模版
	 * 
	 * @param stock
	 * @param session
	 * @return
	 * @throws JMSException
	 */
	protected Message createStockMessage(String stock, Session session) throws JMSException {
		MapMessage message = session.createMapMessage();
		message.setString("stock", stock);
		message.setString("name", "商品");
		message.setDouble("price", 100.00);
		message.setDouble("offer", 50.00);
		message.setBoolean("promotion", false);
		return message;
	}
}

接着是Consumer

publicclass Consumer {

 privatestatic String brokerURL = "tcp://localhost:61616";
 privatestatictransient ConnectionFactory factory;
 privatetransient Connection connection;
 privatetransient Session session;

 public Consumer() throws JMSException {
 factory = new ActiveMQConnectionFactory(brokerURL);
 connection = factory.createConnection();
 connection.start();
 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 }

 publicvoid close() throws JMSException {
 if (connection != null) {
 connection.close();
 }
 }

 publicstaticvoid main(String[] args) throws JMSException {
 Consumer consumer = new Consumer();
 String[] stocks=new String[]{"test"};//数据仓库名
 for (String stock : stocks) {
 Destination destination = consumer.getSession().createTopic("STOCKS." + stock);
 MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);
 messageConsumer.setMessageListener(new MyListener());
 }
 }

 public Session getSession() {
 return session;
 }

}

public class Consumer {  
  
    private static String brokerURL = "tcp://localhost:61616";  
    private static transient ConnectionFactory factory;  
    private transient Connection connection;  
    private transient Session session;  
      
    public Consumer() throws JMSException {  
        factory = new ActiveMQConnectionFactory(brokerURL);  
        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();  
        String[] stocks=new String[]{"test"};//数据仓库名
        for (String stock : stocks) {  
            Destination destination = consumer.getSession().createTopic("STOCKS." + stock);  
            MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);  
            messageConsumer.setMessageListener(new MyListener());  
        }  
    }  
      
    public Session getSession() {  
        return session;  
    }  
  
} 

最后是一个消息监听器，十分简单

publicclass MyListener implements MessageListener {
 publicstaticint a=0;
 @Override
 publicvoid onMessage(Message message) {
 System.out.println("进入监听");
 try {
 MapMessage map = (MapMessage) message;
 String stock = map.getString("stock");
 String name = map.getString("name");
 double price = map.getDouble("price");
 double offer = map.getDouble("offer");
 boolean promotion = map.getBoolean("promotion");
 DecimalFormat df = new DecimalFormat("#,###,###,##0.00");
 System.out.println("仓库号:"+stock + "，商品名："+name+"，销售价格：" + df.format(price) + "，供应价格：" + df.format(offer)
 + "，是否促销：" + (promotion ? "是" : "否")+"，共获得"+ ++a);
 } catch (Exception e) {
 e.printStackTrace();
 }
 }

}

public class MyListener implements MessageListener {
	public static int a=0;
	@Override
	public void onMessage(Message message) {
		System.out.println("进入监听");
		try {
			MapMessage map = (MapMessage) message;
			String stock = map.getString("stock");
			String name = map.getString("name");
			double price = map.getDouble("price");
			double offer = map.getDouble("offer");
			boolean promotion = map.getBoolean("promotion");
			DecimalFormat df = new DecimalFormat("#,###,###,##0.00");
			System.out.println("仓库号:"+stock + "，商品名："+name+"，销售价格：" + df.format(price) + "，供应价格：" + df.format(offer)
				+ "，是否促销：" + (promotion ? "是" : "否")+"，共获得"+ ++a);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

使用方法：（不太喜欢传图片，自己执行代码看结果吧）

启动2个Consumer，成功启动后，启动publisher，可以看到2个Consumer正常打印了publisher传过来的所有数据和条数一一对应。

 

启动1个Consumer，成功启动后启动publisher，在publisher未生产完所有数据前，启动另一个Consumer，可以看到先启动的一个正常接收了所有数据，另一个只接收了他启动后publisher生产的数据。