干货 | 从零开始学 Java - Spring 集成 ActiveMQ 配置

Posted 2021-04-24 e安在线

从上一篇开始说起

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

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

一些基础的准备工作

接着，找到我们需要的 jar 包。其实刚刚下载的文件中就有，我们只需要导入到我们的项目中即可，当然，你也可以从我的示例项目中获得：

开始写代码吧

一、建立cn.mayongfa.activemq包。
        1.新建MessageSender.java消息发送类

/** * ActiveMQ 消息生产类 * * @author Mafly * */
@Component
public class MessageSender {

private Logger log = Logger.getLogger(MessageSender.class);

@Autowired
private 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 */
@Component
public class ConsumerMessageListener implements MessageListener {

private Logger log = Logger.getLogger(ConsumerMessageListener.class);

@Override
public 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>

• queueDestination：这个节点下面的default_queue,gold_queue这就是两个队列的配置了，多队列在这里用逗号分隔配置就可以了。

• failover：从字面意思你应该能明白这个意思，就是启用断线重连机制的。我们也是可以在ActiveMQTransportListener类里边监听到的。

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

总结一下

这篇文章有了干货，用具体的代码及配置完成了简单的消息队列的发送、监听等等，还有多队列的一些处理，以及断线重连机制。

文章出处：http://www.cnblogs.com/mafly/p/activemq.html