消息中间件的引用场景

异步处理、应用解耦、流量削峰

JMS消息模型：

点对点模型（point to point）：即生产者和消费者之间的消息来往；

发布/订阅模型（Pub/Sub）：包含三个角色：主题（Topic），发布者（publisher），订阅者（subscriber），多个发布者将消息发送到topic，系统将这些消息投递到订阅此Topic的订阅者；

点对点模型的特点：

a).每个消息只有一个消费者（即一旦被消费，消息就不再消息队列中）；

b).发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送消息之后，不管接收者有没有在运行，它不会影响消息被发送到队列；

c).接收者在成功接收消息之后需向队列应带成功；

发布/订阅模型的特点：

a).每个消息可以有多个消费者；

b).发布者和订阅者之间有时间上的依赖（消费者先订阅主题，发布者再来发送消息）；

c).订阅者必修保持运行的状态，才能接收发布者发布的消息；

核心API：

       a).Destination：表示消息所走通道的名称；

       b).ConnectionFactory：用于创建连接对象；

      c).Connection：连接接口，用来创建session；

       d).Session：会话接口，这是一个重要的接口，消息发送者、消息接收者以及消息对象本身，都是通过这个会话创建，创建destination；

       e).MessageConsumer：消息的消费者；

       f).MessageProducer：消息的生产者；

使用API创建生产者

创建连接工厂

创建连接

打开链接

创建session会话（两个参数：第一个是是否开启事务，第二个消息的确认机制）

创建destination目的地

创建消息生产者

创建消息

发送消息

释放资源

使用API创建消费者

创建连接工厂

创建连接

打开链接

创建session会话（两个参数：第一个是是否开启事务，第二个消息的确认机制）

创建distination目的地

创建消息消费者

获取消息（使用receive方法）/设置消息监听器来接收消息（不要释放资源）

JMS消息组成格式

结构：

JMS Provider（消息中间件/消息服务器/消息提供者）；

JMS Producer（消息生产者）；

JMS Consumer（消息消费者）；

JMS Message（消息----重要部分）

JMS Message组成

JMS Message消息由三部分组成：

消息头

只有JMSCorrelationID、JMSReplyTo、JMSType可以由开发者设置的

消息体

消息体，JMS API定义了5种消息体格式，也叫消息类型，可以使用不同形式发送接收数据，并可以兼容现有的消息格式。包括：

TextMessage（字符串）

MapMessage（键值对）

BytesMessage（字节的数据流）

ObjectMessage（序列化的对象）

StreamMessage（java原始值的数据流，可以传递字符串，整型等不同类型的数据，数据生产与消费顺序要一致）

注意：ActiveMQ5.12后，为了安全考虑，ActiveMQ默认不接受自定义的序列化对象，需要将自定义的对象加入到受信任的列表

消息属性

消息属性，包含以下三种类型的属性：

1：应用程序自定义设置和添加的属性，比如：Message.setStringProperty("hello", "my name is wangsaichao!");

2：JMS定义的属性：使用“JMSX”作为属性名的前缀，connection.getMetaData().getJMSXPropertyNames()， 方法返回所有连接支持的JMSX 属性的名字。

3：JMS供应商特定的属性：JMS定义的属性如下：

1：JMSXUserID：发送消息的用户标识，发送时提供商设置

2：JMSXAppID：发送消息的应用标识，发送时提供商设置

3：JMSXDeliveryCount：转发消息重试次数,第一次是1，第二次是2，… ，发送时提供商设置

4：JMSXGroupID：消息所在消息组的标识，由客户端设置

5：JMSXGroupSeq：组内消息的序号第一个消息是1，第二个是2，…，由客户端设置

6：JMSXProducerTXID ：产生消息的事务的事务标识，发送时提供商设置

7：JMSXConsumerTXID ：消费消息的事务的事务标识，接收时提供商设置

8：JMSXRcvTimestamp ：JMS 转发消息到消费者的时间，接收时提供商设置

9：JMSXState：假定存在一个消息仓库，它存储了每个消息的单独拷贝，且这些消息从原始消息被发送时开始。每个拷贝的状态有：1（等待），2（准备），3（到期）或4（保留）。由于状态与生产者和消费者无关，所以它不是由它们来提供。它只和在仓库中查找消息相关，因此JMS没有提供这种API。由提供商设置

