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安装

ActiveMQ详细入门使用教程-LMLPHP
我这里提供一个安装好的虚拟机:http://download.csdn.net/download/liuyuanq123/10217892 
服务器运行后,我们可以直接访问到activeMQ的界面: 
ActiveMQ详细入门使用教程-LMLPHP
然后点击queues可以看到现在没有一条消息: 
ActiveMQ详细入门使用教程-LMLPHP

ActiveMQ测试


      编写一个测试类对ActiveMQ进行测试,首先得向pom文件中添加ActiveMQ相关的jar包:

  1. <dependency>
  2. <groupId>org.apache.activemq</groupId>
  3. <artifactId>activemq-all</artifactId>
  4. </dependency>
  • 1
  • 2
  • 3
  • 4
  • 1
  • 2
  • 3
  • 4

queue的发送代码如下:

  1. public void testMQProducerQueue() throws Exception{
  2. //1、创建工厂连接对象,需要制定ip和端口号
  3. ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
  4. //2、使用连接工厂创建一个连接对象
  5. Connection connection = connectionFactory.createConnection();
  6. //3、开启连接
  7. connection.start();
  8. //4、使用连接对象创建会话(session)对象
  9. Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  10. //5、使用会话对象创建目标对象,包含queue和topic(一对一和一对多)
  11. Queue queue = session.createQueue("test-queue");
  12. //6、使用会话对象创建生产者对象
  13. MessageProducer producer = session.createProducer(queue);
  14. //7、使用会话对象创建一个消息对象
  15. TextMessage textMessage = session.createTextMessage("hello!test-queue");
  16. //8、发送消息
  17. producer.send(textMessage);
  18. //9、关闭资源
  19. producer.close();
  20. session.close();
  21. connection.close();
  22. }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

接收代码:

  1. public void TestMQConsumerQueue() throws Exception{
  2. //1、创建工厂连接对象,需要制定ip和端口号
  3. ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
  4. //2、使用连接工厂创建一个连接对象
  5. Connection connection = connectionFactory.createConnection();
  6. //3、开启连接
  7. connection.start();
  8. //4、使用连接对象创建会话(session)对象
  9. Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  10. //5、使用会话对象创建目标对象,包含queue和topic(一对一和一对多)
  11. Queue queue = session.createQueue("test-queue");
  12. //6、使用会话对象创建生产者对象
  13. MessageConsumer consumer = session.createConsumer(queue);
  14. //7、向consumer对象中设置一个messageListener对象,用来接收消息
  15. consumer.setMessageListener(new MessageListener() {
  16. @Override
  17. public void onMessage(Message message) {
  18. // TODO Auto-generated method stub
  19. if(message instanceof TextMessage){
  20. TextMessage textMessage = (TextMessage)message;
  21. try {
  22. System.out.println(textMessage.getText());
  23. } catch (JMSException e) {
  24. // TODO Auto-generated catch block
  25. e.printStackTrace();
  26. }
  27. }
  28. }
  29. });
  30. //8、程序等待接收用户消息
  31. System.in.read();
  32. //9、关闭资源
  33. consumer.close();
  34. session.close();
  35. connection.close();
  36. }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

然后当我们运行queue发送的时候可以看到队列里已经有一条消息了,但没有发送出去: 
ActiveMQ详细入门使用教程-LMLPHP
然后在运行queue 的接收端,可以看到消息已经发出了: 
ActiveMQ详细入门使用教程-LMLPHP
ActiveMQ详细入门使用教程-LMLPHP
接着对topic进行测试,发送代码如下:

  1. public void TestTopicProducer() throws Exception{
  2. //1、创建工厂连接对象,需要制定ip和端口号
  3. ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
  4. //2、使用连接工厂创建一个连接对象
  5. Connection connection = connectionFactory.createConnection();
  6. //3、开启连接
  7. connection.start();
  8. //4、使用连接对象创建会话(session)对象
  9. Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  10. //5、使用会话对象创建目标对象,包含queue和topic(一对一和一对多)
  11. Topic topic = session.createTopic("test-topic");
  12. //6、使用会话对象创建生产者对象
  13. MessageProducer producer = session.createProducer(topic);
  14. //7、使用会话对象创建一个消息对象
  15. TextMessage textMessage = session.createTextMessage("hello!test-topic");
  16. //8、发送消息
  17. producer.send(textMessage);
  18. //9、关闭资源
  19. producer.close();
  20. session.close();
  21. connection.close();
  22. }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

接收代码:

  1. public void TestTopicConsumer() throws Exception{
  2. //1、创建工厂连接对象,需要制定ip和端口号
  3. ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.156.44:61616");
  4. //2、使用连接工厂创建一个连接对象
  5. Connection connection = connectionFactory.createConnection();
  6. //3、开启连接
  7. connection.start();
  8. //4、使用连接对象创建会话(session)对象
  9. Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  10. //5、使用会话对象创建目标对象,包含queue和topic(一对一和一对多)
  11. Topic topic = session.createTopic("test-topic");
  12. //6、使用会话对象创建生产者对象
  13. MessageConsumer consumer = session.createConsumer(topic);
  14. //7、向consumer对象中设置一个messageListener对象,用来接收消息
  15. consumer.setMessageListener(new MessageListener() {
  16. @Override
  17. public void onMessage(Message message) {
  18. // TODO Auto-generated method stub
  19. if(message instanceof TextMessage){
  20. TextMessage textMessage = (TextMessage)message;
  21. try {
  22. System.out.println(textMessage.getText());
  23. } catch (JMSException e) {
  24. // TODO Auto-generated catch block
  25. e.printStackTrace();
  26. }
  27. }
  28. }
  29. });
  30. //8、程序等待接收用户消息
  31. System.in.read();
  32. //9、关闭资源
  33. consumer.close();
  34. session.close();
  35. connection.close();
  36. }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

