activeMQ入门、

秒速五厘米 2022-04-11 12:10 199阅读 0赞

ActiveMQ是什么？优缺点是什么？使用场景、为什么要使用ActiveMQ？ActiveMQ有什么特性？

》

> ActiveMQ官网下载地址：[http://activemq.apache.org/download.html][http_activemq.apache.org_download.html]  
> windows版本解压可用：目录结构如下【了解即可】：  
> \`  
> bin: 脚本文件、注意里面存在32和64位、我64位的32位直接删掉即可、点击activemq.bat启动  
> conf: 基本配置文件  
> data: 日志文件  
> docs: 说明文档  
> examples: 简单的实例  
> lib: activemq所需jar包,  
> webapps: 存放项目的目录  
> webapps: 存放示例项目的目录  
> activemq-all-5.15.8.jar : activemq 综合依赖jar

![1][]

ActiveMQ默认启动时，启动了内置的jetty服务器，提供一个用于监控ActiveMQ的admin应用：  
浏览器访问： [http://127.0.0.1:8161/admin/][http_127.0.0.1_8161_admin]  
默认账号密码均是 admin  
![1][1 1]

###### 1、ActiveMQ是什么： ######

> 如下百科解释：  
> 开源消息总线， JMS Provider的具体实现，那么JMS（java massage service）是一种规范接口；

![1][1 2]

专业术语的了解：

> Provider/MessageProvider：生产者  
> Consumer/MessageConsumer：消费者  
> PTP：Point To Point，点对点通信消息模型  
> Pub/Sub：Publish/Subscribe，发布订阅消息模型  
> Queue：队列，目标类型之一，和PTP结合  
> Topic：主题，目标类型之一，和Pub/Sub结合  
> ConnectionFactory：连接工厂，JMS用它创建连接  
> Connnection：JMS Client到JMS Provider的连接  
> Destination：消息目的地，由Session创建  
> Session：会话，由Connection创建，实质上就是发送、接受消息的一个线程，因此生产者、消费者都是Session创建的

###### 5.2ActiveMQ的优点： ######

> 成熟吧，出的时间较早，维护较慢，但是近期社区维护有活跃起来了，基本上一个月一个小版本、  
> 一般作为解耦和异步使用；  
> 其单级吞吐级别是 万级/秒

###### 5.2ActiveMQ的缺点： ######

> 性能低，需要优化；

###### 5.2ActiveMQ使用场景 ######

> 多个项目之间集成  
> (1) 跨平台  
> (2) 多语言  
> (3) 多项目  
> 降低系统间模块的耦合度，解耦  
> (1) 软件扩展性  
> 系统前后端隔离  
> (1) 前后端隔离，屏蔽高安全区

###### 5.2为什么要使用 ActiveMQ ######

使用ActiveMQ解决系统存在的这样几个问题：

> 1.通信的同步性  
> client端发起调用后，必须等待server处理完成并返回结果后才能继续执行  
> 2.client 和 server 的生命周期耦合太高  
> client进程和server服务进程都必须可用，如果server出现问题或者网络故障，那么client端会收到异常  
> 3.点对点通信,即PTP（point TO point）  
> client端的一次调用只能发送给某一个单独的服务对象，无法一对多

解决这样的问题：

> JMS 通过面向消息中间件 MOM：Message Oriented Middleware 解决以上的问题；JMS是规范，使用ActiveMq 这样的具体实现来解决以上问题；
> 
>  *  client 端发送消息至 消息中间件（消息服务器）ActiveMq ，消息服务器把消息储存在若干队列中，在适合的时候把消息从队列中取出，发送给消息的接受者；
>  *  在这个过程中，发送和接受是异步的，也就是发送无需等待，而且发送者和接受者的生命周期也没有必然关系；
>  *  在经典的 pub/sub （Publish/Subscribe，发布订阅消息模型）模式下、 完成一对多的通信,如下图；  
>     ![2][]

###### 5.1 ActiveMq 的特性 ######

> 多种语言和协议编写客户端。语言: Java, C, C++, C\#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP  
> 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)  
> 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性  
> 通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上  
> 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA  
> 支持通过JDBC和journal提供高速的消息持久化  
> 从设计上保证了高性能的集群,客户端-服务器,点对点  
> 支持Ajax  
> 支持与Axis的整合  
> 可以很容易得调用内嵌JMS provider,进行测试

--------------------

> ###### 代码思路： ######
> 
> 1、创建链接工厂ConnectionFactory  
> 2、创建链接  
> 3、创建session 、  
> 4、通过session 创建Destination 、消息发送接收的地点；
> 
>  *  在PTP模式是 队列queue
>  *  在pub/sub 模式是主题topic
> 
> 5、通过session 创建 生产者/消费者  
> 6、设置持久化、若不设置持久化，MQ的重启，消息丢失；持久化到kahdb /leveldb /jdbc  
> 7、session创建、定义消息对象发送消息
> 
>  *  消息头（路由）+消息属性（消息选择器）+消息体（JMS规范的5种类型消息）
> 
> 8、释放链接  
> 必须close connection，只有这样ActiveMQ才会释放资源  
> 、  
> 源码分享：  
> [https://github.com/medoo-Ai/activemq.git][https_github.com_medoo-Ai_activemq.git]

Session，用于发送和接受消息，而且是单线程的，支持事务的。如果Session开启事务支持，那么Session将保存一组信息，要么commit到MQ，要么回滚这些消息。Session可以创建MessageProducer/MessageConsumer。

1、创建maven工程，引入依赖、

<dependencies>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>5.15.8</version>
            </dependency>
        </dependencies>

2、创建pub生产者

package com.mq;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class Producter { 
    
        //ActiveMq 的默认用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        //ActiveMq 的默认登录密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        //ActiveMQ 的链接地址
        private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
        AtomicInteger count = new AtomicInteger(0);
        //链接工厂
        ConnectionFactory connectionFactory;
        //链接对象
        Connection connection;
        //事务管理
        Session session;
        ThreadLocal<MessageProducer> threadLocal = new ThreadLocal<MessageProducer>();
    
        public void init(){ 
            try { 
                //创建一个链接工厂
                connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);
                //从工厂中创建一个链接
                connection  = connectionFactory.createConnection();
                //开启链接
                connection.start();
                //创建一个事务（这里通过参数可以设置事务的级别）
                session = connection.createSession(true, Session.SESSION_TRANSACTED);
            } catch (JMSException e) { 
                e.printStackTrace();
            }
        }
    
        public void sendMessage(String disname){ 
            try { 
                //创建一个消息队列
                Queue queue = session.createQueue(disname);
                //消息生产者
                MessageProducer messageProducer = null;
                if(threadLocal.get()!=null){ 
                    messageProducer = threadLocal.get();
                }else{ 
                    messageProducer = session.createProducer(queue);
                    threadLocal.set(messageProducer);
                }
               while(true){ 
                    Thread.sleep(1000);
                    int num = count.getAndIncrement();
                    //创建一条消息
                    TextMessage msg = session.createTextMessage(Thread.currentThread().getName()+
                            "productor:我是大帅哥，我现在正在生产东西！,count:"+num);
                    System.out.println(Thread.currentThread().getName()+
                            "productor:我是大帅哥，我现在正在生产东西！,count:"+num);
                    //发送消息
                    messageProducer.send(msg);
                    //提交事务
                    session.commit();
                }
            } catch (JMSException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

2.1、创建consume消费者

package com.mq;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class Comsumer { 
    
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
    
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
    
        private static final String BROKEN_URL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
        ConnectionFactory connectionFactory;
    
        Connection connection;
    
        Session session;
    
        ThreadLocal<MessageConsumer> threadLocal = new ThreadLocal<MessageConsumer>();
        AtomicInteger count = new AtomicInteger();
    
        public void init(){ 
            try { 
                connectionFactory = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEN_URL);
                connection  = connectionFactory.createConnection();
                connection.start();
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            } catch (JMSException e) { 
                e.printStackTrace();
            }
        }
    
    
        public void getMessage(String disname){ 
            try { 
                Queue queue = session.createQueue(disname);
                MessageConsumer consumer = null;
    
                if(threadLocal.get()!=null){ 
                    consumer = threadLocal.get();
                }else{ 
                    consumer = session.createConsumer(queue);
                    threadLocal.set(consumer);
                }
                while(true){ 
                    Thread.sleep(1000);
                    TextMessage msg = (TextMessage) consumer.receive();
                    if(msg!=null) { 
                        msg.acknowledge();
                        System.out.println(Thread.currentThread().getName()+": Consumer:我是消费者，我正在消费Msg"+msg.getText()+"--->"+count.getAndIncrement());
                    }else { 
                        break;
                    }
                }
            } catch (JMSException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

2.2、测试类、

package mq;
    
    import com.mq.Producter;
    
    public class TestMq { 
        public static void main(String[] args){ 
            Producter producter = new Producter();
            producter.init();
            TestMq testMq = new TestMq();
            try { 
                Thread.sleep(1000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            //Thread 1
            new Thread(testMq.new ProductorMq(producter)).start();
            //Thread 2
            new Thread(testMq.new ProductorMq(producter)).start();
            //Thread 3
            new Thread(testMq.new ProductorMq(producter)).start();
            //Thread 4
            new Thread(testMq.new ProductorMq(producter)).start();
            //Thread 5
            new Thread(testMq.new ProductorMq(producter)).start();
        }
    
        private class ProductorMq implements Runnable{ 
            Producter producter;
            public ProductorMq(Producter producter){ 
                this.producter = producter;
            }
    
    
            public void run() { 
                while(true){ 
                    try { 
                        producter.sendMessage("Jaycekon-MQ");
                        Thread.sleep(10000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    }

生产者生产消息、

G:\tools\java_jdk\jdk-8u121\bin\java.exe "-javaagent:G:\Develop\IDE\IntelliJ IDEA 2018.3\lib\idea_rt.jar=50599:G:\Develop\IDE\IntelliJ IDEA 2018.3\bin" -Dfile.encoding=UTF-8 -classpath G:\tools\java_jdk\jdk-8u121\jre\lib\charsets.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\deploy.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\access-bridge-64.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\cldrdata.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\dnsns.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\jaccess.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\jfxrt.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\localedata.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\nashorn.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\sunec.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\sunjce_provider.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\sunmscapi.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\sunpkcs11.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\zipfs.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\javaws.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\jce.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\jfr.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\jfxswt.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\jsse.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\management-agent.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\plugin.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\resources.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\rt.jar;G:\Develop\ideaworkspace2\mq\activemq\target\test-classes;G:\Develop\ideaworkspace2\mq\activemq\target\classes;G:\tools\apache-maven-3.5.4\repository\org\apache\activemq\activemq-all\5.15.8\activemq-all-5.15.8.jar mq.TestMq
     INFO | Successfully connected to tcp://localhost:61616
    Thread-2productor:我是大帅哥，我现在正在生产东西！,count:4
    Thread-4productor:我是大帅哥，我现在正在生产东西！,count:0
    Thread-5productor:我是大帅哥，我现在正在生产东西！,count:3
    Thread-6productor:我是大帅哥，我现在正在生产东西！,count:1
    Thread-3productor:我是大帅哥，我现在正在生产东西！,count:2
    Thread-3productor:我是大帅哥，我现在正在生产东西！,count:5
    Thread-2productor:我是大帅哥，我现在正在生产东西！,count:6
    Thread-6productor:我是大帅哥，我现在正在生产东西！,count:7
    Thread-5productor:我是大帅哥，我现在正在生产东西！,count:8
    Thread-4productor:我是大帅哥，我现在正在生产东西！,count:9

package mq;
    
    import com.mq.Comsumer;
    public class TestConsumer { 
        public static void main(String[] args){ 
            Comsumer comsumer = new Comsumer();
            comsumer.init();
            TestConsumer testConsumer = new TestConsumer();
            new Thread(testConsumer.new ConsumerMq(comsumer)).start();
            new Thread(testConsumer.new ConsumerMq(comsumer)).start();
            new Thread(testConsumer.new ConsumerMq(comsumer)).start();
            new Thread(testConsumer.new ConsumerMq(comsumer)).start();
        }
    
        private class ConsumerMq implements Runnable{ 
            Comsumer comsumer;
            public ConsumerMq(Comsumer comsumer){ 
                this.comsumer = comsumer;
            }
            public void run() { 
                while(true){ 
                    try { 
                        comsumer.getMessage("Jaycekon-MQ");
                        Thread.sleep(10000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    }

消费者消费消息、

> 从中可以看出消费者并不是有序的消费对应生产者生产的的信息的、  
> 所以使用签到机制、签到机制分为手动和自动、以防止线程占用；  
> 正常情况采用手动，因为有些消息未处理，如果被签到，造成信息错误、  
> 只有满足条件以后才能够被签到；

G:\tools\java_jdk\jdk-8u121\bin\java.exe "-javaagent:G:\Develop\IDE\IntelliJ IDEA 2018.3\lib\idea_rt.jar=50616:G:\Develop\IDE\IntelliJ IDEA 2018.3\bin" -Dfile.encoding=UTF-8 -classpath G:\tools\java_jdk\jdk-8u121\jre\lib\charsets.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\deploy.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\access-bridge-64.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\cldrdata.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\dnsns.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\jaccess.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\jfxrt.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\localedata.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\nashorn.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\sunec.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\sunjce_provider.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\sunmscapi.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\sunpkcs11.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\ext\zipfs.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\javaws.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\jce.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\jfr.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\jfxswt.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\jsse.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\management-agent.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\plugin.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\resources.jar;G:\tools\java_jdk\jdk-8u121\jre\lib\rt.jar;G:\Develop\ideaworkspace2\mq\activemq\target\test-classes;G:\Develop\ideaworkspace2\mq\activemq\target\classes;G:\tools\apache-maven-3.5.4\repository\org\apache\activemq\activemq-all\5.15.8\activemq-all-5.15.8.jar mq.TestConsumer
     INFO | Successfully connected to tcp://localhost:61616
    Thread-3: Consumer:我是消费者，我正在消费MsgThread-5productor:我是大帅哥，我现在正在生产东西！,count:3--->0
    Thread-2: Consumer:我是消费者，我正在消费MsgThread-2productor:我是大帅哥，我现在正在生产东西！,count:4--->1
    Thread-4: Consumer:我是消费者，我正在消费MsgThread-6productor:我是大帅哥，我现在正在生产东西！,count:1--->2
    Thread-5: Consumer:我是消费者，我正在消费MsgThread-3productor:我是大帅哥，我现在正在生产东西！,count:2--->3
    Thread-3: Consumer:我是消费者，我正在消费MsgThread-6productor:我是大帅哥，我现在正在生产东西！,count:7--->4
    Thread-2: Consumer:我是消费者，我正在消费MsgThread-4productor:我是大帅哥，我现在正在生产东西！,count:0--->5
    Thread-4: Consumer:我是消费者，我正在消费MsgThread-2productor:我是大帅哥，我现在正在生产东西！,count:6--->6
    Thread-5: Consumer:我是消费者，我正在消费MsgThread-3productor:我是大帅哥，我现在正在生产东西！,count:5--->7
    Thread-3: Consumer:我是消费者，我正在消费MsgThread-2productor:我是大帅哥，我现在正在生产东西！,count:11--->8
    Thread-2: Consumer:我是消费者，我正在消费MsgThread-5productor:我是大帅哥，我现在正在生产东西！,count:8--->9
    Thread-4: Consumer:我是消费者，我正在消费MsgThread-3productor:我是大帅哥，我现在正在生产东西！,count:10--->10
    Thread-5: Consumer:我是消费者，我正在消费MsgThread-4productor:我是大帅哥，我现在正在生产东西！,count:9--->11

[http://127.0.0.1:8161/admin/][http_127.0.0.1_8161_admin] 里面的Queues 中查看我们生产以及消费的消息详情；  
![1][1 3]  
Number Of Consumers：表示在该队列上还有多少消费者在等待接受消息

--------------------

》

> 签收模式：

在通过Connection创建Session的时候，需要设置2个参数，一个是否支持事务，另一个是签收的模式；  
消费者接受到消息后，需要告诉消息服务器，我收到消息了。当消息服务器收到回执后，本条消息将失效。因此签收将对PTP模式产生很大影响。如果消费者收到消息后，并不签收，那么本条消息继续有效，很可能会被其他消费者消费掉。

session源码：  
![3][]

//消费者receive消息的时候自动的签收
        static final int AUTO_ACKNOWLEDGE = 1;
        //消费者receive消息的时候手动的签收
        static final int CLIENT_ACKNOWLEDGE = 2;
        //消费者receive消息的时候无所谓的签不签收
        static final int DUPS_OK_ACKNOWLEDGE = 3;
        static final int SESSION_TRANSACTED = 0;

》  
签收模式选择：

> 接收到了消息，并不意味着成功的处理了消息，假设我们采用手动签收的方式，只有在消息成功处理的前提下才进行签收，那么只要消息处理失败，那么消息还有效，仍然会继续消费，直至成功处理。

》  
消息的优先级以及存活时间TTL

> TTL，消息的存活时间，一句话：生产者生产了消息，如果消费者不来消费，那么这条消息保持多久的有效期  
> priority，消息优先级，0-9。0-4是普通消息，5-9是加急消息，消息默认级别是4。注意，消息优先级只是一个理论上的概念，并不能绝对保证优先级高的消息一定被消费者优先消费！也就是说ActiveMQ并不能保证消费的顺序性！  
> 在上面实例中有体现、

deliveryMode，如果不指定，默认是持久化的消息。如果可以容忍消息的丢失，那么采用非持久化的方式，将会改善性能、减少存储的开销。

[http_activemq.apache.org_download.html]: http://activemq.apache.org/download.html
[1]: /images/20220411/4f682d092ed749e297915f8ee2c9bd54.png
[http_127.0.0.1_8161_admin]: http://127.0.0.1:8161/admin/
[1 1]: /images/20220411/06d71a5c961d4069ac7b552ea855f501.png
[1 2]: /images/20220411/5b5dad2e09d64aecae75232ea8b1dc38.png
[2]: /images/20220411/e7322acec7134d8bb890e4da20c3e71d.png
[https_github.com_medoo-Ai_activemq.git]: https://github.com/medoo-Ai/activemq.git
[1 3]: /images/20220411/c6706c7ea0354812a8482e97cbaf9d57.png
[3]: /images/20220411/bc2bcb209ac447c08179171d2b28a7a5.png