ActiveMQ 是什么？

JMS 是什么？

JMS的对象模型

JMS 的消息模型


JMS的消息结构

JMS的消息结构


ActiveMQ 的特性

单机安装：
演示环境： Centos7、jdk8、activemq5.15.8

下载地址： http://activemq.apache.org/activemq-5158-release.html

解压： tar -zxvf apache-activemq-5.15.8-bin.tar.gz -C /var

修改目录名称 mv /var/apache-activemq-5.15.8/ /var/activemq/

启动： ./bin/activemq start

停止：./bin/activemq stop
操作练习
1、创建一个systemd服务文件：vi /usr/lib/systemd/system/activemq.service
2、 放入内容
[Unit]
Description=ActiveMQ service
After=network.target

[Service]
Type=forking
ExecStart=/root/activemq/bin/activemq start
ExecStop=/root/activemq/bin/activemq stop
User=root
Group=root
Restart=always
RestartSec=9
StandardOutput=syslog
StandardError=syslog
SyslogIdentifier=activemq

[Install]
WantedBy=multi-user.target

3、 找到java命令所在的目录 whereis java
4、设置activemq配置文件/var/activemq/bin/env中的JAVA_HOME
# Location of the java installation
# Specify the location of your java installation using JAVA_HOME, or specify the
# path to the "java" binary using JAVACMD
# (set JAVACMD to "auto" for automatic detection)
JAVA_HOME="/usr/local/java/jdk1.8.0_181"
JAVACMD="auto"

5、 通过systemctl管理activemq启停

启动activemq服务: systemctl start activemq
查看服务状态: systemctl status activemq
创建软件链接：ln -s /usr/lib/systemd/system/activemq.service /etc/systemd/system/multi-user.target.wants/activemq.service
开机自启: systemctl enable activemq
检测是否开启成功(enable)： systemctl list-unit-files |grep activemq

6、 防火墙配置，Web管理端口默认为8161（admin/admin），通讯端口默认为61616

添加并重启防火墙

firewall-cmd --zone=public --add-port=8161/tcp --permanent
firewall-cmd --zone=public --add-port=61616/tcp --permanent
systemctl restart firewalld.service


或者直接关闭防火墙: systemctl stop firewalld.service

7、 修改web管理系统的部分配置,配置文件在/var/activemq/conf

端口修改

<bean id="jettyPort" class="org.apache.activemq.web.WebConsolePort" init-method="start">
  <!-- the default port number for the web console -->
  <property name="host" value="0.0.0.0"/>
  <!--此处即为管理平台的端口-->
  <property name="port" value="8161"/>
</bean>


关闭登录

<bean id="securityConstraint" class="org.eclipse.jetty.util.security.Constraint">
  <property name="name" value="BASIC" />
  <property name="roles" value="user,admin" />
  <!-- 改为false即可关闭登陆 -->
  <property name="authenticate" value="true" />
</bean>


其他配置: /var/activemq/conf/jetty-realm.properties

## ---------------------------------------------------------------------------
# 在此即可维护账号密码，格式：
# 用户名:密码,角色
# Defines users that can access the web (console, demo, etc.)
# username: password [,rolename ...]
admin: admin, admin
user: 123, user

8、 JAVA客户端的使用

标准客户端使用

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


Spring中使用: http://spring.io/guides/gs/messaging-jms/

<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
    <version>5.1.3.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-broker</artifactId>
    <version>5.15.8</version>
    <exclusions>
    <exclusion>
        <artifactId>geronimo-jms_1.1_spec</artifactId>
        <groupId>org.apache.geronimo.specs</groupId>
    </exclusion>
    </exclusions>
</dependency>

支持哪些协议

常用的传输方式和配置
























OpenWite协议如何使用





添加日志输出：
生产环境不建议开启
vim activemq.xml
transport.trace=true&amp;

vim log4j.properties
log4j.logger.org.apache.activemq.transport.TransportLogger=DEBUG

MQTT 协议是什么

为什么使用MQTT协议

MQTT 的发布订阅模式

MQTT服务质量











如何使用MQTT协议







Springh使用MQTT

如何使用AUTO协议

ActiveMQ 即时通讯服务 浅析
 ActiveMQ消息队列的使用及应用
ActiveMQ prefetch机制
ActiveMQ面试专题
activeMQ消息阻塞 丢失 假死--架构

The Failover Transport(失败故障转移传输)
ActiveMq 自学（五） 队列容错机制配置学习

ActiveMQ消息游标
转载于:https://www.cnblogs.com/gogogofh/p/9570823.html