然后运行topic发送: 
ActiveMQ详细入门使用教程-LMLPHP
可以看到消息已经发送出去。再运行topic接收: 
ActiveMQ详细入门使用教程-LMLPHP
可以看到有了一个消费者,但是没有接收的消息,这是因为正常情况下我们的topic消息不会再服务器持久化,所以要先打开消费者,再打开生产者,这个时候我们再运行生产者发送一条消息看到消息已经接收到了: 
ActiveMQ详细入门使用教程-LMLPHP
ActiveMQ详细入门使用教程-LMLPHP

ActiveMQ整合spring及项目中运用


      activeMQ与spring看一整合到一起使用,除了添加ActiveMQ相关的jar包外,还需要添加spring的jar包:

  1. <dependency>
  2. <groupId>org.springframework</groupId>
  3. <artifactId>spring-context</artifactId>
  4. </dependency>
  • 1
  • 2
  • 3
  • 4
  • 1
  • 2
  • 3
  • 4

然后编写applicationContext-activemq.xml文件, 
代码如下:

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3. xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
  4. xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
  5. xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
  6. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  7. xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
  8. http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd
  9. 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
  10. http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd
  11. http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">
  12. <!-- 配置能够产生connection的connectionfactory,由JMS对应的服务厂商提供 -->
  13. <bean id="tagertConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  14. <constructor-arg name="brokerURL" value="tcp://192.168.156.44:61616"/>
  15. </bean>
  16. <!-- 配置spring管理真正connectionfactory的connectionfactory,相当于spring对connectionfactory的一层封装 -->
  17. <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
  18. <property name="targetConnectionFactory" ref="tagertConnectionFactory"/>
  19. </bean>
  20. <!-- 配置生产者 -->
  21. <!-- Spring使用JMS工具类,可以用来发送和接收消息 -->
  22. <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  23. <!-- 这里是配置的spring用来管理connectionfactory的connectionfactory -->
  24. <property name="connectionFactory" ref="connectionFactory"/>
  25. </bean>
  26. <!-- 配置destination -->
  27. <!-- 队列目的地 -->
  28. <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
  29. <constructor-arg value="spring-queue"/>
  30. </bean>
  31. <!-- 话题目的地 -->
  32. <bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic">
  33. <constructor-arg value="item-add-topic"/>
  34. </bean>
  35. </beans>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36

然后在我们淘淘商城中,商品添加到数据库的时候,对应也要添加数据到我们的solr索引中,所以生产者应该在插入数据后创建: 
ActiveMQ详细入门使用教程-LMLPHP
当然,在xml文件中配置好的jmstemplate和destination也要注入进来:

  1. @Autowired
  2. private JmsTemplate jmsTemplate;
  3. @Resource(name="itemAddTopic")
  4. private Destination destination;
  • 1
  • 2
  • 3
  • 4
  • 1
  • 2
  • 3
  • 4

然后消费者应该写在我们的搜索工程中,首先添加spring和activeMQ的jar包,然后配置xml文件,再编写一个监听器,当接收到消息时,就讲数据存入索引库,xml文件代码如下:

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3. xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
  4. xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
  5. xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
  6. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  7. xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd
  8. http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd
  9. 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
  10. http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd
  11. http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-4.2.xsd">
  12. <!-- 配置能够产生connection的connectionfactory,由JMS对应的服务厂商提供 -->
  13. <bean id="tagertConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  14. <constructor-arg name="brokerURL" value="tcp://192.168.156.44:61616"/>
  15. </bean>
  16. <!-- 配置spring管理真正connectionfactory的connectionfactory,相当于spring对connectionfactory的一层封装 -->
  17. <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
  18. <property name="targetConnectionFactory" ref="tagertConnectionFactory"/>
  19. </bean>
  20. <!-- 配置destination -->
  21. <!-- 队列目的地 -->
  22. <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
  23. <constructor-arg value="spring-queue"/>
  24. </bean>
  25. <!-- 话题目的地 -->
  26. <bean id="itemAddTopic" class="org.apache.activemq.command.ActiveMQTopic">
  27. <constructor-arg value="item-add-topic"/>
  28. </bean>
  29. <!-- 配置监听器 -->
  30. <bean id="myListener" class="com.taotao.search.listener.MyListener"/>
  31. <bean id="itemAddListener" class="com.taotao.search.listener.ItemAddListener"/>
  32. <!-- 系统监听器 -->
  33. <!-- <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  34. <property name="connectionFactory" ref="connectionFactory"/>
  35. <property name="destination" ref="queueDestination"/>
  36. <property name="messageListener" ref="myListener"/>
  37. </bean> -->
  38. <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  39. <property name="connectionFactory" ref="connectionFactory"/>
  40. <property name="destination" ref="itemAddTopic"/>
  41. <property name="messageListener" ref="itemAddListener"/>
  42. </bean>
  43. </beans>
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44

接收消息代码: 
ActiveMQ详细入门使用教程-LMLPHP
最后同时打开测试即可。 


本文内容转自:https://blog.csdn.net/liuyuanq123/article/details/79109218

10-06 20:12