ActiveMQ配置和使用

1.XML配置

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:jms="http://www.springframework.org/schema/jms"
    xsi:schemaLocation="http://www.springframework.org/schema/aop 
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd 
        http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms.xsd">

<!-- 配置messageQueue连接工厂 -->
<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
       <property name="brokerURL" value="${mq.brokerURL}"></property>
</bean>
<!-- 配置jms消息发送模板 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="connectionFactory"></property>
</bean>
<!-- 配置消息队列 -->
<bean id="commonMsgQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <!-- 构造函数参数  队列的名称 -->
        <constructor-arg value="${queue.name.commonMsgQueue}"></constructor-arg>
</bean>
<!-- 自定义监听类 -->
<bean id="commonMsgListener" class="listener.CommonMsgListener">
</bean>
<!-- 配置监听器容器 -->
<bean id="commonMsgListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <!-- MQ连接工厂 -->
    <property name="connectionFactory" ref="connectionFactory"></property>
    <!-- 消息队列 -->
    <property name="destination" ref="jpushQueue" ></property>
    <!-- 消息监听器 -->
    <property name="messageListener" ref="commonMsgListener"></property>
</bean>

brokerURL= failover:(tcp://ip1:61616)?randomize=false
ip为ActiveMQ安装的服务器的地址
randomize=false 在使用多个地址时，按照先后顺序进行连接，如 failover:(tcp://ip1:61616,tcp://ip2:61616)?randomize=false ，这时会先尝试与ip1连接，如果连接不上再连接ip2。此参数默认为true

2.加入消息队列 && 监听器获取消息

Put

public class PutMessage {


//加载消息发送模板
@Resource    //根据名字检索加载类，找不到再根据类型检索加载
private JmsTemplate jmsTemplate;
//加载公共消息队列
@Resource   
private ActiveMQQueue commonMsgQueue;

@Transactional
public void put(final Object message){
    try{
        jmsTemplate.send(commonMsgQueue, new MessageCreator(){
            public Message createMessage(Session session) throws JMSException {
                MapMessage mapMessage = session.createMapMessage();
                mapMessage.setString("messageJson",JSON.toJSONString());
            }
        });
        LOGGER.success("加入队列commonMsgQueue成功,消息内容："+JSON.toJSONString(message));
    }catch(JMSException e){
         LOGGER.warn("加入队列commonMsgQueue失败,消息内容："+JSON.toJSONString(message));
    }
}

Get

public Class CommonMsgListener implements messageListener{
 
    public void onMessage(Message message){            
          if(message instanceof ObjectMessage){
              //实现messageListener即可在实现方法中获取到message对象
          }    
    }    
}

3.在服务器上查看队列生产消费情况

菜单	说明
Number of Pending Messages	等待被消费的消息，还未出队列的消息数量
Number of Consumers	消费者的数量，正在监听等待这个队列消息的消费者的数量
Messages Enqueued	进入队列的消息总量
Message Dequeued	队列被消费的消息总量

访问地址 ip:8161/admin
ip为安装ActiveMQ服务的地址，与配置中的brokerURL相同

管理页面头部.png

菜单	说明
Number of Pending Messages	等待被消费的消息，还未出队列的消息数量
Number of Consumers	消费者的数量，正在监听等待这个队列消息的消费者的数量
Messages Enqueued	进入队列的消息总量
Message Dequeued	队列被消费的消息总量