ActiveMQ-阿里云开发者社区

ActiveMQ

下载地址

ActiveMQ官网：http://activemq.apache.org/

ActiveMQ的所有版本下载地址：http://activemq.apache.org/download-archives.html

ActiveMQ原理

首先来看本地通讯的情况，应用程序A和应用程序B运行于同一系统A，它们之间可以借助消息队列技术进行彼此的通讯：应用程序A向队列1发送一条信息，而当应用程序B需要时就可以得到该信息。

其次是远程通讯的情况，如果信息传输的目标改为在系统B上的应用程序C，这种变化不会对应用程序A产生影响，应用程序A向队列2发送一条信息，系统A的MQ发现Q2所指向的目的队列实际上位于系统B，它将信息放到本地的一个特殊队列－传输队列(Transmission Queue)。我们建立一条从系统A到系统B的消息通道，消息通道代理将从传输队列中读取消息，并传递这条信息到系统B，然后等待确认。只有MQ接到系统B成功收到信息的确认之后，它才从传输队列中真正将该信息删除。如果通讯线路不通，或系统B不在运行，信息会留在传输队列中，直到被成功地传送到目的地。这是MQ最基本而最重要的技术–确保信息传输，并且是一次且仅一次(once-and-only-once)的传递。

MQ提供了用于应用集成的松耦合的连接方法，因为共享信息的应用不需要知道彼此物理位置（网络地址）；不需要知道彼此间怎样建立通信；不需要同时处于运行状态；不需要在同样的操作系统或网络环境下运行。

JMS框架基本角色和编程模块

JMS Message Producers : 消息生产者，向消息队列提供消息
JMS Provider(Broker) : JMS提供商，如ActiveMQ，提供JMS Queues/Topics服务
JMS Message listener/Consumer ：接收并使用消息

queue的使用

Product

开发步骤：

创建ActiveMQConnectionFactory，创建工厂需要ip和端口号，指定的是tcp协议
从工厂中获取连接
使用连接的start方法，开启连接
使用连接获取session对象
使用session创建Destination，就是创建queue，（可以设置queue（点对点）和topic（订阅）
使用session创建Product，消息的是生产者
创建消息对象TextMessage，设置消息
发送消息
释放资源，关闭Product、session、connection

public static void main(String[] args) throws Exception {
  // 1. 创建连接工厂ActiveMQConnectionFactory，需要ip和端口61616
  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.37.130:61616");
  // 2. 从连接工厂中创建连接对象
  Connection connection = factory.createConnection();
  // 3. 执行start方法开启连接
  connection.start();
  // 4. 从连接中创建session对象
  // 第一个参数，是否开启事务，JTA分布式事务
  // 第二个参数，是否自动应答，如果第一个参数为true，第二个参数失效
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 从session中创建Destination对象，设置queue名称（有两种类型queue和topic）
  Queue queue = session.createQueue("test-queue");
  // 6. 从session中创建Product对象
  MessageProducer producer = session.createProducer(queue);
  // 7. 创建消息对象
  TextMessage textMessage = new ActiveMQTextMessage();
  // 设置消息内容
  textMessage.setText("开始发消息！");
  // 8. 发送消息
  producer.send(textMessage);
  // 9. 关闭session、连接
  producer.close();
  session.close();
  connection.close();
  }

Consumer

开发步骤

创建连接工厂ActiveMQConnectionFactory，需要ip和端口
从连接工厂中创建连接对象
执行start方法开始连接
从连接中创建session对象
从session中创建Destination对象，设置queue名称（有两种类型queue和topic）
从session中创建Consumer对象
接收消息
打印结果
关闭session、连接

public static void main(String[] args) throws Exception {
  // 1. 创建连接工厂ActiveMQConnectionFactory，需要ip和端口61616
  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.37.130:61616");
  // 2. 使用工厂创建连接
  Connection connection = factory.createConnection();
  // 3. 使用start开启连接
  connection.start();
  // 4. 从连接中创建session对象
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 从session中创建Destination对象，设置queue名字
  Queue queue = session.createQueue("test-queue");
  // 6. 从session中创建Consumer
  MessageConsumer consumer = session.createConsumer(queue);
  // 7， 接收消息,直接获取
  while (true) {
    // 消息超时时间是20秒
    Message message = consumer.receive(20000);
    // 如果消息为空，则跳出死循环
    if (message == null) {
    break;
    }
    // 8. 打印消息
    if (message instanceof TextMessage) {
    // 获取消息
    TextMessage textMessage = (TextMessage) message;
    // 打印
    System.out.println(textMessage.getText());      }
  }
  // 9. 关闭session、连接等
  consumer.close();
  session.close();
  connection.close();
  }

使用监听器

