Spirng集成ActiveMQ

﹏ヽ暗。殇╰゛Y 2022-05-22 06:36 140阅读 0赞

使用Spirng集成ActiveMQ：

1.ConnectionFactory：用于管理连接的连接工厂

2.JmsTemplate：用于发送和接受消息的模板类

3.MessageListener：消息监听器

ConnectionFactory：

1.一个Spring提供的连接池

2.JmsTemplate每次发消息都会重新创建连接，会话和productor

3.Spring中提供了SingleConnectionFactory和CachingConnectionFactory

JmsTemplate:

1.由Spring提供，只需要向Spring容器注册这个类就能使用JmsTemplate方便的操作jms

MessageListerner:

1.实现一个onMessage方法，该方法只接受一个Message参数。

队列模式

1.首先导入jar包，这里我用的是maven

<dependencies>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.7.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-core</artifactId>
                <version>5.7.0</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jms</artifactId>
                <version>4.2.5.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-pool</artifactId>
                <version>5.7.0</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context</artifactId>
                <version>4.2.5.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-test</artifactId>
                <version>4.2.5.RELEASE</version>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-test</artifactId>
                <version>5.0.6.RELEASE</version>
                <scope>compile</scope>
            </dependency>
        </dependencies>

Spring与AMQ的配置文件：

1.因为有重合部分所以单独做了一个Commen配置：（spring-amq-commen.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"
           xsi:schemaLocation="
            http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
    
    
        
        <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
              destroy-method="stop">
            <property name="connectionFactory">
                <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                    <property name="brokerURL">
                        <value>tcp://192.168.1.6:61616</value>
                    </property>
                </bean>
            </property>
            <property name="maxConnections" value="100"></property>
        </bean>
    
        
        <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
            <property name="targetConnectionFactory" ref="jmsFactory"/>
            <property name="sessionCacheSize" value="1"/>
        </bean>
    
    </beans>

2.发送者：（spring-amq-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:context="http://www.springframework.org/schema/context"
           xsi:schemaLocation="
            http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
            http://www.springframework.org/schema/context
            http://www.springframework.org/schema/context/spring-context-3.0.xsd">
    
        <import resource="spring-amq-commen.xml"/>
    
        <context:component-scan base-package="com.zzf.spring.jms.service"/>
    
        
        <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
            <property name="connectionFactory" ref="cachingConnectionFactory"/>
            <property name="messageConverter">
                <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
            </property>
        </bean>
    
        
        <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
            <constructor-arg name="name" value="spring-queue"/>
        </bean>
    
        
        
    
    </beans>

3.接受者：（spring-amq-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"
           xsi:schemaLocation="
            http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
    
        <import resource="spring-amq-commen.xml"/>
    
        
        <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
            <constructor-arg index="0" value="spring-queue"/>
        </bean>
    
        
        
    
        
        <bean id="messageListener" class="com.zzf.spring.jms.service.MyMessageListener"></bean>
    
        
        <bean id="jmsContainers" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="cachingConnectionFactory"/>
            <property name="destination" ref="destinationQueue"/>
            <property name="messageListener" ref="messageListener"/>
        </bean>
    
    </beans>

因为有用到JmsTemplate所以先定义一个接口：

package com.zzf.spring.jms.service;
    
    public interface ProducerServic {
        public void sendMsg(String message);
    }

实现类：

启动类（生产者）：

package com.zzf.spring.jms.service;
    
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    class AppProducer {
        public static void main(String[] args) {
    
            //读取配置文件
            ClassPathXmlApplicationContext context=new ClassPathXmlApplicationContext("spring-amq-producer.xml");
            //ProducerServic producerServic= context.getBean(ProducerServic.class);//通过获取class的方式获取bean
            ProducerServic producerServic= (ProducerServic) context.getBean("producerServiceImpl");
            for (int i = 0; i < 10; i++) {
                producerServic.sendMsg("Spring-test="+i);
            }
            context.close();
        }
    }

启动后可以看到：

![70][]

消费者需要实现MessageListerner做消息监听器：

