ActiveMQ的设置消息时长，事务，确认机制，持久化

时间 2019-11-11

原文原文链接

1.消息事务java

消息事务是在生产者producer到broker或broker到consumer过程当中同一个session中发生的，保证几条消息在发送过程当中的原子性。(Broker:消息队列核心，至关于一个控制中心，负责路由消息、保存订阅和链接、消息确认和控制事务)
在支持事务的session中，producer发送message时在message中带有transactionID。broker收到message后判断是否有transactionID，若是有就把message保存在transaction store中，等待commit或者rollback消息。mysql

消息生产者-异步发送
消息生产者使用持久（persistent）传递模式发送消息的时候，Producer.send() 方法会被阻塞，直到 broker 发送一个确认消息给生产者(ProducerAck)，这个确认消息暗示broker已经成功接收到消息并把消息保存到二级存储中。这个过程一般称为同步发送。
若是应用程序可以容忍一些消息的丢失，那么可使用异步发送。异步发送不会在受到 broker 的确认以前一直阻塞 Producer.send 方法。但有一个例外，当发送方法在一个事务上下文中时，被阻塞的是 commit 方法而不是 send 方法。commit 方法成功返回意味着全部的持久消息都以被写到二级存储中。
想要使用异步，在brokerURL中增长 jms.alwaysSyncSend=false&jms.useAsyncSend=true
若是设置了alwaysSyncSend=true系统将会忽略useAsyncSend设置的值都采用同步
     1) 当alwaysSyncSend=false时，“NON_PERSISTENT”(非持久化)、事务中的消息将使用“异步发送”
     2) 当alwaysSyncSend=false时，若是指定了useAsyncSend=true，“PERSISTENT”类型的消息使用异步发送。若是useAsyncSend=false，“PERSISTENT”类型的消息使用同步发送。
总结：默认状况(alwaysSyncSend=false,useAsyncSend=false)，非持久化消息、事务内的消息均采用异步发送；对于持久化消息采用同步发送。
jms.sendTimeout:发送超时时间，默认等于0，若是jms.sendTimeout>0将会忽略（alwaysSyncSend、useAsyncSend、消息是否持久化）全部的消息都是用同步发送！
即便使用异步发送，也能够经过producerWindowSize来控制发送端无节制的向broker发送消息
producerWindowSize:窗口尺寸，用来约束在异步发送时producer端容许积压的(还没有ACK)的消息的尺寸，且只对异步发送有意义。每次发送消息以后，都将会致使memoryUsage尺寸增长(+message.size)，当broker返回producerAck时，若是达到了producerWindowSize上限，即便是异步调用也会被阻塞，防止不停向broker发送消息。
经过jms.producerWindowSize=。。。来设置

2.消息时长，确认机制
消息消费者-消息确认
一、确认机制(ack_mod)
      AUTO_ACKNOWLEDGE = 1    自动确认
      CLIENT_ACKNOWLEDGE = 2    客户端手动确认
      DUPS_OK_ACKNOWLEDGE = 3    自动批量确认
      SESSION_TRANSACTED = 0    事务提交并确认
ACK_MODE描述了Consumer与broker确认消息的方式(时机),好比当消息被Consumer接收以后,Consumer将在什么时候确认消息。因此ack_mode描述的不是producer于broker之间的关系，而是customer于broker之间的关系。
对于broker而言，只有接收到ACK指令,才会认为消息被正确的接收或者处理成功了,经过ACK，能够在consumer与Broker之间创建一种简单的“担保”机制.
      session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
第一个参数:是否支持事务，若是为true，则会忽略第二个参数，自动被jms服务器设置为SESSION_TRANSACTED。sql

发布主题数据库

