activeMQ使用总结 (spring 配置)

1 引用包

通过maven方式，应用activemq依赖包，pom.xml 添加如下信息，

<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.9.0</version>
</dependency>

2 配置文件

配置server.properties文件如下，

#activity config
activity.mq.brokerURL=failover:(tcp://127.0.0.1:61616)

3 bean配置

配置相关bean，包括监听，消息发送，以及broker，queue/topic ,如下spring-activemq.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:amq="http://activemq.apache.org/schema/core" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.9.0.xsd"> 
<context:annotation-config/>

<!-- 配置JMS连接工厂 -->
<bean id="providerConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="${activity.mq.brokerURL}" />
<property name="useAsyncSend" value="true" />
<property name="clientID" value="providerClientConnect" />
<property name="redeliveryPolicy" ref="activeMQRedeliveryPolicy"/>
</bean>

<amq:redeliveryPolicy id="activeMQRedeliveryPolicy" destination="#queueDestination"  maximumRedeliveries="10"/>

<!-- 定义消息Destination -->
<bean id="topicDestination"  class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="SpringTopic"/>
</bean>
<bean id="queueDestination"  class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="signIncomeQueue"/>
</bean>
<!-- 消息发送者客户端 -->
<bean id="providerJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="providerConnectionFactory" />
<!--<property name="defaultDestination" ref="topicDestination" />-->
<property name="defaultDestination" ref="queueDestination" />
<!-- 开启订阅模式 -->
<property name="pubSubDomain" value="false"/>
<!--<property name="receiveTimeout" value="10000" />-->
<!-- deliveryMode, priority, timeToLive 的开关要生效，必须配置为true，默认false-->
<property name="explicitQosEnabled" value="true"/>
<property name="sessionAcknowledgeModeName" value="CLIENT_ACKNOWLEDGE"/>
<!-- 发送模式
DeliveryMode.NON_PERSISTENT=1:非持久 ;
DeliveryMode.PERSISTENT=2:持久
-->
<property name="deliveryMode" value="2"/>
</bean>

<!-- 配置消息消费监听者 -->
<bean id="consumerMessageListener" class="com.company.project.service.mq.ConsumerMessageListener" />

<bean id="consumerListenerClient" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="providerConnectionFactory" />
<property name="concurrentConsumers" value="10"/>
<!--<property name="concurrency" value="10-20"/>-->
<!-- 开启订阅模式 -->
<property name="pubSubDomain" value="true"/>
<!--<property name="destination" ref="topicDestination" />-->
<property name="destination" ref="topicDestination" />
<property name="subscriptionDurable" value="true"/>
<!---这里是设置接收客户端的ID，在持久化时，但这个客户端不在线时，消息就存在数据库里，直到被这个ID的客户端消费掉-->
<property name="clientId" value="consumerClient"/>
<property name="messageListener" ref="consumerMessageListener" />
<!-- 消息应答方式
Session.AUTO_ACKNOWLEDGE  消息自动签收
Session.CLIENT_ACKNOWLEDGE  客户端调用acknowledge方法手动签收
Session.DUPS_OK_ACKNOWLEDGE 不必必须签收，消息可能会重复发送
-->
<property name="sessionAcknowledgeMode" value="2"/>
</bean>

</beans>

4 发送接口

发送时，java最常用的有两种格式，textMessage和mapMessage，

package com.company.project.service.impl;

import com.alibaba.fastjson.JSON;
import com.hisense.hitv.service.IQueueService;
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.jms.*;
import java.util.HashMap;
import java.util.Map;

@Service
public class QueueServiceImpl implements IQueueService {

@Autowired
private JmsTemplate jmsTemplate;

public boolean pushMessage2QueueIncome(String uid, Integer incomeType, Integer incomeValue, Integer productCode) {
sendMqMessageIncome(null, uid, incomeType, incomeValue, productCode);
return false;
}

/**
* 说明:发送的时候如果这里没有显示的指定destination.将用spring xml中配置的destination
* @param destination
*/
private void sendMqMessageIncome(Destination destination, final String uid,final Integer incomeType,
final Integer incomeValue,final Integer productCode){
if(null == destination){
destination = jmsTemplate.getDefaultDestination();
}
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
Map param = new HashMap();
MapMessage mapMessage = session.createMapMessage();
mapMessage.setString("customerId", uid);
param.put("customerId", uid);
mapMessage.setInt("incomeType", incomeType);
param.put("incomeType", incomeType);
mapMessage.setInt("incomeValue", incomeValue);
param.put("incomeValue", incomeValue);
mapMessage.setInt("productCode",productCode);
param.put("productCode", productCode);
TextMessage message= session.createTextMessage();
message.setText(JSON.toJSONString(param));
return message;
}
});
}
}

5 监听接口

监听通过onMessage接口实现，监听broker推送消息

package com.company.project.service.mq;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.listener.SessionAwareMessageListener;

import javax.jms.*;

public class ConsumerMessageListener implements SessionAwareMessageListener{
private static Logger logger = LoggerFactory.getLogger(ConsumerMessageListener.class);

public void onMessage(Message message, Session session) throws JMSException {
MapMessage tm = (MapMessage) message;

try {

logger.info("---------消息消费---------");
logger.info("消息ID:\t" + tm.getJMSMessageID());
} catch (JMSException e) {
session.recover();//唤起重传
e.printStackTrace();
}
}
}