ActiveMQ的配置以及使用方法

ActiveMQ的简单使用

ActiveMQ是一种开源的，实现了JMS规范的，面向消息(MOM)的中间件，为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信

ActiveMQ接发送消息流程图：

Spring结合ActiveMQ使用

1.pom文件引入依赖，引入jar包

<!--active mq start-->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.13.3</version>
</dependency>
<!--active mq end-->

 2 .spring-mq.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"
xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<!-- ActiveMQ服务地址 -->
<property name="brokerURL" value="${mq.brokerURL}"/>
<property name="userName" value="${mq.userName}"></property>
<property name="password" value="${mq.password}"></property>
<!-- 这里定义重试策略,注意：只有持久化的才会重试-->
<property name="redeliveryPolicyMap" ref="redeliveryPolicyMap"/>
</bean>

<!--这里设置各个消息队列的重发机制-->
<bean id="redeliveryPolicyMap" class="org.apache.activemq.broker.region.policy.RedeliveryPolicyMap">
<property name="redeliveryPolicyEntries">
<list>
<ref bean="smsRedeliveryPolicy"/>
<ref bean="mailRedeliveryPolicy"/>
</list>
</property>
</bean>
<bean id="smsRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
<!--重发次数 延时、延时系数、延时指数开关、目标（重发等待时间1s, 2s, 4s, 8s）-->
<property name="maximumRedeliveries" value="3"/>
<property name="redeliveryDelay" value="1000"/>
<property name="backOffMultiplier" value="2"/>
<property name="useExponentialBackOff" value="true"/>
<property name="destination" ref="smsQueue"/>
</bean>
<bean id="mailRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
<!--重发次数 延时、延时系数、延时指数开关-->
<property name="maximumRedeliveries" value="2"/>
<property name="redeliveryDelay" value="5000"/>
<property name="destination" ref="mailQueue"/>
</bean>

<!--
ActiveMQ为我们提供了一个PooledConnectionFactory，通过往里面注入一个ActiveMQConnectionFactory
可以用来将Connection、Session和MessageProducer池化，这样可以大大的减少我们的资源消耗。
要依赖于 activemq-pool包
-->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="targetConnectionFactory"/>
<property name="maxConnections" value="${mq.pool.maxConnections}"/>
</bean>

<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
<property name="reconnectOnException" value="true"/>
</bean>

<!-- 队列目的地-->
<bean id="smsQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="${sms.queueName}"/>
</bean>
<bean id="mailQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="${mail.queueName}"/>
</bean>

<!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->
<!-- 队列模板 这里配置2个，一个用于分布式业务，一个用于发送邮件-->
<bean id="smsMqJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
<property name="defaultDestination" ref="smsQueue"/>
<!-- 使 deliveryMode, priority, timeToLive设置生效-->
<property name="explicitQosEnabled" value="true" />
<!-- 持久化 如果设置为非持久化MQ服务器重启后MQ中的数据会丢失-->
<property name="deliveryPersistent" value="true"/>
<!--这里注意:如果不开启事务，消息在异常的情况下是不会重试的-->
<property name="sessionTransacted" value="false"/>
</bean>

<bean id="mailMqJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<property name="connectionFactory" ref="connectionFactory"/>
<property name="defaultDestination" ref="mailQueue"/>
<!-- 使 deliveryMode, priority, timeToLive设置生效-->
<property name="explicitQosEnabled" value="true" />
<!-- 持久化 如果设置为非持久化MQ服务器重启后MQ中的数据会丢失-->
<property name="deliveryPersistent" value="true"/>
<!--这里注意:如果不开启事务，消息在异常的情况下是不会重试的-->
<property name="sessionTransacted" value="true"/>
</bean>

<!-- 消息监听实现方法一 -->
<bean id="smsListener" class="com.cn.ssm.mq.listener.SmsMessageListener"/>
<bean id="mailListener" class="com.cn.ssm.mq.listener.MailMessageListener"/>

<!-- 消息接收监听器用于异步接收消息-->
<bean id="smsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="smsQueue"/>
<property name="messageListener" ref="smsListener"/>
<!--这里注意:如果不开启事务，消息在异常的情况下是不会重试的-->
<property name="sessionTransacted" value="true"/>
<!-- 同时启动几个listener消费消息 或者可使用
<property name="concurrency" value="4-8"/>
可以根据消息队列中的消息规模自动调整并行数量，最小4， 最大8个。 -->
<property name="concurrentConsumers" value="1"/>
</bean>

