ActiveMQ简单入门使用

布满荆棘的人生 2022-05-27 07:25 224阅读 0赞

参考学习慕课网教程记录笔记  
[https://www.imooc.com/learn/856][https_www.imooc.com_learn_856]

## **win 安装 activeMQ** ##

下载安装包直接启动

[http://activemq.apache.org/activemq-5153-release.html][http_activemq.apache.org_activemq-5153-release.html]

启动最好右键以管理员身份启动

localhost:8161

![这里写图片描述][70]

![这里写图片描述][70 1]

## **队列模式的消息演示** ##

**创建工厂-》创建连接->会话->消费者，生产者，消息->目的地**

pom

<dependencies>
    <dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.9.0</version>
    </dependency>
    </dependencies>

队列生产者代码

package queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    
    /** * Created by Administrator on 2018/4/13. */
    public class AppProducer { 
    /* 向消息中间件发送消息 */
    
    
        private static final String url     = "tcp://localhost:61616";
        private static final String queueName   = "queue-test";
    
        public static void main( String[] args ) throws JMSException
        {
    /* 1.先创建连接工厂 */
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory( url );
    
    /* 2.创建连接 */
            Connection connection = connectionFactory.createConnection();
    
    /* 3.启动连接 */
            connection.start();
    
    /* 4.创建会话 */
            Session session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );
    
    /* 5.创建目的地 */
            Destination destination = session.createQueue( queueName );
    
    /* 6.创建生产者 */
            MessageProducer producer = session.createProducer( destination );
    
            for ( int i = 0; i < 100; i++ )
            {
    /* 7.创建消息 */
                TextMessage textMessage = session.createTextMessage( "test" + i );
    /* 8.发布消息 */
                producer.send( textMessage );
                System.out.println( "发送消息 " + textMessage.getText() );
            }
    
    /* 9.关闭连接 */
            connection.close();
        }
    }

运行队列生产者代码 在 activeMQ中队列可以看到

![这里写图片描述][70 2]

此时还没消费者消费  
建立消费者消费消息

package queue;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    
    /** * Created by Administrator on 2018/4/13. */
    public class APPConsumer { 
    /* 队列消费者 */
        private static final String url     = "tcp://localhost:61616";
        private static final String queueName   = "queue-test";
    
        public static void main( String[] args ) throws JMSException
        {
    /* 1.先创建连接工厂 */
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory( url );
    
    /* 2.创建连接 */
            Connection connection = connectionFactory.createConnection();
    
    /* 3.启动连接 */
            connection.start();
    
    /* 4.创建会话 */
            Session session = connection.createSession( false, Session.AUTO_ACKNOWLEDGE );
    
    /* 5.创建目的地 */
            Destination destination = session.createQueue( queueName );
    
    /* 6.创建消费者 */
            MessageConsumer consumer = session.createConsumer( destination );
    
    /* 7.创建监听器 */
            consumer.setMessageListener( new MessageListener()
                             {
                                 public void onMessage( Message message )
                                 {
    /* 接收消息 */
                                     TextMessage textMessage = (TextMessage) message;
                                     try {
                                         System.out.println( "接收消息 " + textMessage.getText() );
                                     } catch ( JMSException e ) {
                                         e.printStackTrace();
                                     }
                                 }
                             } );
    
    
    /* * 8.关闭连接 * connection.close(); * 消息监听是异步的，要在程序退出后关闭 */
        }
    }

![这里写图片描述][70 3]

此时 ActiveMQ 消息没了，有一个监听者

![这里写图片描述][70 4]

如果一个生产者，两个监听者

生产者：发送100个

![这里写图片描述][70 5]

消费者 1  
![这里写图片描述][70 6]

消费者 2  
![这里写图片描述][70 7]

**队列模式是将生产者生产的消费一半一半分给了两个消费者**

# **主题模式的消息演示** #

代码与队列类似改几个地方就行

修改 创建会话后创建目标创建的是主题模式

一个生产者两个消费者

生产者：

![这里写图片描述][70 8]

消费者1

![这里写图片描述][70 9]

消费者2  
![这里写图片描述][70 10]  
主题模式的生产者的消息多个消费者接收到一样的消息，都接收一遍

## **Spring JMS** ##

使用 Spring集成JMS连接ActiveMQ

**ConnectionFactory 用于管理连接的连接工厂**

一个 Spring为我们提供的连接池
    JmsTemplate每次发消息都会重新创建连接，会话和 proudctor
    spring中提供了SingleConnectionFactory和CachingConnectionFactory

**JmsTemplate 用于发送和接收消息的模板类**

是 spring 提供的，只需向 spring容器内注册这个类就可以使用 JmsTemplate

方便的操作 jms  
JmsTemplate类是线程安全的，可以在整个应用范围使用

**MessageListerner 消息监听器**

实现一个 onMessage方法，该方法只接收一个Message参数

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
    <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.0http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <groupId>com.recar</groupId>
        <artifactId>jms-spring</artifactId>
        <version>1.0-SNAPSHOT</version>
        <dependencies>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.11</version>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context</artifactId>
                <version>4.2.5.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jms</artifactId>
                <version>4.2.5.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-test</artifactId>
                <version>4.2.5.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-core</artifactId>
                <version>5.7.0</version>
                <exclusions>
                    <exclusion>
                        <artifactId>spring-context</artifactId>
                        <groupId>org.springframework</groupId>
                    </exclusion>
                </exclusions>
            </dependency>
        </dependencies>
    </project>

spring对生产者的配置

producer.xml

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/toolhttp://www.springframework.org/schema/tool/spring-tool.xsdhttp://www.springframework.org/schema/contexthttp://www.springframework.org/schema/context/spring-context.xsd">
        
        <context:annotation-config></context:annotation-config>
        
        <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL" value="tcp://localhost:61616"></property>
        </bean>
        
        <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
            <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
        </bean>
        
        <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
            <constructor-arg value="queue"></constructor-arg>
        </bean>
        
        <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
            <property name="connectionFactory" ref="connectionFactory"></property>
        </bean>
        
        <bean class="jms.producer.ProducerServiceImpl"></bean>
    </beans>

创建接口 Producer和实现类，并将实现来 写在配置文件中

实现类

package jms.producer;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import javax.annotation.Resource;
    import javax.jms.*;
    /*** Created by Administrator on 2018/4/13.*/
    public class ProducerServiceImpl implements ProducerService{ 
        @AutowiredJmsTemplate jmsTemplate;
        //以资源的方式注入,因为 有可能多个目的地
        @Resource(name="queueDestination")
        Destination destination;
        public void sendMessage(final String message) {
            jmsTemplate.send(destination, new MessageCreator() {
                public Message createMessage(Session session) throws JMSException { 
                    TextMessage textMessage = session.createTextMessage(message);
                    System.out.println("发送消息 "+textMessage.getText());return textMessage;
      }}) ;
    }}

创建调用生产者的类

package jms.producer;
    
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    
    /** * Created by Administrator on 2018/4/13. */
    public class AppProducer { 
    /* 启动生产者发送消息 */
        public static void main( String[] args )
        {
            ApplicationContext context = new ClassPathXmlApplicationContext( "producer.xml" );
    
            ProducerService service = context.getBean( ProducerService.class );
    
            for ( int i = 0; i < 100; i++ )
            {
                service.sendMessage( "test " + i );
            }
        }
    }

运行发送成功

![这里写图片描述][70 11]

实现关闭连接

((ClassPathXmlApplicationContext)context).close();

不然一直有

![这里写图片描述][70 12]

创建消费者

配置  
consumer.xml

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/toolhttp://www.springframework.org/schema/tool/spring-tool.xsdhttp://www.springframework.org/schema/contexthttp://www.springframework.org/schema/context/spring-context.xsd">
        
        <context:annotation-config></context:annotation-config>
        
        <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL" value="tcp://localhost:61616"></property>
        </bean>
        
        <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
            <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
        </bean>
        
        <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
            <constructor-arg value="queue"></constructor-arg>
        </bean>
        
        <bean id="consumberMessageListener" class="consumer.ConsumerMessageListener"></bean>
        
        <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="connectionFactory"></property>
            <property name="destination" ref="queueDestination"></property>
            <property name="messageListener" ref="consumberMessageListener"></property>
        </bean>
    </beans>

创建监听器

要实现接口 MessageListener

package consumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    
    /** * Created by Administrator on 2018/4/13. */
    public class ConsumerMessageListener implements MessageListener { 
        public void onMessage( Message message )
        {
            TextMessage textMessage = (TextMessage) message;
            try {
                System.out.println( "接受消息 " + textMessage.getText() );
            } catch ( JMSException e ) {
                e.printStackTrace();
            }
        }
    }

调用

package consumer;
    
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    
    /** * Created by Administrator on 2018/4/13. */
    public class AppConsumer { 
        public static void main( String[] args )
        {
            ApplicationContext context = new ClassPathXmlApplicationContext( "consumer.xml" );
        }
    }

直接启动就可以接收消息

接收消息  
![这里写图片描述][70 13]

**主题模式**

配置文件增加 主题模式目的地

<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg value="topic"></constructor-arg>
    </bean>

然后在生产者的目的地注入修改

//以资源的方式注入,因为 有可能多个目的地
    @Resource(name="topicDestination") Destination destination;

在消息监听器容器修改（消费者配置文件）

<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory"></property>
    <property name="destination" ref="topicDestination"></property>
    <property name="messageListener" ref="consumberMessageListener"></property>
    </bean>

就将队列修改为主题模式

启动两个订阅者也就是消费者,一个生产者

生产者  
![这里写图片描述][70 14]

订阅者1

![这里写图片描述][70 15]

订阅者2  
![这里写图片描述][70 16]

[https_www.imooc.com_learn_856]: https://www.imooc.com/learn/856
[http_activemq.apache.org_activemq-5153-release.html]: http://activemq.apache.org/activemq-5153-release.html
[70]: /images/20220527/382a8b8efeb54d429b25759f60b8314a.png
[70 1]: /images/20220527/ba48909b549c4833bc9de8d092eef082.png
[70 2]: /images/20220527/6858011fb9124d0986a6d9a5b7f59b8e.png
[70 3]: /images/20220527/f3b097c982924932946f9fba0b90b424.png
[70 4]: /images/20220527/10d25442a9b9438096b420cee67d56a6.png
[70 5]: /images/20220527/d01c1b1f9995424991bfdd7d6ef2fb13.png
[70 6]: /images/20220527/69dc8f7052fd4d6cbe351ec6d5bdb247.png
[70 7]: /images/20220527/c0a3857c2eeb4642b0ad12b882ae0832.png
[70 8]: /images/20220527/4a6a4c0dc2af4db09b161b3ec0a2211c.png
[70 9]: /images/20220527/73da72636a7f44a8ba2d0b4c77f55eb8.png
[70 10]: /images/20220527/c441dfde617f465b9c6966b4c0d50b9e.png
[70 11]: /images/20220527/a5757417d4b442d7b63fbc8393781129.png
[70 12]: /images/20220527/cdd8aca6167640e7a7cff4e9438c62c3.png
[70 13]: /images/20220527/9a2c261a8dc94366bc46e5f3ca950c6c.png
[70 14]: /images/20220527/d3bcef9daecd405ea77c1aeeb092a9d7.png
[70 15]: /images/20220527/d1fa48c851c146cfaefb1dce1e3d0609.png
[70 16]: /images/20220527/75f5a46999ef44aab55e9e48bc003692.png