ActiveMq
概述
优点
ActiveMq的消息形式
5中消息正文格式
ActiveMq的安装
访问ActiveMq后台界面
ActiveMq的使用
添加依赖
点对点队列形式
生产者Producer
消费者

发布订阅形式
生产者
消费者


ActiveMq整合Spring
添加依赖
生产者配置
消费者配置
生产者代码
消费者代码
测试类







ActiveMq



概述


  
ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位




优点


  
多种语言和协议编写客户端。语言: 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,进行测试
  



ActiveMq的消息形式


  
一种是点对点的，即一个生产者和一个消费者一一对应 

  另一种是发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收




5中消息正文格式


  
JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。 

  　　· StreamMessage – Java原始值的数据流 

  　　· MapMessage–一套名称-值对 

  　　· TextMessage–一个字符串对象 

  　　· ObjectMessage–一个序列化的 Java对象 

  　　· BytesMessage–一个字节的数据流




ActiveMq的安装


  
安装jdk
  
将apache-activemq-5.11.2-bin.tar.gz上传到 Linux 并解压
  
启动：进入解压后的 bin 目录 

  启动：[root@localhost bin]# ./activemq start 

  关闭：[root@localhost bin]# ./activemq stop 

  查看状态：[root@localhost bin]# ./activemq status
  



访问ActiveMq后台界面

http://192.168.13.3:8161 

账号：admin 

密码：admin 
注意：如果ActiveMQ整合spring使用不要使用activemq-all-5.12.0.jar包,该版本包含了spring的jar包，但是缺少spring的jar包里面的一些方法，导致和spring整合后，启动会报错，建议使用5.11.2版本的jar包



ActiveMq的使用



添加依赖



<!-- activeMq客户端 -->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
</dependency>



点对点队列形式



生产者Producer



    @Test
    public void testQueueProducer() throws Exception {
        // 1、创建一个连接工程ConnectionFactory对象。指定服务的ip及端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.13.3:61616");
        // 2、使用ConnectionFactory对象创建一个Connection对象。
        Connection connection = connectionFactory.createConnection();
        // 3、开启连接，调用start方法
        connection.start();
        // 4、使用Connection对象创建一个Session对象。
        // 第一个参数：是否开启事务，一般不开启。当参数为false时，第二个参数才有意义。
        // 第二个参数：消息的应答模式。手动应答和自动应答。一般使用自动应答
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 5、使用Session对象创建一个Destination，目的地有两种queue、topic
        Queue queue = session.createQueue("test-queue");
        // 6、使用Session对象创建一个Producer对象。
        MessageProducer producer = session.createProducer(queue);
        // 7、使用producer发送消息。
        TextMessage textMessage = new ActiveMQTextMessage();
        textMessage.setText("使用activemq 发送的队列消息1111");
        // TextMessage textMessage2 = session.createTextMessage("使用activemq
        // 发送的队列消息");
        producer.send(textMessage);
        // 8、关闭资源。
        producer.close();
        session.close();
        connection.close();
    }



消费者



