从零开始学 Java - Spring 集成 ActiveMQ 配置(二)

从上一篇开始说起 上一篇 从零开始学 Java - Spring 集成 ActiveMQ 配置(一) 文章中讲了我关于消息队列的思考过程,现在这一

从上一篇开始说起

上一篇 从零开始学 Java - Spring 集成 ActiveMQ 配置(一) 文章中讲了我关于消息队列的思考过程,现在这一篇会讲到 ActivMQ 与 Spring 框架的整合配置,以及我们的多队列配置、断线重连机制,当然还有的生产者及消费监听者的具体代码实现。

ActiveMQ 是 Apache 出品,最流行的、能力强劲的开源消息总线。ActiveMQ 是一个完全支持 JMS1.1 和 J2EE 1.4 规范的 JMS Provider 实现,可以很容易内嵌到使用Spring的系统里面去,所以我们选择它。

一些基础的准备工作

先去官网下载服务器端的安装包,地址: http://activemq.apache.org/activemq-5130-release.html ,根据自己系统下来,并进行安装,这里不再赘述,安装完成后,一般可以打开一个这样的界面:

从零开始学 Java - Spring 集成 ActiveMQ 配置(二)

接着,找到我们需要的 jar 包。其实刚刚下载的文件中就有,我们只需要导入到我们的项目中即可,当然,你也可以从我的示例项目中获得: https://github.com/mafly/SpringDemo/blob/master/WebContent/WEB-INF/lib/

开始写代码吧

一、建立 cn.mayongfa.activemq 包。

1.新建 MessageSender.java 消息发送类

/** * ActiveMQ 消息生产类 * * @author Mafly * */@Componentpublic class MessageSender {private Logger log = Logger.getLogger(MessageSender.class);@Autowiredprivate JmsTemplate jmsTemplate;private String Queue = "default_queue";private String GoldQueue = "gold_queue";private Gson gson = new Gson();/** * 用户登录消息 */public void userLogin(long id, String username) { Map<String, Object> map = new HashMap<String, Object>(); map.put("userid", id); map.put("username", username); System.out.println("发送了一条消息。"); // 发送到金币队列 sendMessage(gson.toJson(map), 1);}/** * 发送到消息队列 * * @param messgae * @param type * 类型,0:默认队列 1:金币队列 ... */public void sendMessage(final String messgae, int type) { try { String destination = this.Queue; if (type == 1) { destination = GoldQueue; } jmsTemplate.send(destination, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { TextMessage textMessage = session.createTextMessage(messgae); return textMessage; } }); } catch (Exception e) { log.error("", e); }}}

这个类是用来往消息队列发送信息的。

2.新建 ConsumerMessageListener.java 消息监听类,继承于 MessageListener

/** * 消费者监听类 * * @author Mafly */@Componentpublic class ConsumerMessageListener implements MessageListener {private Logger log = Logger.getLogger(ConsumerMessageListener.class);@Overridepublic void onMessage(Message arg0) { // 监听发送到消息队列的文本消息,作强制转换。 TextMessage textMessage = (TextMessage) arg0; try { System.out.println("接收到的消息内容是:" + textMessage.getText()); // TODO: 你喜欢的任何事情... } catch (JMSException e) { log.error("", e); }}}

这个类就是具体的消息消费类,我们就可以在这实现具体的业务逻辑处理了。

3.新建 ActiveMQTransportListener.java 消息传输监听类,继承于 TransportListener

/** * 消息传输监听 * @author Mafly * */public class ActiveMQTransportListener implements TransportListener {private Logger log = Logger.getLogger(ActiveMQTransportListener.class);/** * 对消息传输命令进行监控 * @param command */ @Override public void onCommand(Object o) {} /** * 对监控到的异常进行触发 * @param error */ @Override public void onException(IOException error) { log.error("onException -> 消息服务器连接错误......", error);} /** * 当failover时触发 */ @Override public void transportInterupted() { log.warn("transportInterupted -> 消息服务器连接发生中断..."); //这里就可以状态进行标识了} /** * 监控到failover恢复后进行触发 */ @Override public void transportResumed() { log.info("transportResumed -> 消息服务器连接已恢复..."); //这里就可以进行状态标识了}}

这里就可以做消息传输过程中的一些监控及处理,比如常见的异常处理等等。

二、在 applicationContext.xml 文件中配置

重要的是建完包之后要在 applicationContext.xml 文件中配置扫描包,完成 Bean 的注入 。就是下面:

<context:component-scan base-package="cn.mayongfa.activemq" />

applicationContext.xml 文件中加入 JMS 的配置:

<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 --><bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <property name="connectionFactory" ref="connectionFactory" /></bean><!-- 真正可以产生Connection的ConnectionFactory,由ActiveMQ提供 --><bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="failover:tcp://127.0.0.1:61616" /> <!-- 消息传输监听器 处理网络及服务器异常 --> <property name="transportListener"> <bean class="cn.mayongfa.activemq.ActiveMQTransportListener" /> </property></bean><!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --><bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> <property name="targetConnectionFactory" ref="targetConnectionFactory" /></bean><!--这个是队列目的地 --><bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg> <value>default_queue,gold_queue</value> </constructor-arg></bean><!-- 消息监听器 --><bean id="consumerMessageListener" class="cn.mayongfa.activemq.ConsumerMessageListener" /><!-- 消息监听容器 --><bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="connectionFactory" /> <property name="destination" ref="queueDestination" /> <property name="messageListener" ref="consumerMessageListener" /> <!-- 设置固定的线程数 --> <property name="concurrentConsumers" value="2"></property> <!-- 设置动态的线程数 --> <property name="concurrency" value="2-5"></property></bean>

这里的消息服务器地址及队列名称也可以像缓存一样放在配置文件。当然,你不能忘了把配置文件读取到 Spring 容器中管理。

  • queueDestination:这个节点下面的 default_queue,gold_queue 这就是两个队列的配置了,多队列在这里用逗号分隔配置就可以了。
  • failover:从字面意思你应该能明白这个意思,就是启用断线重连机制的。我们也是可以在 ActiveMQTransportListener 类里边监听到的。

配置完成,代码也写完了,接下来就是调用测试了。

从零开始学 Java - Spring 集成 ActiveMQ 配置(二)

总结一下

这篇文章有了干货,用具体的代码及配置完成了简单的消息队列的发送、监听等等,还有多队列的一些处理,以及断线重连机制。希望我的分享对你有所帮助吧,所有这些配置及代码都可以在我的 GitHub 上关于 Spring 的示例项目看到: https://github.com/mafly/SpringDemo/tree/activemq

未登录用户
全部评论0
到底啦