ActiveMQ

标签: 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

开发步骤:

  1. 创建ActiveMQConnectionFactory,创建工厂需要ip和端口号,指定的是tcp协议
  2. 从工厂中获取连接
  3. 使用连接的start方法,开启连接
  4. 使用连接获取session对象
  5. 使用session创建Destination,就是创建queue,(可以设置queue(点对点)和topic(订阅)
  6. 使用session创建Product,消息的是生产者
  7. 创建消息对象TextMessage,设置消息
  8. 发送消息
  9. 释放资源,关闭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

开发步骤

  1. 创建连接工厂ActiveMQConnectionFactory,需要ip和端口
  2. 从连接工厂中创建连接对象
  3. 执行start方法开始连接
  4. 从连接中创建session对象
  5. 从session中创建Destination对象,设置queue名称(有两种类型queue和topic)
  6. 从session中创建Consumer对象
  7. 接收消息
  8. 打印结果
  9. 关闭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>

消息发送

  1. 创建spring容器
  2. 从容器中获取JMSTemplate对象,发送消息
  3. 定义Destination
  4. 使用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;
            }
        });
    }

}

消息接受

  1. 创建一个类实现MessageListener 接口。业务处理在此类中实现。
  2. 在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

这里写图片描述

版权声明:本文为SIMBA1949原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/SIMBA1949/article/details/79376116