ActiveMQ学习笔记

ゝ一纸荒年。 2022-06-11 01:39 168阅读 0赞

**AcitveMQ**是作为一种消息存储和分发组件，涉及到client与broker端数据交互的方方面面，它不仅要担保消息的存储安全性，还要提供额外的手段来确保消息的分发是可靠的。

**ActiveMQ中有两种类型的消息模型**

1.  **点对点**（point-to-point，简称PTP）Queue消息传递模型
2.  **发布/订阅**（publish/subscribe，简称pub/sub）Topic消息传递模型

<table> 
 <thead> 
  <tr> 
   <th align="left">类型</th> 
   <th align="left">Topic</th> 
   <th align="left">Queue</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">概要</td> 
   <td align="left">Publish Subscribe messaging 发布订阅消息</td> 
   <td align="left">Point-to-Point 点对点</td> 
  </tr> 
  <tr> 
   <td align="left">有无状态</td> 
   <td align="left">topic数据默认不落地，是无状态的。</td> 
   <td align="left">Queue数据默认会在mq服务器上以文件形式保存，比如Active MQ一般保存在$AMQ_HOME\data\kr-store\data下面。也可以配置成DB存储。</td> 
  </tr> 
  <tr> 
   <td align="left">完整性保障</td> 
   <td align="left">并不保证publisher发布的每条数据，Subscriber都能接受到。</td> 
   <td align="left">Queue保证每条数据都能被receiver接收。</td> 
  </tr> 
  <tr> 
   <td align="left">消息是否会丢失</td> 
   <td align="left">一般来说publisher发布消息到某一个topic时，只有正在监听该topic地址的sub能够接收到消息；如果没有sub在监听，该topic就丢失了。</td> 
   <td align="left">Sender发送消息到目标Queue，receiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取，也不会丢失。</td> 
  </tr> 
  <tr> 
   <td align="left">消息发布接收策略</td> 
   <td align="left">一对多的消息发布接收策略，监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器</td> 
   <td align="left">一对一的消息发布接收策略，一个sender发送的消息，只能有一个receiver接收。receiver接收完后，通知mq服务器已接收，mq服务器对queue里的消息采取删除或其他操作。</td> 
  </tr> 
 </tbody> 
</table>

## 消息确认模式 ##

客户端成功接收一条消息的标志是这条消息被签收。成功接收一条消息一般包括如  
下三个阶段：  
1. 客户端接收消息；  
2. 客户端处理消息；  
3. 消息被确认。确认可以由ActiveMQ 发起，也可以由客户端发起，取决于Session 确认模式的设置。  
在带事务的 Session 中，确认自动发生在事务提交时。如果事务回滚，所有已经接收的消息将会被再次传送。  
在不带事务的Session 中，一条消息何时和如何被确认取决于Session 的设置。

<table> 
 <thead> 
  <tr> 
   <th align="left">模式</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">Session.AUTO_ACKNOWLEDGE</td> 
   <td align="left">默认的消息确认模式， session 将按照MessageConsumer.receive（） 返回成功或者用于办法回调MessageListener.onMessage（） 返回成功来主动回执消息确认。</td> 
  </tr> 
  <tr> 
   <td align="left">Session.CLIENT_ACKNOWLEDGE</td> 
   <td align="left">在这种模式下，客户端应用法度代码中显式调用Message.acknowledge（）办法，以确认该消息。</td> 
  </tr> 
  <tr> 
   <td align="left">Session.DUPS_OK_ACKNOWLEDGE</td> 
   <td align="left">该模式，jms session可以或许主动的回执消息确认，然则它是一种怠惰的消息确认模式，一旦应用体系故障，一些消息可能会被处理惩罚然则没有回执，一旦体系重启，则会呈现消息的重发题目。<br> 这是一种最快的消息确认模式，然则，花费者必须处理惩罚反复的消息处理惩罚题目。（例如：如何侦测和丢弃重发的消息的处理惩罚逻辑）</td> 
  </tr> 
  <tr> 
   <td align="left">Session.SESSION_TRANSACTED</td> 
   <td align="left">当应用事务时，会话会隐含SESSION_TRANSACTED模式。事务提交的反响，然后相当于消息的确认。<br> 当应用JMS事务来发送一组消息（多个消息），交易模式是很是有效的。但避免应用事务发送一个消息，因为这会带来额外的开销提交或回滚事务。</td> 
  </tr> 
  <tr> 
   <td align="left">ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE</td> 
   <td align="left">它不是一个标准的消息确认模式，很类&amp;＃20284;与CLIENT_ACKNOWLEDGE，只可惜它的消息确认是一种办法回调。在完成消息处理惩罚后不会去向理惩罚消息确认的。</td> 
  </tr> 
 </tbody> 
</table>

ActiveMQ支持两种事务

>  *  JMS 事务。使用Session的commint()和rollback()即可实现。（类似于JDBC Connection对象）
>  *  XA 事务。XASession,XAResource通过与消息中介（Message Broker）通信来实现。