@Test
    public void testQueueConsumer() throws Exception {
        // 创建一连接工厂对象
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.13.3:61616");
        // 使用工厂对象创建一个连接
        Connection connection = connectionFactory.createConnection();
        // 开启连接
        connection.start();
        // 使用连接对象创建一个session对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建一个destination对象，使用queue
        Queue queue = session.createQueue("test-queue");
        // 使用session对象创建一个消费者
        MessageConsumer consumer = session.createConsumer(queue);
        // 使用消费者对象接收消息。
        consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                // 打印消息
                TextMessage textMessage = (TextMessage) message;
                String text = "";
                try {
                    text = textMessage.getText();
                    System.out.println(text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        System.in.read();

        /*
         * while(true) { Message message = consumer.receive(3000); if (message
         * == null) { break; } TextMessage textMessage = (TextMessage) message;
         * String text = ""; try { text = textMessage.getText();
         * System.out.println(text); } catch (JMSException e) {
         * e.printStackTrace(); } }
         */
        // 关闭资源
        consumer.close();
        session.close();
        connection.close();
    }



发布订阅形式



生产者



    @Test
    public void testTopicProducer() throws Exception {
        // 创建一个连接工厂对象
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.13.3:61616");
        // 使用连接工厂对象创建一个连接
        Connection connection = connectionFactory.createConnection();
        // 开启连接
        connection.start();
        // 使用连接对象创建一个session对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 使用session对象创建一个topic
        Topic topic = session.createTopic("test-topic");
        // 使用session创建一个producer，指定目的地时候topic。
        MessageProducer producer = session.createProducer(topic);
        // 创建一个TextMessage对象
        TextMessage message = session.createTextMessage("使用topic发送的消息111");
        // 使用producer对象发送消息。
        producer.send(message);
        // 关闭资源
        producer.close();
        session.close();
        connection.close();
    }



消费者



    @Test
    public void testTopicConsumer() throws Exception {
        // 创建一连接工厂对象
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.168:61616");
        // 使用工厂对象创建一个连接
        Connection connection = connectionFactory.createConnection();
        // 开启连接
        connection.start();
        // 使用连接对象创建一个session对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建一个destination对象，使用topic
        Topic topic = session.createTopic("test-topic");
        // 使用session对象创建一个消费者
        MessageConsumer consumer = session.createConsumer(topic);
        System.out.println("topic消费者3.。。。。");
        // 使用消费者对象接收消息。
        consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                // 打印消息
                TextMessage textMessage = (TextMessage) message;
                String text = "";
                try {
                    text = textMessage.getText();
                    System.out.println(text);
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        //程序等待
        System.in.read();

        // 关闭资源
        consumer.close();
        session.close();
        connection.close();
    }



ActiveMq整合Spring



添加依赖



        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context-support</artifactId>
        </dependency>



生产者配置



    <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.13.3:61616" />
    </bean>
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory"
        class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
        <property name="targetConnectionFactory" ref="targetConnectionFactory" />
    </bean>
    <!-- 配置生产者 -->
    <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
        <property name="connectionFactory" ref="connectionFactory" />
    </bean>
    <!--这个是队列目的地，点对点的 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>spring-queue</value>
        </constructor-arg>
    </bean>
    <!--这个是主题目的地，一对多的 -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="item-add-topic" />
    </bean>



消费者配置



    <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.13.3:61616" />
    </bean>
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory"
        class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
        <property name="targetConnectionFactory" ref="targetConnectionFactory" />
    </bean>
    <!--这个是队列目的地，点对点的 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>spring-queue</value>
        </constructor-arg>
    </bean>
    <!--这个是主题目的地，一对多的 -->
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="item-add-topic" />
    </bean>

    <!-- 配置消息监听器 -->
    <bean id="myMessageListener" class="com.mydeertrip.listener.MyMessageListener"/>
    <!-- 配置监听容器 -->
    <!-- 消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="queueDestination" />
        <property name="messageListener" ref="myMessageListener" />
    </bean>
    <!-- 配置消息监听器 -->
    <bean id="itemAddListener" class="com.mydeertrip.listener.ItemAddListener"/>
    <!-- 配置监听容器 -->
    <!-- 消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="topicDestination" />
        <property name="messageListener" ref="itemAddListener" />
    </bean>



生产者代码



@Test
    public void testQueueProducer() throws Exception {
        //初始化spring容器
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("classpath:/spring/applicationContext-activemq.xml");
        //从容器中获得JMSTemplate对象
        JmsTemplate jmsTemplate = applicationContext.getBean(JmsTemplate.class);
        //从容器在获得Destination对象
        Queue queue = applicationContext.getBean(Queue.class);
        //第一个参数：指定发送的目的地
        //第二个参数：消息的构造器对象
        jmsTemplate.send(queue, new MessageCreator() {

            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage textMessage = session.createTextMessage("使用spring和active整合发送queue消息aaaaaa");
                return textMessage;
            }
        });

    }



消费者代码



public class MyMessageListener implements MessageListener {


    @Override
    public void onMessage(Message message) {
        //取消息的内容
        try {
            TextMessage textMessage = (TextMessage) message;
            //取内容
            String text = textMessage.getText();
            System.out.println(text);
            //其他业务逻辑
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}



测试类



    @Test
    public void testQueueConsumer() throws Exception {
        //初始化spring容器
        ApplicationContext applicationContext = new ClassPathXmlApplicationContext("classpath:spring/applicationContext-activemq.xml");
        System.in.read();
    }


  
测试类的作用是用来初始化spring容器，实例化消息生产者对象

一下载+安装+启动
1.activeMQ私下里都被叫做MQ，所以起初在百度搜索的时候搜不到，直到把关键字换成activemq
2.activemq有window版本和unix版本之分，unix版本还有binary，src和dev版本之分，我倾向binary，安装较简单。
3.去mq官网下载5.9.1版本http://www.apache.org/dyn/closer.cgi?path=/activemq/5.9.1/apache-activemq-5.9.1-bin.tar.gz
4.tar -zxvf 解压 apache-activemq-5.9.1-bin.tar.gz
5.cd apache-activemq-5.9.1/bin
6.chmod 755 activemq
7. ./activemq
8.查看localhost:8161/admin,其中localhost是机器名，8161是activemq默认端口
9.admin/admin登录