ActiveMQ

Spring整合ActiveMQ

pom.xml依赖

<!-- pom.xml增加依赖-->
<!-- activemq所对JMS的支持，整合Spring和Activemq-->
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
    <version>4.3.23.RELEASE</version>
</dependency>
<!--activemq所需要的pool包配置-->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
    <version>5.15.9</version>
</dependency>
<!--activemq相关依赖配置-->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.15.9</version>
</dependency>
<dependency>
    <groupId>org.apache.xbean</groupId>
    <artifactId>xbean-spring</artifactId>
    <version>3.16</version>
</dependency>

applicationContext.xml

<!-- applicationContext.xml-->
<!-- 开启包的自动扫描-->
<context:component-scan bean-package="*" />
<!--配置生产者-->
<bean id ="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
    <property name="connectionFactory">
        <!--真正可以产生Connection的ConnectionFactory，由对应的JMS服务厂商提供-->
        <bean class="org.apache.activemq.ActiveMQConnectionFactory">
        	<property name="brokerURL" value="tcp://192.168.1.132:61616"></property>
        </bean>
    </property>
    <property name="maxConnections" value="100"></property>
</bean>

<!--这是个队列目的地，点对点的-->
<bean id ="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
    <!--构造注入-->
    <constructor-arg index="0" value="spring-active-queue" />
</bean>
 
<!-- <bean id ="desctinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg index="0" value="spring-active-topic">
</bean> -->

<!--Spring提供的JMS工具类，它可以进行消息、接收等-->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="jmsFactory" />
    <!--消息默认目的地，ref根据提供的bean配置Queue/Topic的bean id-->
    <property name="defaultDestination" ref="destinationQueue" />
    <property name="messageConverter">
        <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />
    </property>
</bean>

生产者

@Service
public class SpringMQ_Produce {
    @Autowired
    private JmsTemplate jmsTemplate;
    public static void main(String[] args){
        ApplicationContext ctx = new ClassPathXmlAppalicationContext("applicationContext.xml");
        SpringMQ_Produce produce = (SpringMQ_Produce)ctx.getBean("springMQ_Produce");
        produce.jmsTemplate.send(new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage message = session.createTextMessage("Spring和ActiveMQ的整合");
                return message;
            }
        });
    }
}

消费者

@Service
public class SpringMQ_Consumer {
    @Autowired
    private JmsTemplate jmsTemplate;
    public static void main(String[] args){
        ApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");
        SpringMQ_Consumer consumer = (SpringMQ_Consumer)ctx.getBean("springMQ_Consumer");
        String value = (String)consumer.jmsTemplate.receiveAndConvert();
        System.out.println("消费者收到消息：" + value);
    }
}

监听器配置

<!-- applicationContext.xml 增加bean id-->
<!--配置监听程序-->
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="jmsFactory" />
    <property name="destination" ref="destinationQueue" />
    <!--public class MyMessageListener implements MessageListener-->
    <property name="messageListener" ref="myMessageListener"/>
</bean>

@Component
public class MyMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
         if (message != null && message instanceof TextMessage) {
             TextMessage textMessage = (TextMessage) message;
             try {
                 System.out.println(textMessage.getText());
             } catch (JMSException e) {
                 e.printStackTrace();
             }
         }
    }
}