【ActiveMQ入门-10】ActiveMQ学习-通配符+异步接收

时间:2024-10-12 15:34:07
通配符介绍:

一般情况下,我们使用层次结构的方式来组织队列,比如A.B.C.D,这样便于归类和管理。
 我们也可以使用通配符来配置或是操作多个队列。
 通配符有三个:
  1. .  用来分隔路径;
  2. * 用来匹配路径中的一节
  3. > 用来匹配任意节的路径

然而, 通配符中是为消费者服务的。==>即:消费者可以使用通配符来匹配目的地,而生产者不行。

如果你发送了这样的一个主题:" rugby.>.", 这个消息会发送到命名为“rugby.>.”的主题,并不是所有以rugby开头的的主题。

 这里有一种  方法,使消息生产者能将一条。 
 消息发送到多个目的地。通过使用   composite destination。《本小节暂不讲解》

程序概要:

1. 消费者:异步接收,实现

MessageListener类;

2.从消息中获取 目的地名称: message.getJMSDestination().toString();
3.消费模式:发布/订阅模式;
4.目的地名称:采用通配符形式:on3000.topic.*;
【ActiveMQ入门-10】ActiveMQ学习-通配符+异步接收

环境:

1.JmsMessageListener.java
2.Publisher.java
3.Test.java
4.applicationContext-2.xml

【ActiveMQ入门-10】ActiveMQ学习-通配符+异步接收
所使用的jar包
【ActiveMQ入门-10】ActiveMQ学习-通配符+异步接收
activemq-all-5.8.0.jar
activemq-all-5.8.0.source.jar
commons-logging-1.1.1.jar
spring-asm-3.1.2.RELEASE.jar
spring-beans-3.1.2.RELEASE.jar
spring-context-3.1.2.RELEASE.jar
spring-context-support-3.1.2.RELEASE.jar
spring-core-3.1.2.RELEASE.jar
spring-expression-3.1.2.RELEASE.jar
spring-jms-3.1.2.RELEASE.jar
spring-tx-3.1.2.RELEASE.jar



源文件:

JmsMessageListener.java 异步接收

  1. package com.nari.spring.jms2;
  2. import javax.jms.JMSException;
  3. import javax.jms.Message;
  4. import javax.jms.MessageListener;
  5. import javax.jms.TextMessage;
  6. public class JmsMessageListener implements MessageListener {
  7. public void onMessage(Message message) {
  8. System.out.println("消息全部内容:" + message.toString());
  9. try {
  10. System.out.println("消息主题:" + message.getJMSDestination().toString());
  11. } catch (JMSException e1) {
  12. e1.printStackTrace();
  13. }
  14. TextMessage tm = (TextMessage) message;
  15. try {
  16. System.out.println("消息体:" + tm.getText());
  17. } catch (JMSException e) {
  18. e.printStackTrace();
  19. }
  20. }
  21. }