public static void main(String[] args) throws Exception {
  // 1. 创建连接工厂ActiveMQConnectionFactory，需要ip和端口61616
  ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.37.130:61616");
  // 2. 使用工厂创建连接
  Connection connection = factory.createConnection();
  // 3. 使用start开启连接
  connection.start();
  // 4. 从连接中创建session对象
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 从session中创建Destination对象，设置queue名字
  Queue queue = session.createQueue("test-queue");
  // 6. 从session中创建Consumer
  MessageConsumer consumer = session.createConsumer(queue);
  // 7.接收消息
  // 监听器的方式实际上是开启了一个新的线程，专门处理消息的接受
  // 现在的情况是，主线程执行完就结束了，新的线程也跟着没了
  consumer.setMessageListener(new MessageListener() {
    @Override
    public void onMessage(Message message) {
    if (message instanceof TextMessage) {
      // 获取消息
      TextMessage textMessage = (TextMessage) message;
      try {
      // 打印
      System.out.println(textMessage.getText());
      } catch (JMSException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      }
    }
    }
  });
  // 让主线程等待一会，监听器能够有时间执行
  Thread.sleep(10000);
  // 9. 关闭session、连接等
  consumer.close();
  session.close();
  connection.close();
  }

topic的使用

Product

public static void main(String[] args) throws Exception {
  // 1. 创建连接工厂ActiveMQConnectionFactory
  ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
    "tcp://192.168.37.130:61616");
  // 2. 使用工厂创建连接
  Connection connection = activeMQConnectionFactory.createConnection();
  // 3. 使用start方法开启连接
  connection.start();
  // 4. 从连接创建session
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 从session创建Destination对象，设置topic名称
  Topic topic = session.createTopic("test-topic");
  // 6. 从session创建Product
  MessageProducer producer = session.createProducer(topic);
  // 7. 创建消息对象
  TextMessage textMessage = new ActiveMQTextMessage();
  textMessage.setText("topic消息");
  // 8. 发送消息
  producer.send(textMessage);
  // 9. 关闭session、连接等
  producer.close();
  session.close();
  connection.close();
  }
}

Consumer1

public static void main(String[] args) throws Exception {
  // 1. 创建连接工厂ActiveMQConnectionFactory
  ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
    "tcp://192.168.37.130:61616");
  // 2. 从连接工厂创建连接
  Connection connection = activeMQConnectionFactory.createConnection();
  // 3. 使用start方法开启连接
  connection.start();
  // 4. 从连接创建session对象
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 从session创建安Destination，设置topic名称
  Topic topic = session.createTopic("test-topic");
  // 6. 从session创建Consumer对象
  MessageConsumer consumer = session.createConsumer(topic);
  // 7. 接收消息，直接接受
  while (true) {
    Message message = consumer.receive(20000);
    if (message == null) {
    break;
    }
    if (message instanceof TextMessage) {
    TextMessage textMessage = (TextMessage) message;
    // 8. 打印消息
    System.out.println(textMessage.getText());
    }
  }
  // 9. 关闭session、消息等
  consumer.close();
  session.close();
  connection.close();
  }

Consumer2

public static void main(String[] args) throws Exception {
  // 1. 创建连接工厂ActiveMQConnectionFactory
  ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
    "tcp://192.168.37.130:61616");
  // 2. 从连接工厂创建连接
  Connection connection = activeMQConnectionFactory.createConnection();
  // 3. 使用start方法开启连接
  connection.start();
  // 4. 从连接创建session对象
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  // 5. 从session创建安Destination，设置topic名称
  Topic topic = session.createTopic("test-topic");
  // 6. 从session创建Consumer对象
  MessageConsumer consumer = session.createConsumer(topic);
  // 7. 接收消息，直接接受
  // while (true) {
  // Message message = consumer.receive(20000);
  //
  // if (message == null) {
  // break;
  // }
  //
  // if (message instanceof TextMessage) {
  // TextMessage textMessage = (TextMessage) message;
  // // 8. 打印消息
  // System.out.println(textMessage.getText());
  // }
  // }
  // 7.接受消息，使用监听器
  consumer.setMessageListener(new MessageListener() {
    @Override
    public void onMessage(Message message) {
    if (message instanceof TextMessage) {
      TextMessage textMessage = (TextMessage) message;
      try {
      // 打印消息
      System.out.println(textMessage.getText());
      } catch (JMSException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
      }
    }
    }
  });
  // 等待监听器执行
  Thread.sleep(10000);
  // 9. 关闭session、消息等
  consumer.close();
  session.close();
  connection.close();
  }

ActiveMQ整合Spring