ActiveMQ的高级特性

ActiveMQ消息持久化

消息持久化是保证消息不丢失的重要方式。

ActiveMQ提供了一下三种的消息存储方式

1. Memory消息存储-基于内存的消息存储；
2. 基于日志消息存储方式，KahaDB是ActiveMQ的默认日志存储方式，它提供了容量和恢复能力；
3. 基于JDBC的消息存储方式-数据存储于数据库中（例如：Mysql）；

ActiveMQ持久化机制流程：默认是基于日志消息存储方式

ActiveMQ基于JDBC消息存储（springBoot）

a).修改application.yml文件：

spring:
activemq:
broker-url: tcp://192.168.1.129:61616
user: admin
password: admin
  jms:
    template:
delivery-mode:

b).修改ActiveMQ/conf下的activemq.xml文件：

<bean name="mysql-ds" class="com.alibaba.druid.pool.DruidDataSource" destory-method="close">
<property name="driverClassName" valuue="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://192.168.1.49:3306/test" />
<property name="username" value="root"/>
<property name="password" value="root"/>

</bean>

<persistenceAdapter>
<jdbcPersistenceAdapter dataSource="#mysql-ds"/>

</persistenceAdapter>

c).拷贝mysql及druid数据源的jar包到activemq/lib目录下

d).重启activemq

消息事务

消息事务，是保证消息传递原子性的一个重要特征，和JDBC的事务特征类似；

一个事务性发送，其中一组（比如：5条消息为一组）消息要么能够全部保证到达服务器，要么都不到达服务器；

生产者、消费者与消息服务器直接都支持事务性；

ActiveMQ的事务主要偏向在生产者的应用；

ActiveMQ消息事务流程图：

实现方式（使用SpringBoot）：使用原生JMS，使用spring的JmsTransactionManager

方式一：使用原生JMS

生产者：

       @Test

public void ptpProducerWithTx() {Session session = null;try {//获取连接工厂
ConnectionFactory connectionFactory = jmsMessagingTemplate.getConnectionFactory();//创建连接
Connection connection = connectionFactory.createConnection();//打开连接
connection.start();/*** 参数一： 是否开启消息事务
* 参数二：消息的确认机制
* Session.AUTO_ACKNOWLEDGE 自动确认
* Session.CLIENT_ACKNOWLEDGE* Session.DUPS_OK_ACKNOWLEDGE* Session.SESSION_TRANSACTED*/session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);//创建目的地
Destination destination = session.createQueue(name);//创建生产者
MessageProducer producer = session.createProducer(destination);for (int i = 1; i <= 10; i++) {//异常模拟
//if (i == 4) {//    int a = 10 / 0;//}//创建消息
TextMessage textMessage = session.createTextMessage("springboot-activemq-with-tx-message" + i);//发送消息
producer.send(destination, textMessage);}//注意：一旦开启事务发送，那么就必修使用commint方法进行事务的提交，否则消息无法到达MQ服务器
session.commit();System.out.println("消息发送成功");} catch (Exception e) {e.printStackTrace();//事务的回滚
try {session.rollback();} catch (JMSException ex) {ex.printStackTrace();}}}

消费者：

@JmsListener(destination = "${activemq.name}")

public void receiveMessage(Message message, Session session) {if (message instanceof TextMessage) {TextMessage objectMessage = (TextMessage) message;try {String object = objectMessage.getText();System.out.println(name + " 接收消息：" + object);//模拟异常
// int a = 10/0;//提交事务
session.commit();} catch (JMSException e) {e.printStackTrace();//消息回滚
try {session.rollback();} catch (JMSException ex) {ex.printStackTrace();}}}}

方式二：使用spring的JmsTransactionManager

生产者：

配置ActiveMQ的事务管理器

@Configurationpublic class ActiveMQConfig {/*** 添加ActiveMQ事务管理器配置
*/@Beanpublic PlatformTransactionManager createPlatformTransactionManager(ConnectionFactory connectionFactory) {return new JmsTransactionManager(connectionFactory);}}

模拟消息发送业务