public class TopicPub {
        public static void main(String[] args) throws JMSException {
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            Connection connection = factory.createConnection();
            connection.start();
 
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            /**
             Session javax.jms.Connection.createSession(boolean transacted, int acknowledgeMode) throws JMSException
             1.transacted事务，事务成功commit,才会将消息发送到mom中
             2.acknowledgeMode消息确认机制
             1）、带事务的session
             若是session带有事务，而且事务成功提交，则消息被自动签收。若是事务回滚，则消息会被再次传送。
             消息事务是在生产者producer到broker或broker到consumer过程当中同一个session中发生的，
             保证几条消息在发送过程当中的原子性。
             在支持事务的session中，producer发送message时在message中带有transactionID。
             broker收到message后判断是否有transactionID，若是有就把message保存在transaction store中，
             等待commit或者rollback消息。
             2）、不带事务的session
             不带事务的session的签收方式，取决于session的配置。
             Activemq支持一下三種模式：
             Session.AUTO_ACKNOWLEDGE  消息自动签收
             Session.CLIENT_ACKNOWLEDGE  客戶端调用acknowledge方法手动签收
             Session.DUPS_OK_ACKNOWLEDGE 不是必须签收，消息可能会重复发送。在第二次从新传送消息的时候，消息
             头的JmsDelivered会被置为true标示当前消息已经传送过一次，客户端须要进行消息的重复处理控制。
             代码示例以下：
             session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
             textMsg.acknowledge();
             */
            Topic topic = session.createTopic("wm5920.topic");
 
            MessageProducer producer = session.createProducer(topic);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);//设置非持久化
            //producer.setTimeToLive(5000);//5秒后过时，这个对点对点模式有效
            TextMessage message = session.createTextMessage();
            message.setText("message_" + System.currentTimeMillis());
            producer.send(message);
            System.out.println("Sent message: " + message.getText());
            //带有事务得commit
            //session.commit();
            session.close();
            connection.stop();
            connection.close();
        }
 
    }

订阅主题，注：若是在发布主题前，没有订阅，是收不到消息的，这跟点对点的队列模式不一样apache

