返回 登录
8

ActiveMQ几种模式总结

原文地址:http://blog.csdn.net/czp11210/article/details/8822070
Queue(点到点)模式
在点对点的传输方式中,消息数据被持久化,每条消息都能被消费,没有监听QUEUE地址也能被消费,数据不会丢失,一对一的发布接受策略,保证数据完整。
创建MAVEN项目
点击下一步,填写grupId为me.czp,artifactId为example-mq,name为exampleMq,点完成
然后在Package Explorer看到example-mq项目,如下:
创建生产者
package mq.p2p;

import java.util.Date;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.MessageProducer;

import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;

importorg.apache.activemq.ActiveMQConnectionFactory;

public class Producer {

     publicstatic void main(String[] args) {

               Stringuser = ActiveMQConnection.DEFAULT_USER;

               Stringpassword = ActiveMQConnection.DEFAULT_PASSWORD;

               Stringurl = ActiveMQConnection.DEFAULT_BROKER_URL;

               Stringsubject = "test.queue";

               ConnectionFactorycontectionFactory = new ActiveMQConnectionFactory( user, password, url);

               try{

                        Connectionconnection = contectionFactory.createConnection();

                        connection.start();

                        Sessionsession = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

                        Destinationdestination = session.createQueue(subject);

                        MessageProducerproducer = session.createProducer(destination);

                        for(int i = 0; i <= 20; i++) {

                                 MapMessagemessage = session.createMapMessage();

                                 Datedate = new Date();

                                 message.setLong("count",date.getTime());

                                 Thread.sleep(1000);

                                 producer.send(message);

                                 System.out.println("--发送消息:" +date);

                        }

                        Thread.sleep(2000);

                        session.commit();

                        session.close();

                        connection.close();

               }catch (JMSException e) {

                        e.printStackTrace();

               }catch (InterruptedException e) {

                        e.printStackTrace();

               }

     }

}

创建消费者
package mq.p2p;

import java.util.Date;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Customer {

public static void main(String[] args) {

    Stringuser = ActiveMQConnection.DEFAULT_USER;

    Stringpassword = ActiveMQConnection.DEFAULT_PASSWORD;

    Stringurl = ActiveMQConnection.DEFAULT_BROKER_URL;

    Stringsubject = "test.queue";

    ConnectionFactoryconnectionFactory = new ActiveMQConnectionFactory( user, password, url);

    Connectionconnection;

    try {

        connection= connectionFactory.createConnection();

        connection.start();

        final Session session =connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

        Destinationdestination = session.createQueue(subject);

        MessageConsumermessage = session.createConsumer(destination);

        message.setMessageListener(new MessageListener() {

            publicvoid onMessage(Message msg){

                MapMessagemessage = (MapMessage) msg;

                try {

                    System.out.println("--收到消息:" +new Date());

                    session.commit();

                }catch(JMSException e) {

                    e.printStackTrace();

                }

            }

        });

        Thread.sleep(30000);

        session.close();

        Thread.sleep(30000);

        connection.close();

        Thread.sleep(30000);

    }catch(JMSException e) {

        e.printStackTrace();

    }catch(InterruptedException e) {

        e.printStackTrace();

    }

}

}

package mq.p2p;

import java.util.Date;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

/*

*第二个消费者

*/

public class Customer2 {

publicstatic void main(String[] args) {

    Stringuser = ActiveMQConnection.DEFAULT_USER;

    Stringpassword = ActiveMQConnection.DEFAULT_PASSWORD;

    Stringurl = ActiveMQConnection.DEFAULT_BROKER_URL;

    Stringsubject = "test.queue";

    ConnectionFactoryconnectionFactory = new ActiveMQConnectionFactory( user, password, url);

    Connectionconnection;

    try{

        connection= connectionFactory.createConnection();

        connection.start();

        finalSession session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);

        Destinationdestination = session.createQueue(subject);

        MessageConsumermessage = session.createConsumer(destination);

        message.setMessageListener(newMessageListener() {

            publicvoid onMessage(Message msg) {

                MapMessagemessage = (MapMessage) msg;

                try{

                    System.out.println("--收到消息2:"+ new Date(message.getLong("count")));

                    session.commit();

                }catch (JMSException e) {

                    e.printStackTrace();

                }

            }

        });

        Thread.sleep(30000);

        session.close();

        Thread.sleep(30000);

        connection.close();

        Thread.sleep(30000);

    }catch (JMSException e) {

        e.printStackTrace();

    }catch (InterruptedException e) {

        e.printStackTrace();

    }

}

}

学习体会
1. 运行结果,一个消息只能由一个消费者消费,不能同时被多个消费者获取
2. 测试表明,消息消费者消费消息时机是队列没有被消息提供者锁住,也就是说只有消息提供者执行了session.close()后消费者才会执行onMessage()方法
3. 启动顺序没有要求,可以先启动消费者,再启动提供者,也可以先启动提供者,然后再启动消费者
Topic(发布/订阅)模式
创建生产者
package mq.topic;

import java.util.Date;

import javax.jms.Connection;

import javax.jms.Destination;

import javax.jms.MapMessage;

import javax.jms.MessageProducer;

