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

一、下载 apache-activemq-5.13.2-bin.zip后解压，运行apache-activemq-5.13.2\bin\win64路径下的activemq.bat
 
 
运行成功如下
 
 
注：运行失败的话可以把apache-activemq-5.13.2\conf下的activemq.xml文件0.0.0.0:61616改成127.0.0.1:61616
 
        <transportConnectors>
            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
            <transportConnector name="openwire" uri="tcp://127.0.0.1:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="amqp" uri="amqp://127.0.0.1:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="stomp" uri="stomp://127.0.0.1:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="mqtt" uri="mqtt://127.0.0.1:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="ws" uri="ws://127.0.0.1:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
        </transportConnectors>
 
二、访问http://localhost:8161/admin/queues.jsp用户名密码默认都是admin
 
 
三、使用Maven导入相关依赖
 
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-all</artifactId>
      <version>5.15.4</version>
    </dependency>
 
注：不同版本jdk对应不同的MQ，具体如下
 
MQ版本号
Build-Jdk
依赖JDK
apache-activemq-5.0.0
1.5.0_12
1.5+
apache-activemq-5.1.0
1.5.0_12
1.5+
apache-activemq-5.2.0
1.5.0_15
1.5+
apache-activemq-5.3.0
1.5.0_17
1.5+
apache-activemq-5.4.0
1.5.0_19
1.5+
apache-activemq-5.5.0
1.6.0_23
1.6+
apache-activemq-5.6.0
1.6.0_26
1.6+
apache-activemq-5.7.0
1.6.0_33
1.6+
apache-activemq-5.8.0
1.6.0_37
1.6+
apache-activemq-5.9.0
1.6.0_51
1.6+
apache-activemq-5.10.0
1.7.0_12-ea
1.7+
apache-activemq-5.11.0
1.7.0_60
1.7+
apache-activemq-5.12.0
1.7.0_80
1.7+
apache-activemq-5.13.0
1.7.0_80
1.7+
apache-activemq-5.14.0
1.7.0_80
1.7+
apache-activemq-5.15.0
1.8.0_112
1.8+
四、Queue队列模式
 
（1）创建生产者
 
package springboot.activeMQ;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Producer {
	private static final String url="tcp://127.0.0.1:61616";//服务地址，端口默认61616
	private static final String queueName="queue-test";//要创建的消息名称
	public static void main(String[] args) throws JMSException {
		//1.创建ConnectiongFactory,绑定地址
		ConnectionFactory factory=new ActiveMQConnectionFactory(url);
		//2.创建Connection
		Connection connection= factory.createConnection();
		//3.启动连接
		connection.start();
		//4.创建会话
		Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		//5.创建一个目标
		Destination destination=session.createQueue(queueName);
		//6.创建一个生产者
		MessageProducer producer=session.createProducer(destination);
		for (int i = 0; i < 20; i++) {
			//7.创建消息
			TextMessage textMessage=session.createTextMessage("我是消息生产者："+i);
			//8.发送消息
			producer.send(textMessage);
			System.out.println("发送消息："+i);
		}
		connection.close();
	}
}
 
运行结果
 
 
（2）创建消费者
 