@Servicepublic class MessageService {@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;@Value("${activemq.name}")private String name;@Transactional //对消息发送加入事务管理（同事也对JDBC数据库的事务生效）
public void sendMessage() {for (int i = 1; i <= 10; i++) {//异常模拟
//if (i == 4) {//    int a = 10 / 0;//}jmsMessagingTemplate.convertAndSend(name, "springboot-activemq-no-tx-message" + i);}}}

发送消息：

@Autowiredprivate MessageService messageService;/*** 加入事务的消息发送--方案二：spring的JmsTransactionManager功能（JMS事务管理器）
*/@Testpublic void ptpProducerWithTx2() {messageService.sendMessage();}

消费者：同方案一的消费者

消息确认机制

JMS消息只有在被确认之后，才认为已经被成功地消费了。消息的成功消费通常包含三个阶段：客户接受消息、客户处理消息和消息被确认。在事务性会话中，当一个事务被提交的时候，确认自动发生。在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement mode）。该参数有以下三个可选值（只在非事务下生效，事务下只有自动确认 ）：

注意：消息确认机制与事务机制是冲突的，只能选其中一种，所以演示消息确认前，先关闭事务

Client端指定了ACK_MODE,但是在Client与broker在交换ACK指令的时候,还需要告知ACK_TYPE,ACK_TYPE表示此确认指令的类型，不同的ACK_TYPE将传递着消息的状态，broker可以根据不同的ACK_TYPE对消息进行不同的操作。

ActiveMQ中定义了如下几种ACK_TYPE(参看MessageAck类):

DELIVERED_ACK_TYPE = 0    消息"已接收"，但尚未处理结束；

STANDARD_ACK_TYPE = 2    "标准"类型,通常表示为消息"处理成功"，broker端可以删除消息了；

POSION_ACK_TYPE = 1    消息"错误",通常表示"抛弃"此消息，比如消息重发多次后，都无法正确处理时，消息将会被删除或者DLQ(死信队列)；

REDELIVERED_ACK_TYPE = 3    消息需"重发"，比如consumer处理消息时抛出了异常，broker稍后会重新发送此消息；

INDIVIDUAL_ACK_TYPE = 4    表示只确认"单条消息",无论在任何ACK_MODE下；

UNMATCHED_ACK_TYPE = 5    BROKER间转发消息时,接收端"拒绝"消息；

消息投递方式

消息投递方式包含三种方式：异步投递、延迟投递、定时投递

1、异步投递 VS 同步投递

同步发送：消息生产者使用持久（Persistent）传递模式发送消息的时候，Producer.send()方法会被阻塞，直到broker发送一个确认消息给生产者（ProducerAck），这个确认消息按时broker已经成功接收到消息并把消息保存到二级存储中。

异步发送：如果应用程序能够容忍一些消息的丢失，那么可以使用异步发送。异步发送不会在收到broker的确认之前一直阻塞Producer.send()方法。

想要使用异步，在brokerUrl中增加jms.alwaysSyncSend=false&jms.userAsyncSend=true属性

a).如果设置了alwaysSyncSend=true（同步）系统会忽略userAsyncSend设置的值都采用同步；

b).当alwaysSyncSend=false（异步）时，“NON_PERSISTENT”（非持久化）、事务中的消息将使用“异步发送”；

c).当alwaysSyncSend=false（异步）时，如果指定userAsyncSend=true（异步），“PERSISTENT”（持久化）类型的消息使用异步发送。如果userAsyncSend=false（同步），“PERSISTENT”（持久化）类型的消息使用同步发送；

总结：默认情况（alwaysSyncSend=false，userAsyncSend=false），非持久化消息、事务内的消息均采用异步发送；对于持久化消息采用同步发送！

没有使用事务并且正在发送持久性消息，则每个发送都是同步并阻塞

配置异步投递方式

方式一（原生JMS）：

a).在连接上配置

new ActiveMQConnectionFactory("tcp://192.168.1.128:61616?jms.userAsyncSend=true");

b).通过ConnectionFactory

((ActiveMQConnectionFactory)connectionFactory).setUseAsyncSend(true);

c).通过connection

((ActiveMQConnection) connection).setUseAsyncSend(true);

注意：如果是Spring或者SpringBoot项目，通过修改JmsTemplate的默认参数实现异步或同步投递

