1.消息事务

2.消息时长，确认机制 

  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();
        }
 
    }

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

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();
        }
    }

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

<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>

    这个表用于记录哪个Broker是当前的Master Broker。

>> AMQ方式

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

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

    Data logs：