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>

ActiveMQ安装

1.上传ActiveMQ，并解压

2.进入解压后的文件夹的bin目录，启动ActiveMQ

[root@root bin]# ./activemq start

3.访问http://192.168.37.130:8161/admin/

原始账号：admin

原始密码：admin

ActiveMQ

ActiveMQ

下载地址

ActiveMQ原理

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

queue的使用

Product

Consumer

使用监听器

topic的使用

Product

Consumer1

Consumer2

ActiveMQ整合Spring

消息发送

消息接受

queue方式配置Spring

topic方式配置Spring

ActiveMQ安装

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

ActiveMQ

ActiveMQ

下载地址

ActiveMQ原理

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

queue的使用

Product

Consumer

使用监听器

topic的使用

Product

Consumer1

Consumer2

ActiveMQ整合Spring

消息发送

消息接受

queue方式配置Spring

topic方式配置Spring

ActiveMQ安装

热门文章

最新文章

相关课程

相关电子书

相关实验场景