加入依赖

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <parent>
  <groupId>com.taotao</groupId>
  <artifactId>taotao-parent</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  </parent>
  <groupId>cn.itcast.activemq</groupId>
  <artifactId>itcast-activemq</artifactId>
  <version>1.0.0-SNAPSHOT</version>
  <dependencies>
  <!-- 加入ActiveMQ依赖 -->
  <dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
  </dependency>
  <!-- 加入spring-jms依赖 -->
  <dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
  </dependency>
  <!-- Spring -->
  <dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-webmvc</artifactId>
  </dependency>
  </dependencies>
</project>

消息发送

创建spring容器
从容器中获取JMSTemplate对象，发送消息
定义Destination
使用JMSTemplate对象发送消息

public class Product {
  public static void main(String[] args) {
  // 1. 创建spring容器
  ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
    "applicationContext-activemq.xml");
  // 2. 从容器中获取JMSTemplate对象
  JmsTemplate jmsTemplate = context.getBean(JmsTemplate.class);
  // 3. 从容器中获取Destination对象
  Destination destination = context.getBean(Destination.class);
  // 4. 使用JMSTemplate发送消息
  jmsTemplate.send(destination, new MessageCreator() {
    @Override
    public Message createMessage(Session session) throws JMSException {
    // 创建消息对象
    TextMessage textMessage = new ActiveMQTextMessage();
    // 设置消息内容
    textMessage.setText("spring整合ActiveMQ");
    // 打印消息
    System.out.println(textMessage.getText());
    return textMessage;
    }
  });
  }
}

消息接受

创建一个类实现MessageListener 接口。业务处理在此类中实现。
在spring容器中配置DefaultMessageListenerContainer对象，引用MessageListener 实现类对象接收消息。

public class MyMessageListener implements MessageListener {
  @Override
  public void onMessage(Message message) {
  if (message instanceof TextMessage) {
    TextMessage textMessage = (TextMessage) message;
    try {
    // 获取消息内容
    String msg = textMessage.getText();
    // 打印消息
    System.out.println("接受消息:" + msg);
    } catch (JMSException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    }
  }
  }
}

queue方式配置Spring

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context"
  xmlns:p="http://www.springframework.org/schema/p" xmlns:aop="http://www.springframework.org/schema/aop"
  xmlns:tx="http://www.springframework.org/schema/tx" xmlns:jms="http://www.springframework.org/schema/jms"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
  http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
  http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
  http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
  http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
  http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.0.xsd">
  <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL" value="tcp://192.168.37.130:61616" />
  </bean>
  <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
  <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
  <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
  <property name="targetConnectionFactory" ref="targetConnectionFactory" />
  </bean>
  <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->
  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
  <property name="connectionFactory" ref="connectionFactory" />
  </bean>
  <!--这个是队列目的地，点对点的 -->
  <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
  <constructor-arg>
    <value>queue</value>
  </constructor-arg>
  </bean>
  <!--这个是主题目的地，一对多的 -->
  <!-- <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> -->
  <!-- <constructor-arg value="topic" /> -->
  <!-- </bean> -->
  <!-- messageListener实现类 -->
  <bean id="myMessageListener" class="top.simba.activemq.spring.MyMessageListener"></bean>
  <!-- 配置一个jsm监听容器 -->
  <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="queueDestination" />
  <property name="messageListener" ref="myMessageListener" />
  </bean>
</beans>

topic方式配置Spring

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
  xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
  xmlns:jms="http://www.springframework.org/schema/jms" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
  http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd
  http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
  http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
  http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
  http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.0.xsd">
  <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL" value="tcp://192.168.37.130:61616" />
  </bean>
  <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
  <bean id="connectionFactory"
  class="org.springframework.jms.connection.SingleConnectionFactory">
  <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
  <property name="targetConnectionFactory" ref="targetConnectionFactory" />
  </bean>
  <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->
  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
  <property name="connectionFactory" ref="connectionFactory" />
  </bean>
  <!--这个是队列目的地，点对点的 -->
  <!-- <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> -->
  <!-- <constructor-arg> -->
  <!-- <value>queue</value> -->
  <!-- </constructor-arg> -->
  <!-- </bean> -->
  <!--这个是主题目的地，一对多的 -->
  <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
  <constructor-arg value="topic" />
  </bean>
  <!-- messageListener实现类 -->
  <bean id="myMessageListener" class="top.simba.activemq.spring.MyMessageListener"></bean>
  <!-- messageListener实现类 -->
  <bean id="myMessageListener2" class="top.simba.activemq.spring.MyMessageListener2"></bean>
  <!-- 配置一个jsm监听容器 -->
  <bean id="jmsContainer"
  class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="topicDestination" />
  <property name="messageListener" ref="myMessageListener" />
  </bean>
  <!-- 配置一个jsm监听容器 -->
  <bean id="jmsContainer2"
  class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="topicDestination" />
  <property name="messageListener" ref="myMessageListener2" />
  </bean>
</beans>