package com.zzf.spring.jms.service;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    public class MyMessageListener implements MessageListener {
    
        public void onMessage(Message msg) {
            if (msg instanceof TextMessage) {
                try {
                    TextMessage txtMsg = (TextMessage) msg;
                    String message = txtMsg.getText();
                    System.out.println("消息内容: " + message);
                } catch (JMSException e) {
                    throw new RuntimeException(e);
                }
            } else {
                throw new IllegalArgumentException("Message must be of type TextMessage");
            }
        }
    }

启动类：

package com.zzf.spring.jms.service;
    
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class AppConsumer {
        public static void main(String[] args) {
            ClassPathXmlApplicationContext context=new ClassPathXmlApplicationContext("spring-amq-consumer.xml");
        }
    }

启动后：

![70 1][]

![70 2][]

![70 3][]

OK……Spring与AMQ的队列模式就这么能用了

主题模式

1.修改xml文件添加主题模式：commen.xml

修改配置容器，增加一个topic.xml

原来的consumer.xml如下所示：

修改接口：

package com.zzf.spring.jms.service;
    
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.Resource;
    import javax.jms.*;
    
    @Service
    public class ProducerServiceImpl implements ProducerServic {
    
        @Autowired
        JmsTemplate jmsTemplate;
    
        @Resource(name = "destinationQueue")
        Destination destination;
    
        @Resource(name = "destinationTopic")
        Destination destinationTopic;
    
        public void sendMsg(final String message) {
            //使用JmsTemplate发送消息
            jmsTemplate.send(destination, new MessageCreator() {
                //创建一个消息
                public Message createMessage(Session session) throws JMSException {
                    TextMessage textMessage = session.createTextMessage(message);
                    return textMessage;
                }
            });
            System.out.println("发送消息"+message);
        }
    
        public void sendMsgTopic(final String message) {
            //使用JmsTemplate发送消息
            jmsTemplate.send(destinationTopic, new MessageCreator() {
                //创建一个消息
                public Message createMessage(Session session) throws JMSException {
                    TextMessage textMessage = session.createTextMessage(message);
                    return textMessage;
                }
            });
            System.out.println("发送消息"+message);
        }
    }

只需要修改Resource的name也行

发布者启动类代码：

package com.zzf.spring.jms.service;
    
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class AppTopicProducer {
    
        public static void main(String[] args) {
    
            //读取配置文件
            ClassPathXmlApplicationContext context=new ClassPathXmlApplicationContext("spring-amq-producer.xml");
            //ProducerServic producerServic= context.getBean(ProducerServic.class);//通过获取class的方式获取bean
            ProducerServic producerServic= (ProducerServic) context.getBean("producerServiceImpl");
            for (int i = 0; i < 10; i++) {
                producerServic.sendMsgTopic("Spring-test="+i);
            }
            context.close();
        }
    }

订阅者启动类代码：

package com.zzf.spring.jms.service;
    
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class AppTopicConsumer {
    
        public static void main(String[] args) {
            ClassPathXmlApplicationContext context=new ClassPathXmlApplicationContext("spring-amq-topic.xml");
        }
    }

效果显示：

![70 4][]

![70 5][]

![70 6][]

![70 7][]

每个订阅者都收到了发布者发送的消息~

[70]: /images/20220522/0267ff9cc7994a569a34610779a8ea1c.png
[70 1]: /images/20220522/dbfddd497a96445f9fab18d972e5ac60.png
[70 2]: /images/20220522/c9ba710807e64be3acda096c3acc7c0d.png
[70 3]: /images/20220522/2ea3c4dc55104d01b605d5fedf588194.png
[70 4]: /images/20220522/e328c4f90f814c6ea07bfa5a8dc50bd5.png
[70 5]: /images/20220522/3c3264a89f874819bc79193ab60d0545.png
[70 6]: /images/20220522/4406ba237ffe4d8ca964ebc5c64e9e95.png
[70 7]: /images/20220522/52942c007a1947db91ca922e68bec692.png