方式二：SpringBoot的配置

/*** 配置用于异步发送的非持久化JmsTemplate*/@Autowired@Beanpublic JmsTemplate asyncJmsTemplate(ConnectionFactory connectionFactory){JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);jmsTemplate.setExplicitQosEnabled(true);//deliveryMode, priority, timeToLive 的开关
jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);//非持久化
return jmsTemplate;}/*** 配置用于同步发送的非持久化JmsTemplate* 默认为同步
*/@Autowired@Beanpublic JmsTemplate syncJmsTemplate(ConnectionFactory connectionFactory){JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);return jmsTemplate;}

异步投递如何确认发送成功？

异步发送丢失场景：生产者设置了useAsyncSend=true（异步），使用producer.send(message)持续发送消息。由于消息不阻塞，生产者会认为所有send的消息都被成功发送到MQ了。如果此时MQ突然宕机，生产者端内存中尚未发送至MQ的消息就会丢失。

这时，可以给异步投递方法接收回调函数，已确认消息是否发送成功。

@Value("${activemq.name}")private String name;@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;/*** 测试异步/同步投递
*/@Testpublic void asyncAndSyncSend() {Session session = null;try {//获取连接工厂
ActiveMQConnectionFactory connectionFactory = (ActiveMQConnectionFactory) jmsMessagingTemplate.getConnectionFactory();
// 设置消息发送模式是AsyncSend模式，异步模式
connectionFactory.setUseAsyncSend(true);//创建连接
Connection connection = connectionFactory.createConnection();//打开连接
connection.start();/*** 参数一： 是否开启消息事务
* 参数二：消息的确认机制
* Session.AUTO_ACKNOWLEDGE 自动确认
* Session.CLIENT_ACKNOWLEDGE* Session.DUPS_OK_ACKNOWLEDGE* Session.SESSION_TRANSACTED*/session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);//创建目的地
Destination destination = session.createQueue(name);//创建生产者
ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(destination);        //非持久化

        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

for (int i = 1; i <= 10; i++) {
//创建消息
            TextMessage textMessage = session.createTextMessage("springboot-activemq-with-async-message" + i);
//设置消息唯一的ID
String msgId = UUID.randomUUID().toString().replaceAll("-", "");
//这一步没有效果，因为即使开发者设置值，但是消息提供者会自己生成一个id，该id使用来记录本地发送失败的数据的
            textMessage.setJMSMessageID(msgId);
//发送消息
            producer.send(textMessage, new AsyncCallback() {
@Override
public void onSuccess() {
//使用msgId标识来进行消息发送成功的处理
                    System.out.println("msgId:" + msgId);
}
@Override
public void onException(JMSException exception) {
//使用msgId标识来进行消息发送失败的处理
                    System.out.println("msgId:" + msgId + "发送失败");
exception.printStackTrace();
}
});
}
//注意：一旦开启事务发送，那么就必修使用commint方法进行事务的提交，否则消息无法到达MQ服务器
        session.commit();
System.out.println("消息发送成功");
} catch (Exception e) {
e.printStackTrace();
//事务的回滚
        try {
session.rollback();
} catch (JMSException ex) {
ex.printStackTrace();
}
}
}

2、延迟投递

生产者提供两个发送消息的方法，一个是即使发送消息，一个是延时发送消息。

步骤如下：

1、修改activemq.xml

注意：修改activemq/conf目录下的activemq.xml文件中的broker标签，添加schedulerSupport=”true”配置 开启延时投递，重启activemq；

<broker xmlns="http://activemq.apache.org/schema/core" 
brokerName="localhost" dataDirectory="${activemq.data}"
schedulerSupport="true" />

2、在代码中设置延时时长

@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;

/**
* 延时投递
 */
@Test
public void lazySendMessage() {
Session session = null;
try {
//获取连接工厂
        ConnectionFactory connectionFactory = jmsMessagingTemplate.getConnectionFactory();
//创建连接
        Connection connection = connectionFactory.createConnection();
//打开连接
        connection.start();
/**
* 参数一： 是否开启消息事务
         * 参数二：消息的确认机制
         * Session.AUTO_ACKNOWLEDGE 自动确认
         * Session.CLIENT_ACKNOWLEDGE
* Session.DUPS_OK_ACKNOWLEDGE
* Session.SESSION_TRANSACTED
*/
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
//创建目的地
        Destination destination = session.createQueue(name);
//创建生产者
        ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);//非持久化
        //创建消息
        TextMessage textMessage = session.createTextMessage("springboot-activemq-with-tx-message");
//设置延时时长（延时10s）
        textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 10000);