Publisher.java

  1. package com.nari.spring.jms2;
  2. import java.util.Scanner;
  3. import javax.jms.Destination;
  4. import javax.jms.JMSException;
  5. import javax.jms.Message;
  6. import javax.jms.Session;
  7. import javax.jms.TextMessage;
  8. import org.apache.activemq.command.ActiveMQTopic;
  9. import org.springframework.jms.core.JmsTemplate;
  10. import org.springframework.jms.core.MessageCreator;
  11. public class Publisher {
  12. //下面两个field,在applicationContext-*.xml配置
  13. private JmsTemplate template;
  14. private Destination destination;
  15. public void sendMessage() {
  16. long keyValue = 302001011;
  17. boolean sendMsgFlag = true;
  18. int addIndex = 0;
  19. System.out.println("输入主题内容,输入N停止发送消息:");
  20. while (sendMsgFlag) {
  21. // 从终端输入信息
  22. Scanner cin = new Scanner(System.in);
  23. String text = cin.nextLine();
  24. if (text.equals("N")) {
  25. sendMsgFlag = false;
  26. }
  27. // 目的地地址为:topic://on3000.topic.*
  28. int startIndex = destination.toString().indexOf("//");
  29. int endIndex = destination.toString().indexOf("*");
  30. // 拼接新的主题:类似 on3000.topic.30200101112
  31. String subTopicDestination = destination.toString().substring(
  32. startIndex + 2, endIndex)
  33. + Long.toString(keyValue + addIndex);
  34. //发送消息
  35. jmsTemplateSend(subTopicDestination, text);
  36. addIndex++;
  37. }
  38. }
  39. /**
  40. * 向指定主题发送指定消息
  41. *
  42. * @param destinationString
  43. * :主题
  44. * @param strMessage
  45. * :消息内容
  46. */
  47. protected void jmsTemplateSend(String destinationString,
  48. final String strMessage) {
  49. ActiveMQTopic topicDestination = new ActiveMQTopic(destinationString);
  50. template.send(topicDestination, new MessageCreator() {
  51. public Message createMessage(Session session) throws JMSException {
  52. TextMessage message = session.createTextMessage();
  53. message.setText(strMessage);
  54. return message;
  55. }
  56. });
  57. }
  58. public JmsTemplate getJmsTemplate() {
  59. return template;
  60. }
  61. public void setJmsTemplate(JmsTemplate template) {
  62. this.template = template;
  63. }
  64. public Destination getDestination() {
  65. return destination;
  66. }
  67. public void setDestination(Destination destination) {
  68. this.destination = destination;
  69. }
  70. }


Test.java

  1. package com.nari.spring.jms2;
  2. import org.springframework.context.ApplicationContext;
  3. import org.springframework.context.support.ClassPathXmlApplicationContext;
  4. public class Test {
  5. public static void main(String[] args) {
  6. ApplicationContext context = new ClassPathXmlApplicationContext(
  7. "applicationContext-2.xml");
  8. // applicationContext-*.xml初始化时,貌似会自动开启subscribe,下面两行注不注释掉都可以
  9. // DefaultMessageListenerContainer subscribe = (DefaultMessageListenerContainer)context.getBean("consumer");
  10. // subscribe.start();
  11. Publisher publisher = (Publisher) context.getBean("publisher");
  12. publisher.sendMessage();
  13. }
  14. }


配置文件:

applicationContext-2.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p"
  4. xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
  5. <!--创建连接工厂 -->
  6. <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  7. <property name="brokerURL" value="tcp://localhost:61616" />
  8. </bean>
  9. <!-- 声明目标,ActiveMQQueue或ActiveMQTopic -->
  10. <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"
  11. autowire="constructor">
  12. <!-- 通配符,匹配多个主题 -->
  13. <constructor-arg value="on3000.topic.*" />
  14. </bean>
  15. <!-- 创建JMS发送信息的模板的对象 -->
  16. <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  17. <property name="connectionFactory" ref="connectionFactory" />
  18. <property name="defaultDestination" ref="topicDestination" />
  19. <property name="receiveTimeout" value="6000" />
  20. </bean>
  21. <!-- 生成者 -->
  22. <bean id="publisher" class="com.nari.spring.jms2.Publisher">
  23. <property name="jmsTemplate" ref="jmsTemplate" />
  24. <property name="destination" ref="topicDestination" />
  25. </bean>
  26. <!-- 消息监听接口 -->
  27. <bean id="jmsMessageListener" class="com.nari.spring.jms2.JmsMessageListener">
  28. </bean>
  29. <!-- 消费者,通过消息侦听器实现 -->
  30. <bean id="consumer"
  31. class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  32. <property name="connectionFactory" ref="connectionFactory" />
  33. <property name="destination" ref="topicDestination" />
  34. <property name="messageListener" ref="jmsMessageListener" />
  35. </bean>
  36. </beans>


运行结果:

【ActiveMQ入门-10】ActiveMQ学习-通配符+异步接收



附件列表