**事务原理**  
在支持事务的session中，producer发送message时在message中带有transactionID。broker收到message后判断是否有transactionID，如果有就把message保存在transaction store中，等待commit或者rollback消息。所以ActiveMQ的事务是针对broker而不是producer,不管session是否commit,broker都会收到message。

如果producer发送模式选择了persistent，那么message过期后会进入死亡队列。在message进入死亡队列之前，ActiveMQ会删除message中的transaction ID，这样过期的message就不在事务中了，不会保存在transaction store中，会直接进入死亡队列。

### 点对点的发送端 ###

import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class PTPSend {
        //连接账号
        private String userName = "";
        //连接密码
        private String password = "";
        //连接地址
        private String brokerURL = "tcp://192.168.0.130:61616";
        //connection的工厂
        private ConnectionFactory factory;
        //连接对象
        private Connection connection;
        //一个操作会话
        private Session session;
        //目的地，其实就是连接到哪个队列，如果是点对点，那么它的实现是Queue，如果是订阅模式，那它的实现是Topic
        private Destination destination;
        //生产者，就是产生数据的对象
        private MessageProducer producer;
        
        public static void main(String[] args) {
            PTPSend send = new PTPSend();
            send.start();
        }
        
        public void start(){
            try {
                //根据用户名，密码，url创建一个连接工厂
                factory = new ActiveMQConnectionFactory(userName, password, brokerURL);
                //从工厂中获取一个连接
                connection = factory.createConnection();
                //测试过这个步骤不写也是可以的，但是网上的各个文档都写了
                connection.start();
                //创建一个session
                //第一个参数:是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
                //第二个参数为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
                //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
                //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
                //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                //创建一个到达的目的地，其实想一下就知道了，activemq不可能同时只能跑一个队列吧，这里就是连接了一个名为"text-msg"的队列，这个会话将会到这个队列，当然，如果这个队列不存在，将会被创建
                destination = session.createQueue("text-msg");
                //从session中，获取一个消息生产者
                producer = session.createProducer(destination);
                //设置生产者的模式，有两种可选
                //DeliveryMode.PERSISTENT 当activemq关闭的时候，队列数据将会被保存
                //DeliveryMode.NON_PERSISTENT 当activemq关闭的时候，队列里面的数据将会被清空
                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
                
                //创建一条消息，当然，消息的类型有很多，如文字，字节，对象等,可以通过session.create..方法来创建出来
                TextMessage textMsg = session.createTextMessage("呵呵");
                for(int i = 0 ; i < 100 ; i ++){
                    //发送一条消息
                    producer.send(textMsg);
                }
                
                System.out.println("发送消息成功");
                //即便生产者的对象关闭了，程序还在运行哦
                producer.close();
                
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

### 点对点的接收端 ###

import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class PTPReceive {
        //连接账号
        private String userName = "";
        //连接密码
        private String password = "";
        //连接地址
        private String brokerURL = "tcp://192.168.0.130:61616";
        //connection的工厂
        private ConnectionFactory factory;
        //连接对象
        private Connection connection;
        //一个操作会话
        private Session session;
        //目的地，其实就是连接到哪个队列，如果是点对点，那么它的实现是Queue，如果是订阅模式，那它的实现是Topic
        private Destination destination;
        //消费者，就是接收数据的对象
        private MessageConsumer consumer;
        public static void main(String[] args) {
            PTPReceive receive = new PTPReceive();
            receive.start();
        }
        
        public void start(){
            try {
                //根据用户名，密码，url创建一个连接工厂
                factory = new ActiveMQConnectionFactory(userName, password, brokerURL);
                //从工厂中获取一个连接
                connection = factory.createConnection();
                //测试过这个步骤不写也是可以的，但是网上的各个文档都写了
                connection.start();
                //创建一个session
                //第一个参数:是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
                //第二个参数为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
                //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
                //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
                //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                //创建一个到达的目的地，其实想一下就知道了，activemq不可能同时只能跑一个队列吧，这里就是连接了一个名为"text-msg"的队列，这个会话将会到这个队列，当然，如果这个队列不存在，将会被创建
                destination = session.createQueue("text-msg");
                //根据session，创建一个接收者对象
                consumer = session.createConsumer(destination);
                
                
                //实现一个消息的监听器
                //实现这个监听器后，以后只要有消息，就会通过这个监听器接收到
                consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        try {
                            //获取到接收的数据
                            String text = ((TextMessage)message).getText();
                            System.out.println(text);
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                });
                //关闭接收端，也不会终止程序哦
    //            consumer.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

个人感觉写的比较好的相关文章

[基于ActiveMQ的消息中间件系统 OneMM逻辑与物理架构设计详解][ActiveMQ_ OneMM]

[ActiveMQ消息队列的使用及应用][ActiveMQ]

[ActiveMQ_ OneMM]: http://www.csdn.net/article/2015-09-29/2825817-ActiveMQ-OneMM
[ActiveMQ]: http://www.cnblogs.com/zhuxiaojie/p/5564187.html