ActiveMQ（一）-入门篇

分手后的思念是犯贱 2022-06-01 11:05 232阅读 0赞

# 一、目录分析 #

![ActiveMQ目录][ActiveMQ]

从上面看出一般的项目目录类似，简单分析下其作用：

*  bin：为启动目录。
 *  conf：为配置文件，核心文件有activemq.xml（核心配置，加载jetty.xml等文件和管理数据持久化）、jetty.xml（配置启动端口）、jetty-realm.properties（配置用户名密码）几个。在登录web控制台的时候，需要用户名和密码。  
    
    
     *  data：是消息持久化的地方，默认使用kahadb，当然我们可以采用leveldb，或者采用JDBC存储到MySQL，或者干脆不使用持久化机制。 -
     *  webapps：使用jetty提供的web控制台

# 二、启动ActiveMQ #

![启动脚本][1240]  
在win下直接点击如下脚本即可启动，访问目录：[http://localhost:8161][http_localhost_8161]

![activemq登录成功][activemq]

用户名密码可在jetty-realm.properties文件中查看。

# 用户名: ，密码， [,角色 ...]
    admin: admin, admin
    user: user, user

# 三、 HelloWorld #

基于P2P的简单例子

//参数1:是否启用事务;参数2:签收模式:
    session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);

简单说，就是消费者接受到消息后，需要告诉消息服务器，
    我收到消息了。当消息服务器收到回执后，本条消息将失效。
    因此签收将对PTP模式产生很大影响。如果消费者收到消息后，
    并不签收，那么本条消息继续有效，很可能会被其他消费者消费掉!

\- AUTO\_ACKNOWLEDGE:表示在消费者receive消息的时候自动的签收  
\- CLIENT\_ACKNOWLEDGE:表示消费者receive消息后必须手动的调用acknowledge()方法进行签收。更好用，我们可以手动控制，如果消息处理失败，那么消息还有效，任然会继续处理，直到成功。  
\- DUPS\_OK\_ACKNOWLEDGE:签不签收无所谓了，只要消费者能够容忍重复的消息接受，当然这样会降低Session的开销

package com.lw.activemq.p2p;
    
    import javax.jms.Connection;
    import javax.jms.DeliveryMode;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.management.j2ee.statistics.JMSProducerStats;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    /** * @author liwen * @date:2017年12月13日 下午2:27:15 * @Function: 生产者 * @version 1.0 */
    public class JMSProducter { 
    
        private static ActiveMQConnectionFactory connectionFactory;
        private static Connection conn;
        private static Queue destination;
        private static Session session;
        private static MessageProducer producer;
    
        public static void main(String[] args) {
    
            try {
                // 创建连接工厂
                connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                        ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);
                // 获取连接
                conn = connectionFactory.createConnection();
    
                conn.start();
                // 创建session
                //参数1:是否启用事务;参数2:签收模式:
                session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
                // 创建消息队列
                destination = session.createQueue("helloworld");
                // 创建生产者
                producer = session.createProducer(destination);
                // 设置(非)持久化特性,如果非持久化,则意味MQ消息重启后会导致消息丢失
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
                // 发送消息
                sendMessage(session, producer);
                // 提交
                session.commit();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (conn != null) {
                    try {
                        conn.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
        private static void sendMessage(Session session, MessageProducer producer) throws Exception {
            for (int i = 0; i < 10; i++) {
                // 创建文本消息
                TextMessage msg = session.createTextMessage("activemq 消息:" + i);
                System.out.println("Producter 消息:" + i );
                producer.send(msg);
            }
        }
    }
    
    package com.lw.activemq.p2p;
    
    import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /** * @author liwen * @date:2017年12月13日 下午2:27:24 * @Function: 消费者 * @version 1.0 */
    public class JMSConsumer { 
    
        private static org.apache.activemq.ActiveMQConnectionFactory connectionFactory;
        private static Connection conn;
        private static Session session;
        private static Queue destination;
        private static MessageConsumer consumer;
    
        public static void main(String[] args) {
            try {
                connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                        ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);
                conn = connectionFactory.createConnection();
                conn.start();
                session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
                // 创建队列,要么是Queue,要么是Topic
                destination = session.createQueue("helloworld");
                consumer = session.createConsumer(destination);
                // 接收消息
                /* * while (true) { TextMessage text = (TextMessage) consumer.receive(); if (text * != null) { System.out.println("consumer收到的消息:" + text.getText()); } else { * break; } } */
                consumer.setMessageListener(new MessageListener() {
    
                    public void onMessage(Message message) {
                        TextMessage text = (TextMessage) message;
                        try {
                            String msg = text.getText();
                            System.out.println("consumer收到消息:" + msg);
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                });
    
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

上面只是演示消息为字符串的例子，实际项目中可能为java类的比较多。ActiveMQ当然也支持。

![ActiveMQ支持的消息类型][ActiveMQ 1]

从上图可以看出，ActiveMQ可支持流、Map、java类（需要序列化）等消息格式。

# 四、 保证消息的成功处理 #

使用CLIENT\_ACKNOWLEDGE模式解决，写在接收端。如果接收端不确认消息，那么activemq将会把这条消息一直保留，直到有一个接收端确认了消息的接收。即调用acknowledge方法。  
![\[ 保证消息的成功处理\]  
][Link 1]

[ActiveMQ]: /images/20220601/87c6a8c5acc742a78396d0ebec9534a4.png
[1240]: /images/20220601/4fc1f440cfc44ba69d57920c125f61d7.png
[http_localhost_8161]: http://localhost:8161
[activemq]: /images/20220601/f0a95c81455b436a9c029de0d66ddf6c.png
[ActiveMQ 1]: /images/20220601/f24ca291d4c34d3b803d8bd329bcbf74.png
[Link 1]: http://upload-images.jianshu.io/upload_images/3592443-0ce53bd6f067e41e.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240