//发送消息
        producer.send(destination, textMessage);
//注意：一旦开启事务发送，那么就必修使用commint方法进行事务的提交，否则消息无法到达MQ服务器
        session.commit();
System.out.println("消息发送成功");
} catch (Exception e) {
e.printStackTrace();
//事务的回滚
        try {
session.rollback();
} catch (JMSException ex) {
ex.printStackTrace();
}
}
}

ScheduledMessage投递的四个属性

定时投递

步骤如下：

1、启动类添加定时注解

@SpringBootApplication
@EnableScheduling //开启定时任务
public class ProducerApplication {
public static void main(String[] args) {
SpringApplication.run(ProducerApplication.class, args);
}
}

2、在生产者添加@Scheduled设置定时

/**
* 定时任务消息发送
 */
@Component
public class Producer {

@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;

@Value("${activemq.name}")
private String name;

/**
* 定时发送消息
     */
@Scheduled(fixedDelay = 3000)
public void sendMessage() {
jmsMessagingTemplate.convertAndSend(name, "springboot-activemq-scheduled");
}

}

死信队列

DLQ-死信队列(Dead Letter Queue)用来保存处理失败或者过期的消息。

出现以下情况时，消息会被redelivered（重新交付）：

A transacted session is used and rollback() is called.

A transacted session is closed before commit() is called.

A session is using CLIENT_ACKNOWLEDGE and Session.recover() is called.

A client connection times out (perhaps the code being executed takes longer than the configured time-out period).

（

1.事务会话中，当还未进行session.commit()时，进行session.rollback()，那么所有还没commit的消息都会进行重发。

2.所有未ack的消息，在没有commint()之前进行session.closed()关闭事务，那么所有还没ack的消息broker端都会进行重发，而且是马上重发。

3.使用客户端手动确认的方式时（非事务会话），还未进行确认并且执行Session.recover()，那么所有还没acknowledge的消息都会进行重发。

4.消息被消费者拉取之后，超时没有响应ack，消息会被broker重发。

）

当一个消息被redelivered超过maximumRedeliveries(缺省为6次)次数时，会给broker发送一个" poison_ack"，这个消息被认为是a poison pill，这时broker会将这个消息发送到DLQ，以便后续处理。

注意两点：

1、缺省持久消息过期，会被送到DLQ，非持久消息不会送到DLQ

2、缺省的死信队列是ActiveMQ.DLQ，如果没有特别指定，死信都会被发送到这个队列。

可以通过配置文件(activemq.xml)来调整死信发送策略，方案如下：

官网链接：http://activemq.apache.org/message-redelivery-and-dlq-handling

1. （方案一）为每个队列创建死信队列

修改activemq/conf目录下的activemq.xml文件：

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" schedulerSupport="true">

<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry queue=">" ><!--为每个队列建立死信队列 --><deadLetterStrategy>
<individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true"/>
</deadLetterStrategy>
</policyEntry>
<policyEntry topic=">" ><!--死信队列的默认配置 --><pendingMessageLimitStrategy>
<constantPendingMessageLimitStrategy limit="1000"/>
</pendingMessageLimitStrategy>
</policyEntry>
</policyEntries>
</policyMap>

</destinationPolicy>

2、（方案二）RedeliveryPolicy重发策略设置

在消费者方配置如下：