<!- 可定义多个消息监听容器，监听不同队列内容 -->
<bean id="mailContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="mailQueue"/>
<property name="messageListener" ref="mailListener"/>
<!--这里注意:如果不开启事务，消息在异常的情况下是不会重试的-->
<property name="sessionTransacted" value="true"/>
<property name="concurrentConsumers" value="1"/>
</bean>
</beans>

发送端代码：
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:application.xml")
public class ActiveMqProducer {

private final Logger log = LoggerFactory.getLogger(ActiveMqProducer.class);

@Autowired
private JmsTemplate smsMqJmsTemplate;

@Test
public void smsSend() throws Exception {
bizMqJmsTemplate.setSessionTransacted(true);
for (int i = 0; i < 1; i++) {
log.info("==>send message" + i);
bizMqJmsTemplate.send(new MessageCreator() {
@Override
public Message createMessage(Session session)
throws JMSException {
log.info("getTransacted:" + session.getTransacted());
Sms sms = new Sms("你好，你的验证码是", 365986);
return session.createTextMessage(JSONObject
.toJSONString(sms));
}
});
ccbe

log.info("==>finish send message" + i);
}

}

}

接收端代码：
@Component
public class SmsMessageListener implements SessionAwareMessageListener<Message> {

private static final Logger log = LoggerFactory.getLogger(TransactionBizMessageListener.class);
private final String transactionBiz = "testDistributedTransaction";

@Autowired
private TransactionBizService transactionBizService;

/**
* @param message
* @param session
*/
public void onMessage(Message message, Session session) throws JMSException{
//这里建议不要try catch，让异常抛出，通过redeliveryPolicy去重试，达到重试次数进入死信DLQ(Dead Letter Queue)
ActiveMQTextMessage msg = (ActiveMQTextMessage) message;
String ms = ms = msg.getText();
log.info("==>receive message:" + ms);
// 转换成相应的对象
Sms sms = JSONObject.parseObject(ms, Sms.class);
if (sms != null ) {
// do something
//throw new RuntimeException("throw runtimeExcetpion");
} else {
log.info("==>message:" + ms + " sms is null!");
}
}
}

//普通java 代码类实现
//发送端
public static void main(String[] args) {
ConnectionFactory connectionFactory;
Connection connection;
Session session;
Destination destination;
MessageProducer producer;
connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.1.100:61616");
try {
connection = connectionFactory.createConnection();
connection.start();
//第一个参数是是否是事务型消息，设置为true,第二个参数无效
//第二个参数是
//Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。异常也会确认消息，应该是在执行之前确认的
//Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会删除消息。可以在失败的
//时候不确认消息,不确认的话不会移出队列，一直存在，下次启动继续接受。接收消息的连接不断开，其他的消费者也不会接受（正常情况下队列模式不存在其他消费者）
//DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。在需要考虑资源使用时，这种模式非常有效。
//待测试
session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
destination = session.createQueue("test-queue");
producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);//是否持久化，不持久化，一旦服务挂掉 重启，消息会丢失
//优先级不能影响先进先出。。。那这个用处究竟是什么呢呢呢呢

for(int i=0;i<100;i++){

producer.send(session.createTextMessage("send:"+i));
}
producer.close();

} catch (JMSException e) {
e.printStackTrace();
}
}

// 接收端
public static void main(String[] args) {
ConnectionFactory connectionFactory;
// Connection ：JMS 客户端到JMS Provider 的连接
Connection connection = null;
// Session： 一个发送或接收消息的线程
Session session;
// Destination ：消息的目的地;消息发送给谁.
Destination destination;
// 消费者，消息接收者
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.1.100:61616");
try {
// 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操作连接
//这个最好还是有事务
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置
destination = session.createQueue("test-queue");
consumer = session.createConsumer(destination);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
try {
if (null != message) {
System.out.println("收到消息" +((TextMessage)message).getText());
}
} catch (Exception e) {
// TODO: handle exception
}
}
});
} catch (Exception e) {
e.printStackTrace();
}
}