• springboot内嵌ActiveMQ
  • 1、增加依赖
  • 2、增加MQ优化的配置
  • 3、启动activeMq的内嵌连接
  • 4、增加生产者
  • 5、增加消费者
  • 6、增加消息调用MQ
  • 7、查看是否调用成功

springboot内嵌ActiveMQ

ps: springboot 2.2.7

1、增加依赖

	org.springframework.boot
	spring-boot-starter-activemq



	org.messaginghub
	pooled-jms

2、增加MQ优化的配置

package com.platform.gis.common.config;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.SimpleMessageListenerContainer;

import javax.jms.ConnectionFactory;

@Configuration
public class ActiveMQConfig {

    @Bean
    public JmsListenerContainerFactory<?> queueListenerFactory() {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        SimpleMessageListenerContainer container=new SimpleMessageListenerContainer();
        container.setConcurrentConsumers(3);
        container.setConnectionFactory(connectionFactory());
        factory.setPubSubDomain(false);
        factory.setConnectionFactory(connectionFactory());
        //连接数
        factory.setConcurrency("3-15");
        //重连间隔时间
        factory.setRecoveryInterval(1000L);
        factory.setSessionAcknowledgeMode(4);
        return factory;
    }
    @Bean
    public ConnectionFactory connectionFactory() {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setTrustAllPackages(true);
        connectionFactory.setRedeliveryPolicy(redeliveryPolicy());
        return connectionFactory;
    }

    @Bean
    public RedeliveryPolicy redeliveryPolicy(){
        RedeliveryPolicy redeliveryPolicy=new RedeliveryPolicy();
        //是否在每次尝试重新发送失败后,增长这个等待时间
        redeliveryPolicy.setUseExponentialBackOff(true);
        //重发次数,默认为6次
        redeliveryPolicy.setMaximumRedeliveries(3);
        //重发时间间隔,默认为1秒
        redeliveryPolicy.setInitialRedeliveryDelay(1);
        //第一次失败后重新发送之前等待500毫秒,第二次失败再等待500 * 2毫秒,这里的2就是value
        redeliveryPolicy.setBackOffMultiplier(2);
        //是否避免消息碰撞
        redeliveryPolicy.setUseCollisionAvoidance(false);
        //设置重发最大拖延时间-1 表示没有拖延只有UseExponentialBackOff(true)为true时生效
        redeliveryPolicy.setMaximumRedeliveryDelay(-1);
        return redeliveryPolicy;
    }

    @Bean
    public JmsTemplate jmsTemplate(){
        JmsTemplate jmsTemplate=new JmsTemplate(connectionFactory());
        //进行持久化配置 1表示非持久化，2表示持久化
        jmsTemplate.setDeliveryMode(1);
        //jmsTemplate.setConnectionFactory(activeMQConnectionFactory);
        //客户端签收模式
        jmsTemplate.setSessionAcknowledgeMode(4);
        return jmsTemplate;
    }

    @Bean
    public JmsMessagingTemplate jmsMessageTemplate(){
        return new JmsMessagingTemplate(connectionFactory());
    }

}

3、启动activeMq的内嵌连接

@EnableJms

工具类启动mq服务

package com.platform.gis.common.util;

import org.apache.activemq.broker.BrokerService;

public class ActiveMQUtils {

    private static final String BROKERURL = "tcp://localhost:61616";

    public static void startBrokerService(){
        try {
            BrokerService broker = new BrokerService();
            broker.addConnector(BROKERURL);
            broker.setPersistent(false);
            broker.start();
        }catch (Exception e){

        }
    }
}

	public static void main(String[] args) {
		//启动内嵌Mq
		ActiveMQUtils.startBrokerService();
		SpringContextUtil.setApplicationContext(SpringApplication.run(AdminServiceApplication.class, args));
		
	}

4、增加生产者

package com.platform.gis.common.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.jms.Destination;

@Service
public class JmsProduce {

    @Autowired
    private JmsMessagingTemplate jmsTemplate;

    // 发送消息，destination是发送到的队列，message是待发送的消息
    public void sendMessage(Destination destination, final Object message) {
        jmsTemplate.convertAndSend(destination, message);
    }
}

5、增加消费者

package com.platform.gis.admin.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.platform.gis.admin.constant.SysConstanst;
import com.platform.gis.admin.dao.SysMessageDao;
import com.platform.gis.admin.log.po.SysInterfaceLog;
import com.platform.gis.admin.message.po.SysMessage;
import com.platform.gis.admin.message.vo.SysMessageVo;
import com.platform.gis.admin.service.SysLogService;
import com.platform.gis.admin.service.SysSendMessageService;
import com.platform.gis.common.constant.CommonConstant;
import com.platform.gis.common.util.Pages;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.util.HashMap;
import java.util.Map;

@Component
public class JmsConsumer {

