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相同
菜单 | 说明 |
---|---|
Number of Pending Messages | 等待被消费的消息,还未出队列的消息数量 |
Number of Consumers | 消费者的数量,正在监听等待这个队列消息的消费者的数量 |
Messages Enqueued | 进入队列的消息总量 |
Message Dequeued | 队列被消费的消息总量 |
网友评论