package springboot.activeMQ;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Consumer {
	private static final String url="tcp://127.0.0.1:61616";//端口默认
	private static final String queueName="queue-test";//要消费的消息名称
	public static void main(String[] args) throws JMSException {
		//1.创建ConnectiongFactory,绑定地址
		ConnectionFactory factory=new ActiveMQConnectionFactory(url);
		//2.创建Connection
		Connection connection= factory.createConnection();
		//3.启动连接
		connection.start();
		//4.创建会话
		Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		//5.创建一个目标
		Destination destination=session.createQueue(queueName);
		//6.创建一个消费者
		MessageConsumer consumer=session.createConsumer(destination);
		//7.创建一个监听器
		consumer.setMessageListener(new MessageListener() {
			public void onMessage(Message arg0) {
				TextMessage textMessage=(TextMessage)arg0;
				try {
					System.out.println("接收消息："+textMessage.getText());
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		});
	}
}
 
运行结果
 
 
再启动一次消费者
 
 
运行一次生产者，此时两个消费者分别显示
 
 
 
五、Topic主题模式
 
（1）创建生产者
 
package springboot.activeMQ;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class TipicProducer {
	private static final String url="tcp://127.0.0.1:61616";//服务地址，端口默认61616
	private static final String topicName="topic-test";//要创建的消息名称
	public static void main(String[] args) throws JMSException {
		//1.创建ConnectiongFactory,绑定地址
		ConnectionFactory factory=new ActiveMQConnectionFactory(url);
		//2.创建Connection
		Connection connection= factory.createConnection();
		//3.启动连接
		connection.start();
		//4.创建会话
		Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		//5.创建一个目标
		Destination destination=session.createTopic(topicName);
		//6.创建一个生产者
		MessageProducer producer=session.createProducer(destination);
		for (int i = 0; i < 15; i++) {
			//7.创建消息
			TextMessage textMessage=session.createTextMessage("我是消息生产者："+i);
			//8.发送消息
			producer.send(textMessage);
			System.out.println("发送消息："+i);
		}
		connection.close();
	}
}
 
运行结果
 
 
（2）创建消费者
 
package springboot.activeMQ;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class TipicConsumer {
	private static final String url="tcp://127.0.0.1:61616";//端口默认
	private static final String topicName="topic-test";//要消费的消息名称
	public static void main(String[] args) throws JMSException {
		//1.创建ConnectiongFactory,绑定地址
		ConnectionFactory factory=new ActiveMQConnectionFactory(url);
		//2.创建Connection
		Connection connection= factory.createConnection();
		//3.启动连接
		connection.start();
		//4.创建会话
		Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		//5.创建一个目标
		Destination destination=session.createTopic(topicName);
		//6.创建一个消费者
		MessageConsumer consumer=session.createConsumer(destination);
		//7.创建一个监听器
		consumer.setMessageListener(new MessageListener() {
			public void onMessage(Message arg0) {
				TextMessage textMessage=(TextMessage)arg0;
				try {
					System.out.println("接收消息："+textMessage.getText());
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		});
	}
}
 
运行结果
 
 
先运行消费者再运行生产者，消费者才能接收到
 
再次运行生产者后
 
 
再一次启动消费者后启动生产者，此时两个消费者接收的消息一样
 
 

 六、总结
 
（1）队列模式的消费者在运行后也能收到之前的消息，不过属于轮流进行消费，每个消费者接收不到完整的消息
 
（2）主题模式的消费者在运行后只能收到之后的消息，运行之前生产的消息接收不到；并且每个消费者都能接收全部的消息

无意中发现了一个巨牛的人工智能教程，忍不住分享一下给大家。教程不仅是零基础，通俗易懂，而且非常风趣幽默，像看小说一样！觉得太牛了，所以分享给大家。点这里可以跳转到教程。
 
 
 
ActiveMQ介绍
 
     MQ是消息中间件，是一种在分布式系统中应用程序借以传递消息的媒介，常用的有ActiveMQ，RabbitMQ，kafka。ActiveMQ是Apache下的开源项目，完全支持JMS1.1和J2EE1.4规范的JMS Provider实现。 
 特点： 
 1、支持多种语言编写客户端 
 2、对spring的支持，很容易和spring整合 
 3、支持多种传输协议：TCP,SSL,NIO,UDP等 
 4、支持AJAX 
 消息形式： 
 1、点对点（queue） 
 2、一对多（topic） 
 
ActiveMQ安装
 

 我这里提供一个安装好的虚拟机：http://download.csdn.net/download/liuyuanq123/10217892 
 服务器运行后，我们可以直接访问到activeMQ的界面： 

 然后点击queues可以看到现在没有一条消息： 
 
ActiveMQ测试
 

       编写一个测试类对ActiveMQ进行测试，首先得向pom文件中添加ActiveMQ相关的jar包：
 
     <dependency>  
         <groupId>org.apache.activemq</groupId>  
         <artifactId>activemq-all</artifactId>  
    </dependency>  
 
queue的发送代码如下:
 
    public void testMQProducerQueue() throws Exception{
        //1、创建工厂连接对象，需要制定ip和端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
        //2、使用连接工厂创建一个连接对象
        Connection connection = connectionFactory.createConnection();
        //3、开启连接
        connection.start();
        //4、使用连接对象创建会话（session）对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
        Queue queue = session.createQueue("test-queue");
        //6、使用会话对象创建生产者对象
        MessageProducer producer = session.createProducer(queue);
        //7、使用会话对象创建一个消息对象
        TextMessage textMessage = session.createTextMessage("hello!test-queue");
        //8、发送消息
        producer.send(textMessage);
        //9、关闭资源
        producer.close();
        session.close();
        connection.close();
    }
 
接收代码：
 
    public void TestMQConsumerQueue() throws Exception{
        //1、创建工厂连接对象，需要制定ip和端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
        //2、使用连接工厂创建一个连接对象
        Connection connection = connectionFactory.createConnection();
        //3、开启连接
        connection.start();
        //4、使用连接对象创建会话（session）对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
        Queue queue = session.createQueue("test-queue");
        //6、使用会话对象创建生产者对象
        MessageConsumer consumer = session.createConsumer(queue);
        //7、向consumer对象中设置一个messageListener对象，用来接收消息
        consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                // TODO Auto-generated method stub
                if(message instanceof TextMessage){
                    TextMessage textMessage = (TextMessage)message;
                    try {
                        System.out.println(textMessage.getText());
                    } catch (JMSException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        });
        //8、程序等待接收用户消息
        System.in.read();
        //9、关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
 
然后当我们运行queue发送的时候可以看到队列里已经有一条消息了，但没有发送出去： 

 然后在运行queue 的接收端，可以看到消息已经发出了： 


 接着对topic进行测试，发送代码如下：
 
    public void TestTopicProducer() throws Exception{
        //1、创建工厂连接对象，需要制定ip和端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
        //2、使用连接工厂创建一个连接对象
        Connection connection = connectionFactory.createConnection();
        //3、开启连接
        connection.start();
        //4、使用连接对象创建会话（session）对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
        Topic topic = session.createTopic("test-topic");
        //6、使用会话对象创建生产者对象
        MessageProducer producer = session.createProducer(topic);
        //7、使用会话对象创建一个消息对象
        TextMessage textMessage = session.createTextMessage("hello!test-topic");
        //8、发送消息
        producer.send(textMessage);
        //9、关闭资源
        producer.close();
        session.close();
        connection.close();
    }
 
接收代码：
 
    public void TestTopicConsumer() throws Exception{
        //1、创建工厂连接对象，需要制定ip和端口号
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
        //2、使用连接工厂创建一个连接对象
        Connection connection = connectionFactory.createConnection();
        //3、开启连接
        connection.start();
        //4、使用连接对象创建会话（session）对象
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5、使用会话对象创建目标对象，包含queue和topic（一对一和一对多）
        Topic topic = session.createTopic("test-topic");
        //6、使用会话对象创建生产者对象
        MessageConsumer consumer = session.createConsumer(topic);
        //7、向consumer对象中设置一个messageListener对象，用来接收消息
        consumer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                // TODO Auto-generated method stub
                if(message instanceof TextMessage){
                    TextMessage textMessage = (TextMessage)message;
                    try {
                        System.out.println(textMessage.getText());
                    } catch (JMSException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        });
        //8、程序等待接收用户消息
        System.in.read();
        //9、关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
 
 
 
然后运行topic发送： 

 可以看到消息已经发送出去。再运行topic接收： 

 可以看到有了一个消费者，但是没有接收的消息，这是因为正常情况下我们的topic消息不会再服务器持久化，所以要先打开消费者，再打开生产者，这个时候我们再运行生产者发送一条消息看到消息已经接收到了： 

 
ActiveMQ整合spring及项目中运用
 

       activeMQ与spring看一整合到一起使用，除了添加ActiveMQ相关的jar包外，还需要添加spring的jar包：
 
    <dependency>  
        <groupId>org.springframework</groupId>  
        <artifactId>spring-context</artifactId>  
    </dependency>  
 
然后编写applicationContext-activemq.xml文件， 
 代码如下：
 
<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"  
    xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"  
    xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd  
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd  
    http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd  
    http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd  
    http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">  

    <!-- 配置能够产生connection的connectionfactory，由JMS对应的服务厂商提供 -->
    <bean id="tagertConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <constructor-arg name="brokerURL" value="tcp://192.168.156.44:61616"/>
    </bean>
    <!-- 配置spring管理真正connectionfactory的connectionfactory，相当于spring对connectionfactory的一层封装 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="tagertConnectionFactory"/>
    </bean>
    <!-- 配置生产者 -->
    <!-- Spring使用JMS工具类，可以用来发送和接收消息 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <!-- 这里是配置的spring用来管理connectionfactory的connectionfactory -->
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>
    <!-- 配置destination -->
    <!-- 队列目的地 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="spring-queue"/>
    </bean>
    <!-- 话题目的地 -->
    <bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="item-add-topic"/>
    </bean>
</beans>  
 
然后在我们淘淘商城中，商品添加到数据库的时候，对应也要添加数据到我们的solr索引中，所以生产者应该在插入数据后创建： 

 当然，在xml文件中配置好的jmstemplate和destination也要注入进来：
 
    @Autowired
    private JmsTemplate jmsTemplate;
    @Resource(name="itemAddTopic")
    private Destination destination;
 
然后消费者应该写在我们的搜索工程中，首先添加spring和activeMQ的jar包，然后配置xml文件，再编写一个监听器，当接收到消息时，就讲数据存入索引库，xml文件代码如下：
 
<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"  
    xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"  
    xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd  
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd  
    http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.2.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.2.xsd  
    http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd  
    http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">  

    <!-- 配置能够产生connection的connectionfactory，由JMS对应的服务厂商提供 -->
    <bean id="tagertConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <constructor-arg name="brokerURL" value="tcp://192.168.156.44:61616"/>
    </bean>
    <!-- 配置spring管理真正connectionfactory的connectionfactory，相当于spring对connectionfactory的一层封装 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="tagertConnectionFactory"/>
    </bean>
    <!-- 配置destination -->
    <!-- 队列目的地 -->
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="spring-queue"/>
    </bean>
    <!-- 话题目的地 -->
    <bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="item-add-topic"/>
    </bean>
    <!-- 配置监听器 -->
    <bean id="myListener" class="com.taotao.search.listener.MyListener"/>
    <bean id="itemAddListener" class="com.taotao.search.listener.ItemAddListener"/>
    <!-- 系统监听器 -->
<!--    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="queueDestination"/>
        <property name="messageListener" ref="myListener"/>
    </bean> -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="itemAddTopic"/>
        <property name="messageListener" ref="itemAddListener"/>
    </bean>
</beans>  
 
接收消息代码： 

 最后同时打开测试即可。 
 
 
 
本文内容转自：https://blog.csdn.net/liuyuanq123/article/details/79109218