    protected final Logger LOG = LoggerFactory.getLogger(getClass());

    @Autowired
    SysSendMessageService sysSendMessageService;

    @Autowired
    SysLogService sysLogService;

    @Autowired
    private SysMessageDao sysMessageDao;

    @Autowired
    protected RedisTemplate jsonRedisTemplate;

    /**
     * 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
     * 

     * 发送邮件队列
     */
    @JmsListener(destination = SysConstanst.MESSAGE_QUEUE_EMAIL)
    public void emailReceiveQueue(Message message, Session session) throws JMSException {
        SysMessageVo sysMessageVo = new SysMessageVo();
        String text = "";
        try {
            TextMessage msg = (TextMessage) message;
            text = msg.getText();
            JSONObject jsonObject = JSON.parseObject(text);
            sysMessageVo = JSON.toJavaObject(jsonObject, SysMessageVo.class);
            if (sysMessageVo != null) {
                sysSendMessageService.sendMail(sysMessageVo);
                sysMessageVo.setStatus("1");
            }
            //手动确认
            message.acknowledge();
        } catch (Exception e) {
            LOG.error(e.getLocalizedMessage(), e);
            if (sysMessageVo != null) {
                sysMessageVo.setStatus("2");
                sysMessageVo.setErrorReason(ExceptionUtils.getStackTrace(e));
            }
            //消息重发
            session.recover();
        } finally {
            if (sysMessageVo != null) {
                SysMessage sysMessage = Pages.convert(sysMessageVo, SysMessage.class);
                sysMessageDao.updateById(sysMessage);
            }
        }
    }

    /**
     * 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
     * 

     * 发送短信队列
     */
    @JmsListener(destination = SysConstanst.MESSAGE_QUEUE_SMS)
    public void smsReceiveQueue(Message message, Session session) throws JMSException {
        SysMessageVo sysMessageVo = new SysMessageVo();
        String text = "";
        Map msgResult = new HashMap();
        try {
            TextMessage msg = (TextMessage) message;
            text = msg.getText();
            JSONObject jsonObject = JSON.parseObject(text);
            sysMessageVo = JSON.toJavaObject(jsonObject, SysMessageVo.class);
            if (sysMessageVo != null) {
                msgResult = sysSendMessageService.sendSMS(sysMessageVo);
                JSONObject smsResult = JSONObject.parseObject(msgResult.get("result"));
                if ("SUC0000".equals(smsResult.get("rtcode").toString())) {//短信发送成功
                    sysMessageVo.setStatus("1");
                } else {
                    String errorMessage = smsResult.get("rtmsg").toString();
                    LOG.error("SMS failed to send,the reason is:" + errorMessage);
                    sysMessageVo.setErrorReason(errorMessage);
                }
            }
            //手动确认
            message.acknowledge();
        } catch (Exception e) {
            LOG.error(e.getLocalizedMessage(), e);
            if (sysMessageVo != null) {
                sysMessageVo.setStatus("2");
                sysMessageVo.setErrorReason(ExceptionUtils.getStackTrace(e));
            }
            //消息重发
            session.recover();
        } finally {
            if (sysMessageVo != null) {
                sysMessageVo.setContent(msgResult.get("sendJson"));
                SysMessage sysMessage = Pages.convert(sysMessageVo, SysMessage.class);
                sysMessageDao.updateById(sysMessage);
            }
        }
    }

    /**
     * 定时执行审计日志的保存
     */
    @Scheduled(fixedDelay = SysConstanst.SCHEDULED_AUDIT_LOG_TIME)
    public void auditLogQueue() {
        sysLogService.saveAuditLogToDb();
    }


    /**
     * 接口请求日志的保存
     */
    @JmsListener(destination = CommonConstant.MESSAGE_QUEUE_INTERFACE_LOG)
    public void interfaceLogQueue(Message message, Session session) throws JMSException {
        SysInterfaceLog sysInterfaceLog = null;
        String text = "";
        try {
            TextMessage msg = (TextMessage) message;
            text = msg.getText();
            JSONObject jsonObject = JSON.parseObject(text);
            sysInterfaceLog = JSON.toJavaObject(jsonObject, SysInterfaceLog.class);
            if (sysInterfaceLog != null) {
                sysLogService.saveInterfaceLog(sysInterfaceLog);
            }
            //手动确认
            message.acknowledge();
        } catch (Exception e) {
            LOG.error(e.getLocalizedMessage(), e);
            //消息重发
            session.recover();
        }
    }

}

6、增加消息调用MQ

注入

@Autowired
private JmsProduce producer;

增加管道（生产不同的queue：不同的queue相当于不同的消费逻辑）

Destination destination = destination = new ActiveMQQueue(SysConstanst.MESSAGE_QUEUE_EMAIL);
producer.sendMessage(destination, JSON.toJSONString(sysMessageVo));

7、查看是否调用成功