@Configurationpublic class ActiveMQConfig {/*** RedeliveryPolicy*/@Beanpublic RedeliveryPolicy redeliveryPolicy() {RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();//是否在每次尝试重新发送失败后，增长这个等待时间
redeliveryPolicy.setUseExponentialBackOff(true);//重发次数，默认为6次，这里设置为10次
redeliveryPolicy.setMaximumRedeliveries(10);//重发时间间隔，默认为1sredeliveryPolicy.setInitialRedeliveryDelay(2000);//第一次失败后重新发送之前等待500毫秒，第二次失败再等待500*2毫秒，第三次500*2*2，这里的2就是valueredeliveryPolicy.setBackOffMultiplier(2);//是否避免消息碰撞
redeliveryPolicy.setUseCollisionAvoidance(false);//设置重发最大拖延时间，-1表示没有拖延只有UseExponentialBackOff(true)为true时生效
//当重连时间间隔大的最大重连时间间隔时，以后每次重连时间间隔都为最大重连时间间隔。
redeliveryPolicy.setMaximumRedeliveryDelay(-1);return redeliveryPolicy;}@Beanpublic ActiveMQConnectionFactory activeMQConnectionFactory(@Value("${spring.activemq.broker-url}") String url, RedeliveryPolicy redeliveryPolicy) {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("admin", "admin", url);activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);return activeMQConnectionFactory;}/*** 添加ActiveMQ事务管理器配置
*/@Beanpublic PlatformTransactionManager createPlatformTransactionManager(ConnectionFactory connectionFactory) {return new JmsTransactionManager(connectionFactory);}/*** 消费者 消息确认机制配置
** @param connectionFactory* @return*/@Bean("jmsListenerContainerFactory")public DefaultJmsListenerContainerFactory defaultJmsListenerContainerFactory(ConnectionFactory connectionFactory, PlatformTransactionManager platformTransactionManager) {DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();factory.setConnectionFactory(connectionFactory);//连接工厂
factory.setTransactionManager(platformTransactionManager);//事务管理器
//是否关闭事务，开启事务
factory.setSessionTransacted(true);//修改消息确认机制
//自动确认
factory.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);return factory;}}

@Componentpublic class MyListener {@Value("${activemq.name}")private String name;private int count = 0;/*** 消费方事务的配置，使用session形参控制
** @param message* @param session*/@JmsListener(destination = "${activemq.name}",containerFactory = "jmsListenerContainerFactory")public void receiveMessage(Message message, Session session) {if (message instanceof TextMessage) {TextMessage objectMessage = (TextMessage) message;try {String object = objectMessage.getText();System.out.println("重复提交次数count:" + (++count));System.out.println(name + " 接收消息：" + object);//模拟异常
int a = 10 / 0;//提交事务
session.commit();} catch (JMSException e) {e.printStackTrace();//消息回滚
try {session.rollback();} catch (JMSException ex) {ex.printStackTrace();}}}}}

其他方案：https://www.cnblogs.com/rainwang/p/5146223.html

包括：

非持久消息保存到死信队列

<deadLetterStrategy>

<sharedDeadLetterStrategy  processNonPersistent="true"/>

</deadLetterStrategy>

过期消息不保存到死信队列

<deadLetterStrategy>

<sharedDeadLetterStrategy processExpired="false" />

</deadLetterStrategy>

持久消息不保存到死信队列

ActiveMQ企业经典面试问题

ActiveMQ宕机了怎么办？

1. ActiveMQ主从集群方案：Zookeeper集群+Replicate LevelDB（kahadb）+ActiveMQ集群

官网链接：http://activemq.apache.org/replicated-leveldb-store

（LevelDB存储已被弃用，不再受支持或建议使用。推荐的store是KahaDB）

如何防止消费者消费重复消息？

如果因为网络延迟等原因，MQ无法及时接收到消息方的应答，导致MQ重试。在重试过程中造成重复消费的问题。

解决思路：

1. 如果消费方是做数据库操作，那么可以报消息的ID作为表的唯一主键或者唯一约束，这样在重试的情况下，会触发冲突，从而避免数据出现脏数据。
2. 如果消费方不是数据库操作，那么可以借助第三方的应用，例如redis，来记录消费记录。每次消费被消费完成的时候，把当前消息的ID作为key存入redis，每次存入redis，每次消费前，先到redis查询有没有该消息的消费者记录。

如何防止消息丢失？

以下手段可以防止消息丢失：

1. 在消息生产者和消费者使用事务；
2. 在消费者采用手动消息确认（ACK）
3. 消息持久化，例如JDBC或kahadb

什么是死信队列？

DLQ-死信队列(Dead Letter Queue)用来保存处理失败或者过期的消息。