import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public classPublisher {

public static void main(String[] arg){

    Stringuser = ActiveMQConnection.DEFAULT_USER;

    Stringpassword = ActiveMQConnection.DEFAULT_PASSWORD;

    Stringurl = "tcp://localhost:61616";

    Stringsubject = "mq.topic";

    ActiveMQConnectionFactory amcf = newActiveMQConnectionFactory(user, password, url);

    try {

        Connectionconn = amcf.createConnection();

        conn.start();

        Sessionsession = conn.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);

        Destinationd = session.createTopic(subject);

        MessageProducerproducer = session.createProducer(d);

        for (int i = 0; i <= 20; i++){

            MapMessagemessage = session.createMapMessage();

            Datedate = newDate();

            message.setLong("count",date.getTime());

            Thread.sleep(1000);

            producer.send(message);

            System.out.println("--发送消息:" + date);

        }

        session.commit();

        session.close();

        conn.close();

    }catch(Exception e) {

        e.printStackTrace();

    }

}

}

创建消费者
package mq.topic;

import java.util.Date;

import javax.jms.Connection;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.Session;

import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class SubscriberFirst {

public static void main(String[] args) {

    Stringuser = ActiveMQConnection.DEFAULT_USER;

    Stringpassword = ActiveMQConnection.DEFAULT_PASSWORD;

    Stringurl = "tcp://localhost:61616";

    Stringsubject = "mq.topic";

    ActiveMQConnectionFactoryfactory = newActiveMQConnectionFactory(user, password, url);

    Connectionconnection;

    try {

        connection= factory.createConnection();

        connection.start();

        final Session session =connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

        Topictopic = session.createTopic(subject);

        MessageConsumerconsumer = session.createConsumer(topic);

        consumer.setMessageListener(new MessageListener() {

            publicvoid onMessage(Message msg){

                MapMessagemessage = (MapMessage) msg;

                try {

                    System.out.println("--订阅者一收到消息:" +new Date(message.getLong("count")));

                    session.commit();

                }catch(JMSException e) {

                    e.printStackTrace();

                }

            }

        });

    }catch(JMSException e) {

        e.printStackTrace();

    }

}

}

package mq.topic;

import java.util.Date;

import javax.jms.Connection;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.Session;

import javax.jms.Topic;

importorg.apache.activemq.ActiveMQConnection;

importorg.apache.activemq.ActiveMQConnectionFactory;

public class SubscriberSecond {

     publicstatic void main(String[] args) {

               Stringuser = ActiveMQConnection.DEFAULT_USER;

               Stringpassword = ActiveMQConnection.DEFAULT_PASSWORD;

               Stringurl = "tcp://localhost:61616";

               Stringsubject = "mq.topic";

               ActiveMQConnectionFactoryfactory = new ActiveMQConnectionFactory(user, password, url);

               Connectionconnection;

               try{

                        connection= factory.createConnection();

                        connection.start();

                        finalSession session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);

                        Topictopic = session.createTopic(subject);

                        MessageConsumerconsumer = session.createConsumer(topic);

                        consumer.setMessageListener(newMessageListener() {

                                 publicvoid onMessage(Message msg) {

                                           MapMessagemessage = (MapMessage) msg;

                                           try{

                                                    System.out.println("--订阅者二收到消息:"+ new Date(message.getLong("count")));

                                                    session.commit();

                                           }catch (JMSException e) {

                                                    e.printStackTrace();

                                           }

                                 }

                        });

               }catch (JMSException e) {

                        e.printStackTrace();

               }

     }

}

学习体会
1.生产者和消费者之间有时间上的相关性。订阅一个主题的消费者只能消费自它订阅之后发布的消息。JMS 规范允许客户创建持久订阅,这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。
2.订阅者一和订阅者二都能收到一样的消息,也就是说在发布订阅模式下,一份消息可以被多个消费者消费
消息相关概念
消息类型
JMS 消息由以下三部分组成:
 消息头。每个消息头字段都有相应的getter 和setter 方法。
 消息属性。如果需要除消息头字段以外的值,那么可以使用消息属性。
 消息体。消息主题内容,JMS定义消息包括TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage类型
消息确认
JMS 消息只有在被确认之后,才认为已经被成功地消费了。消息的成功消费通
常包含三个阶段:客户接收消息、客户处理消息和消息被确认。
在事务性会话中,当一个事务被提交的时候,确认自动发生。在非事务性会
话中,消息何时被确认取决于创建会话时的应答模式(acknowledgementmode)。
该参数有以下三个可选值:
Session.AUTO_ACKNOWLEDGE。当客户成功的从receive方法返回的时候,
或者从MessageListener.onMessage 方法成功返回的时候,会话自动确认
客户收到的消息。
 Session.CLIENT_ACKNOWLEDGE。客户通过消息的acknowledge 方法确认消
息。需要注意的是,在这种模式中,确认是在会话层上进行:确认一个被
消费的消息将自动确认所有已被会话消费的消息。例如,如果一个消息消
费者消费了10 个消息,然后确认第5 个消息,那么所有10 个消息都被确
认。
 Session.DUPS_ACKNOWLEDGE。该选择只是会话迟钝的确认消息的提交。如
果JMS provider 失败,那么可能会导致一些重复的消息。如果是重复的
消息,那么JMS provider 必须把消息头的JMSRedelivered 字段设置为
true。
消息持久化
AMQ Message Store
AMQ Message Store 是ActiveMQ5.0缺省的持久化存储。Message commands 被
保存到transactional journal(由rolling data logs 组成)。Messages 被保
存到data logs 中,同时被reference store 进行索引以提高存取速度。Date logs
由一些单独的data log 文件组成,缺省的文件大小是32M,如果某个消息的大
小超过了data log 文件的大小,那么可以修改配置以增加data log 文件的大小。
Generated by FoxitPDF Creator © Foxit Software
http://www.foxitsoftware.comFor evaluation only.
如果某个data log 文件中所有的消息都被成功消费了,那么这个data log 文件
将会被标记,以便在下一轮的清理中被删除或者归档。以下是其配置的一个例子:
Xml 代码

评论