ActiveMQ与Spring整合

Posted 2021-02-19 18岁程序员想当歌手

1、jar包依赖

　　因为ActiveMQ5.12后的jar包自带spring，不方便与spring整合，可以使用ActiveMQ5.11.2的jar包

　　(ActiveMQ默认的端口号为:8161)

 

<!-- spring对于jms支持的jar包 -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
        </dependency>
        <!-- activemq -->
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
        </dependency>

 

2、Spring配置文件

　　2.1、queue:点对点主题(连接ActiveMQ默认的端口号为:61616)

　　　　在该主题下，会自动将生产者每次提供的信息持久化，当出现多个消费者时，只有第一个消费者能获取生产者的信息

　　　　生产者的配置如下:

    <!-- jsm服务厂商提供的连接工厂 -->
    <bean id="targetConnectionFactroy" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.25.132:61616"></property>
    </bean>
    
    <!-- spring用于管理真正的ConnectionFactory -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnectionFactroy"></property>
    </bean>
    
    <!-- spring提供的工具类 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <constructor-arg name="connectionFactory" ref="connectionFactory"></constructor-arg>
    </bean>
    
    <!-- queue -->
    <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg name="name" value="queue_text"></constructor-arg>     //自定义生产者名字
    </bean>

　　生产者demo:

@Component
public class QueueProducer {
    @Autowired
    private JmsTemplate jmsTemplate;
    
    @Autowired
    private Destination queueTextDestination;
    
    /**
     * 发送消息
     */
    public void sendTextMessage(final String text){
        jmsTemplate.send(queueTextDestination, new MessageCreator() {
            
            @Override
            public Message createMessage(Session session) throws JMSException {
                
                return session.createTextMessage(text);
            }
        });
    }
}

　　queue主题下消费者的配置:

    
    <!-- 监听类（用来监听到mq中是否有消息） -->
    <bean id="myMessageListener" class="com.jd.demo.MyMessageListener"></bean>
    <!-- 消息的监听器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="queueTextDestination" />
        <property name="messageListener" ref="myMessageListener" />
    </bean>

　　

　　　　2.1、topic:订阅式主题(连接ActiveMQ默认的端口号为:61616)

　　　　　　在该主题下，生产者的消息不会被持久化，如果没有消费者接收，则消息就会丢失；如果有多个消费者在等待接收消息，生产者的消息会同时被所有消费者接收。

　　　　　　生产者的配置如下:

    <!-- jsm服务厂商提供的连接工厂 -->
    <bean id="targetConnectionFactroy" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.25.132:61616"></property>
    </bean>
    
    <!-- spring用于管理真正的ConnectionFactory -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnectionFactroy"></property>
    </bean>
    
    <!-- spring提供的工具类 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <constructor-arg name="connectionFactory" ref="connectionFactory"></constructor-arg>
    </bean>
    
    <!-- topic -->
    <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg name="name" value="topic_text"></constructor-arg>
    </bean>

　　　　生产者demo:

@Component
public class TopicProducer {
    @Autowired
    private JmsTemplate jmsTemplate;
    
    @Autowired
    private Destination topicTextDestination;
    
    //发送消息
    public void sendTextMessage(final String text){
        jmsTemplate.send(topicTextDestination, new MessageCreator() {
            
            @Override
            public Message createMessage(Session session) throws JMSException {
                
                return session.createTextMessage(text);
            }
        });
    }

}

 

　　　　消费者配置如下:

<!-- 监听类（用来监听到mq中是否有消息） -->
    <bean id="myMessageListener" class="com.jd.demo.MyMessageListener"></bean>
    <!-- 消息的监听器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="topicTextDestination" />
        <property name="messageListener" ref="myMessageListener" />
    </bean>

3、消费者demo

public class MyMessageListener implements MessageListener{

    @Override
    public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage) message;
        try {
            System.out.println("接收到的消息是：" + textMessage.getText());
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}