package com.activemq;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
 
 
    public class TopicSubs{
        public static void main(String[] args) throws JMSException {
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
            Connection connection = factory.createConnection();
            connection.setClientID("wm5920");
            connection.start();
 
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic("wm5920.topic");
 
            //持久订阅方式，不会漏掉信息
            TopicSubscriber subs=session.createDurableSubscriber(topic, "wm5920");
            subs.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage tm = (TextMessage) message;
                    try {
                        System.out.println("Received message: " + tm.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
 
            //非持久订阅方式
//        MessageConsumer consumer = session.createConsumer(topic);
//        consumer.setMessageListener(new MessageListener() {
//            public void onMessage(Message message) {
//                TextMessage tm = (TextMessage) message;
//                try {
//                    System.out.println("Received message: " + tm.getText());
//                } catch (JMSException e) {
//                    e.printStackTrace();
//                }
//            }
//        });
//        session.commit();
//      session.close();
//      connection.stop();
//      connection.close();
        }
    }

1.ActiveMQ的几种消息持久化机制
为了不意外宕机之后丢失信息，须要作到重启后能够恢复消息队列，消息系统通常都会采用持久化机制。
ActiveMQ的消息持久化机制有JDBC，AMQ，KahaDB和LevelDB，不管使用哪一种持久化方式，消息的存储逻辑都是一致的。
就是在发送者将消息发送出去后，消息中心首先将消息存储到本地数据文件、内存数据库或者远程数据库等，而后试图将消息发送给接收者，发送成功则将消息从存储中删除，失败则继续尝试。
消息中心启动之后首先要检查指定的存储位置，若是有未发送成功的消息，则须要把消息发送出去。
>> JDBC持久化方式
使用JDBC持久化方式，数据库会建立3个表：activemq_msgs，activemq_acks和activemq_lock。
activemq_msgs用于存储消息，Queue和Topic都存储在这个表中。
（1）配置方式
配置持久化的方式，都是修改安装目录下conf/acticvemq.xml文件，
首先定义一个mysql-ds的MySQL数据源，而后在persistenceAdapter节点中配置jdbcPersistenceAdapter而且引用刚才定义的数据源。缓存

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

dataSource指定持久化数据库的bean，createTablesOnStartup是否在启动的时候建立数据表，默认值是true，这样每次启动都会去建立数据表了，通常是第一次启动的时候设置为true，以后改为false。
使用MySQL配置JDBC持久化:
服务器

<beans>
        <broker brokerName="test-broker" persistent="true" xmlns="http://activemq.apache.org/schema/core">
            <persistenceAdapter>
                <jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="false"/>
            </persistenceAdapter>
        </broker>
        <bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
            <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
            <property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>
            <property name="username" value="activemq"/>
            <property name="password" value="activemq"/>
            <property name="maxActive" value="200"/>
            <property name="poolPreparedStatements" value="true"/>
        </bean>
    </beans>

（2）数据库表信息
activemq_msgs用于存储消息，Queue和Topic都存储在这个表中：
ID：自增的数据库主键
CONTAINER：消息的Destination
MSGID_PROD：消息发送者客户端的主键
MSG_SEQ：是发送消息的顺序，MSGID_PROD+MSG_SEQ能够组成JMS的MessageID
EXPIRATION：消息的过时时间，存储的是从1970-01-01到如今的毫秒数
MSG：消息本体的Java序列化对象的二进制数据
PRIORITY：优先级，从0-9，数值越大优先级越高session

activemq_acks用于存储订阅关系。若是是持久化Topic，订阅者和服务器的订阅关系在这个表保存：
主要的数据库字段以下：
    CONTAINER：消息的Destination
    SUB_DEST：若是是使用Static集群，这个字段会有集群其余系统的信息
    CLIENT_ID：每一个订阅者都必须有一个惟一的客户端ID用以区分
    SUB_NAME：订阅者名称
    SELECTOR：选择器，能够选择只消费知足条件的消息。条件能够用自定义属性实现，可支持多属性AND和OR操做
    LAST_ACKED_ID：记录消费过的消息的ID。异步

    表activemq_lock在集群环境中才有用，只有一个Broker能够得到消息，称为Master Broker，
    其余的只能做为备份等待Master Broker不可用，才可能成为下一个Master Broker。
    这个表用于记录哪一个Broker是当前的Master Broker。tcp

>> AMQ方式

性能高于JDBC，写入消息时，会将消息写入日志文件，因为是顺序追加写，性能很高。为了提高性能，建立消息主键索引，而且提供缓存机制，进一步提高性能。每一个日志文件的大小都是有限制的（默认32m，可自行配置）。
当超过这个大小，系统会从新创建一个文件。当全部的消息都消费完成，系统会删除这个文件或者归档（取决于配置）。
主要的缺点是AMQ Message会为每个Destination建立一个索引，若是使用了大量的Queue，索引文件的大小会占用不少磁盘空间。
并且因为索引巨大，一旦Broker崩溃，重建索引的速度会很是慢。
配置片断以下：

<persistenceAdapter>
         <amqPersistenceAdapter directory="${activemq.data}/activemq-data" maxFileLength="32mb"/>
    </persistenceAdapter>

虽然AMQ性能略高于下面的Kaha DB方式，可是因为其重建索引时间过长，并且索引文件占用磁盘空间过大，因此已经不推荐使用。
>> KahaDB方式
    KahaDB是从ActiveMQ 5.4开始默认的持久化插件，也是咱们项目如今使用的持久化方式。
    KahaDb恢复时间远远小于其前身AMQ而且使用更少的数据文件，因此能够彻底代替AMQ。
    kahaDB的持久化机制一样是基于日志文件，索引和缓存。
    配置方式：

<persistenceAdapter>
        <kahaDB directory="${activemq.data}/activemq-data" journalMaxFileLength="16mb"/>
    </persistenceAdapter>

directory : 指定持久化消息的存储目录
journalMaxFileLength : 指定保存消息的日志文件大小，具体根据你的实际应用配置　

（1）KahaDB主要特性
    一、日志形式存储消息；
    二、消息索引以B-Tree结构存储，能够快速更新；
    三、彻底支持JMS事务；
    四、支持多种恢复机制；

（2）KahaDB的结构
    消息存储在基于文件的数据日志中。若是消息发送成功，变标记为可删除的。系统会周期性的清除或者归档日志文件。
    消息文件的位置索引存储在内存中，这样能快速定位到。按期将内存中的消息索引保存到metadata store中，避免大量消息未发送时，消息索引占用过多内存空间。
    Data logs：
    Data logs用于存储消息日志，消息的所有内容都在Data logs中。
    同AMQ同样，一个Data logs文件大小超过规定的最大值，会新建一个文件。一样是文件尾部追加，写入性能很快。
    每一个消息在Data logs中有计数引用，因此当一个文件里全部的消息都不须要了，系统会自动删除文件或放入归档文件夹。

    Metadata cache ：
    缓存用于存放在线消费者的消息。若是消费者已经快速的消费完成，那么这些消息就不须要再写入磁盘了。
    Btree索引会根据MessageID建立索引，用于快速的查找消息。这个索引一样维护持久化订阅者与Destination的关系，以及每一个消费者消费消息的指针。

    Metadata store
    在db.data文件中保存消息日志中消息的元数据，也是以B-Tree结构存储的，定时从Metadata cache更新数据。Metadata store中也会备份一些在消息日志中存在的信息，这样可让Broker实例快速启动。
    即使metadata store文件被破坏或者误删除了。broker能够读取Data logs恢复过来，只是速度会相对较慢些。
    >>LevelDB方式

从ActiveMQ 5.6版本以后，又推出了LevelDB的持久化引擎。目前默认的持久化方式仍然是KahaDB，不过LevelDB持久化性能高于KahaDB，多是之后的趋势。在ActiveMQ 5.9版本提供了基于LevelDB和Zookeeper的数据复制方式，用于Master-slave方式的首选数据复制方案。

ActiveMQ的设置消息时长，事务，确认机制 ，持久化

ActiveMQ的设置消息时长，事务，确认机制，持久化