springboot整合activemq，应答模式，消息重发机制，消息持久化

准备工作：

activemq的消息确认机制就是文档中说的ack机制有：

AUTO_ACKNOWLEDGE = 1    自动确认

CLIENT_ACKNOWLEDGE = 2    客户端手动确认   

DUPS_OK_ACKNOWLEDGE = 3    自动批量确认

SESSION_TRANSACTED = 0    事务提交并确认

INDIVIDUAL_ACKNOWLEDGE = 4    单条消息确认 activemq 独有

ACK模式描述了Consumer与broker确认消息的方式(时机),比如当消息被Consumer接收之后,Consumer将在何时确认消息。

对于broker而言，只有接收到ACK指令,才会认为消息被正确的接收或者处理成功了,通过ACK，可以在consumer（/producer）

与Broker之间建立一种简单的“担保”机制.

手动确认和单条消息确认需要手动的在客户端调用message.acknowledge()

消息重发机制RedeliveryPolicy 有几个属性如下：

RedeliveryPolicy redeliveryPolicy= new RedeliveryPolicy();
//是否在每次尝试重新发送失败后,增长这个等待时间
redeliveryPolicy.setUseExponentialBackOff(true);
//重发次数,默认为6次 这里设置为10次
redeliveryPolicy.setMaximumRedeliveries(10);
//重发时间间隔,默认为1秒
redeliveryPolicy.setInitialRedeliveryDelay(1);
//第一次失败后重新发送之前等待500毫秒,第二次失败再等待500 * 2毫秒,这里的2就是value
redeliveryPolicy.setBackOffMultiplier(2);
//是否避免消息碰撞
redeliveryPolicy.setUseCollisionAvoidance(false);
//设置重发最大拖延时间-1 表示没有拖延只有UseExponentialBackOff(true)为true时生效
redeliveryPolicy.setMaximumRedeliveryDelay(-1);
以下情况会导致消息重发:

1．在使用事务的Session中，调用rollback()方法；

2．在使用事务的Session中，调用commit()方法之前就关闭了Session;

3．在Session中使用CLIENT_ACKNOWLEDGE签收模式或者INDIVIDUAL_ACKNOWLEDGE模式，并且调用了recover()方法。

可以通过设置ActiveMQConnectionFactory来定制想要的再次传送策略。

需要注意的是：使用手动签收模式，如果客户端没有调用message.acknowledge()方法是不会立刻重发消息的，只有当前Coustomer重启时才能重新接受消息

spring boot 整合activemq 需要jar

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

因自己的需求，自己在默认的配置类中增加自己的配置
ActiveMQ4Config如下：

package com.zyc.activemq;

import javax.jms.Queue;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.core.JmsTemplate;

@EnableJms
@Configuration
public class ActiveMQ4Config {

@Bean
public Queue queue(){
return new ActiveMQQueue("queue1");
}

@Bean
public RedeliveryPolicy redeliveryPolicy(){
RedeliveryPolicy redeliveryPolicy= new RedeliveryPolicy();
//是否在每次尝试重新发送失败后,增长这个等待时间
redeliveryPolicy.setUseExponentialBackOff(true);
//重发次数,默认为6次 这里设置为10次
redeliveryPolicy.setMaximumRedeliveries(10);
//重发时间间隔,默认为1秒
redeliveryPolicy.setInitialRedeliveryDelay(1);
//第一次失败后重新发送之前等待500毫秒,第二次失败再等待500 * 2毫秒,这里的2就是value
redeliveryPolicy.setBackOffMultiplier(2);
//是否避免消息碰撞
redeliveryPolicy.setUseCollisionAvoidance(false);
//设置重发最大拖延时间-1 表示没有拖延只有UseExponentialBackOff(true)为true时生效
redeliveryPolicy.setMaximumRedeliveryDelay(-1);
return redeliveryPolicy;
}
@Bean
public ActiveMQConnectionFactory activeMQConnectionFactory (@Value("${activemq.url}")String url,RedeliveryPolicy redeliveryPolicy){
ActiveMQConnectionFactory activeMQConnectionFactory =
new ActiveMQConnectionFactory(
"admin",
"admin",
url);
activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
return activeMQConnectionFactory;
}

@Bean
public JmsTemplate jmsTemplate(ActiveMQConnectionFactory activeMQConnectionFactory,Queue queue){
JmsTemplate jmsTemplate=new JmsTemplate();
jmsTemplate.setDeliveryMode(2);//进行持久化配置 1表示非持久化，2表示持久化
jmsTemplate.setConnectionFactory(activeMQConnectionFactory);
jmsTemplate.setDefaultDestination(queue); //此处可不设置默认，在发送消息时也可设置队列
jmsTemplate.setSessionAcknowledgeMode(4);//客户端签收模式
return jmsTemplate;
}

//定义一个消息监听器连接工厂，这里定义的是点对点模式的监听器连接工厂
@Bean(name = "jmsQueueListener")
public DefaultJmsListenerContainerFactory jmsQueueListenerContainerFactory(ActiveMQConnectionFactory activeMQConnectionFactory) {
DefaultJmsListenerContainerFactory factory =
new DefaultJmsListenerContainerFactory();
factory.setConnectionFactory(activeMQConnectionFactory);
//设置连接数
factory.setConcurrency("1-10");
//重连间隔时间
factory.setRecoveryInterval(1000L);
factory.setSessionAcknowledgeMode(4);
return factory;
}

} 消费者如下：使用异步监听（使用监听器形式）

package com.zyc.activemq.consumer;

import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class Consumer {

private final static Logger logger = LoggerFactory
.getLogger(Consumer.class);

@JmsListener(destination = "queue1", containerFactory = "jmsQueueListener")
public void receiveQueue(final TextMessage text, Session session)
throws JMSException {
try {
logger.debug("Consumer收到的报文为:" + text.getText());
text.acknowledge();// 使用手动签收模式，需要手动的调用，如果不在catch中调用session.recover()消息只会在重启服务后重发
} catch (Exception e) {
session.recover();// 此不可省略 重发信息使用
}
}
}

生产者如下：
package com.zyc.activemq.producer;

import javax.jms.Destination;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsTemplate;
impo
4000
rt org.springframework.stereotype.Component;

@Component
public class Producer {

@Autowired
private JmsTemplate jmsTemplate;
/**
* 发送消息，estination是发送到的队列，message是待发送的消息
* @param destination
* @param message
*/
public void sendMessage(Destination destination, final String message) {
System.out.println(jmsTemplate.getDeliveryMode());
jmsTemplate.convertAndSend(destination, message);
}
/**
* 发送消息，message是待发送的消息
* @param message
*/
public void sendMessage(final String message) {
System.out.println(jmsTemplate.getDeliveryMode());
jmsTemplate.convertAndSend("queue1",message);
}

}

application.properties配置文件如下：

spring.datasource.url=jdbc:mysql://localhost:3306/mydb
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

activemq.url=failover:(tcp://127.0.0.1:61616)测试如下：如果不知道springboot junit 测试可参考springboot junit测试
package com.zyc;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit4.SpringRunner;
import com.zyc.activemq.producer.Producer;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = App.class)
public class ApplicationTest {

@Autowired
private Producer producer;
@Test
public void testActivemq(){
producer.sendMessage("look this is a message==zycc==");
while(true){}
}
}


本文参考了
http://shift-alt-ctrl.iteye.com/blog/2020182

http://blog.csdn.net/varyall/article/details/49907995