ActiveMQ消息队列使用和配置
Posted oldboyooxx
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了ActiveMQ消息队列使用和配置相关的知识,希望对你有一定的参考价值。
安装ActiveMQ
使用brew安装https://blog.csdn.net/u010046908/article/details/54728375
直接下载安装https://blog.csdn.net/ytangdigl/article/details/77740100
启动
activemq start
帐号密码都是admin
默认端口8161
搭建项目框架
这边拿一个简易的spring+springMVC为框架
添加pom依赖
<!--activity工作流依赖--> <dependency> <groupId>org.activiti</groupId> <artifactId>activiti-engine</artifactId> <version>${activiti.version}</version> </dependency> <!-- activiti 与 Spring 集成 --> <dependency> <groupId>org.activiti</groupId> <artifactId>activiti-spring</artifactId> <version>${activiti.version}</version> </dependency> <!--activeMq依赖包--> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-all</artifactId> <version>5.13.2</version> </dependency> <!--spring和mq的依赖包--> <!-- https://mvnrepository.com/artifact/org.springframework/spring-jms --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>4.3.8.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-messaging</artifactId> <version>4.3.8.RELEASE</version> </dependency>
添加spring-config.xml中的schema
xmlns:amq="http://activemq.apache.org/schema/core" xmlns:jms="http://www.springframework.org/schema/jms" http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
添加配置
<!--mq配置--> <!--mq的消息中心,brokerURL中心的位置--> <amq:connectionFactory brokerURL="tcp://localhost:61616" userName="admin" password="admin" id="amqconnectionFactory"/> <!--spring和jms的连接targetConnectionFactory是写死的--> <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <property name="targetConnectionFactory" ref="amqconnectionFactory"/> <!--缓存大小--> <property name="sessionCacheSize" value="10"/> </bean> <!--配置两种模式--> <!--点对点模式--> <!--发布订阅--> <!--id为jmsTemplate,可能id会重复--> <bean id="queueTemplate" class="org.springframework.jms.core.JmsTemplate"> <constructor-arg ref="connectionFactory"/> <!--不接收 pubSubDomain:是不是队列--> <property name="pubSubDomain" value="false"/> <!--消息不会持久化了--> <!--<property name="deliveryMode" value="1"/>--> </bean> <!--订阅者--> <bean id="topicTemplate" class="org.springframework.jms.core.JmsTemplate"> <constructor-arg ref="connectionFactory"/> <!--接收--> <property name="pubSubDomain" value="true"/> </bean> <!--监听--> <!--acknowledge消息课靠性传输,属性auto默认自动确认机制,--> <!-- prefetch="100" 传输次数,和幂有关系--> <!--destination-type=urableTopic可以设置持久化和非持久化--> <!--接收信息queue点对点--> <jms:listener-container destination-type="queue" connection-factory="connectionFactory"> <!-- destination监听的对列--> <jms:listener destination="oldboy.queue" ref="queueReceicer_one"/> <jms:listener destination="oldboy.queue" ref="queueReceicer_two"/> </jms:listener-container> <!--发送信息queue发布订阅--> <jms:listener-container destination-type="topic" connection-factory="connectionFactory"> <!--destination监听的对列--> <jms:listener destination="oldboy.topic" ref="topicReceicer_one"/> <jms:listener destination="oldboy.topic" ref="topicReceicer_two"/> </jms:listener-container>
创建和配置文件对应的类
P2P模式(点point对点point)
QueueReceicer_one.java
import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; /** * Created by peng on 18/6/26. */ @Component("queueReceicer_one") public class QueueReceicer_one implements MessageListener { @Override public void onMessage(Message message) { try { System.out.println("queueReceicer_one:"+((TextMessage) message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }
QueueReceicer_two.java
import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; /** * Created by peng on 18/6/26. */ @Component("queueReceicer_two") public class QueueReceicer_two implements MessageListener{ @Override public void onMessage(Message message) { try { System.out.println("queueReceicer_two:"+((TextMessage)message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }
QueueSender.java
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; /** * Created by peng on 18/6/26. */ @Component public class QueueSender { @Autowired @Qualifier("queueTemplate") private JmsTemplate template; public void send(String queueName , final String message){ template.send(queueName, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { return session.createTextMessage(message); } }); } }
发布订阅模式
TopicReceicer_one.java
import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; /** * Created by peng on 18/6/26. */ @Component("topicReceicer_one") public class TopicReceicer_one implements MessageListener { @Override public void onMessage(Message message) { try { System.out.println("topicReceicer_one:"+((TextMessage)message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }
TopicReceicer_two.java
import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; /** * Created by peng on 18/6/26. */ @Component("topicReceicer_two") public class TopicReceicer_two implements MessageListener { @Override public void onMessage(Message message) { try { System.out.println("topicReceicer_two:"+((TextMessage)message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }
TopicSender.java
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; import org.springframework.stereotype.Component; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; /** * Created by peng on 18/6/26. */ @Component public class TopicSender { @Autowired @Qualifier("topicTemplate") private JmsTemplate template; public void send(String queueName , final String message){ template.send(queueName, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { return session.createTextMessage(message); } }); } }
调用
MqTest.java
import oldboy.vip.controller.service.QueueSender; import oldboy.vip.controller.service.TopicSender; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody; /** * Created by peng on 18/6/26. */ @Controller @RequestMapping("mq") public class MqTest { @Autowired private QueueSender queueSender; @Autowired private TopicSender topicSender; @RequestMapping("/test1") @ResponseBody public String test1(){ queueSender.send("oldboy.queue","oldboy");//oldboy.queue和配置文件中destination对应 topicSender.send("oldboy.topic","oldboy");//oldboy.topic和配置文件中destination对应 return "oldboy.vip"; } }
最终目录结构
启动项目
控制台接收到消息
点对点的只能收一个人收到
消息订阅的可以所有人收到
可以在acitveMQ中看到(可以查看12是因为测试的次数比较多。。)
代码已经扔到github上了https://github.com/oldboyooxx/activeMQDemo对你有帮助喜欢点个喜欢~
以上是关于ActiveMQ消息队列使用和配置的主要内容,如果未能解决你的问题,请参考以下文章
ActiveMQ:使用队列(具有并发消费者)和主题的正确配置
WSO2 ESB 5.0.0 配置 JMS 传输(ActiveMQ)- 队列消息生产与消费
WSO2 ESB 5.0.0 配置 JMS 传输(ActiveMQ)- 队列消息生产与消费