自定义对rocketmq生产者以及消费者的封装

一. 生产者抽象类 :

package com.canxiusi.common.service.customize_mq.producer;

import com.canxiusi.common.service.customize_mq.constant.MqConstant;
import com.canxiusi.common.service.exception.SystemException;
import com.canxiusi.common.service.exception.enums.CommonExceptionEnum;
import com.canxiusi.common.service.log4j2.spi.Logger;
import com.canxiusi.common.service.log4j2.spi.LoggerFactory;
import com.canxiusi.common.service.utils.LogUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;


public abstract class AbstractSysLogProducer {

    
    private static final Logger logger = LoggerFactory.getLogger(AbstractSysLogProducer.class);

    private DefaultMQProducer sysLogProducer;

    
    @PostConstruct
    public void initProducer() {
        sysLogProducer = new DefaultMQProducer(getProducerGroup());
        sysLogProducer.setNamesrvAddr(getNameServer());
        sysLogProducer.setInstanceName(getInstanceName());
        try {
            sysLogProducer.start();
            logger.info(LogUtils.format("[SysLog生产者初始化完成], producer=<{0}>", sysLogProducer));
        } catch (MQClientException e) {
            logger.error("[SysLog生产者初始化失败]", e);
            System.exit(-1);
        }
    }

    @PreDestroy
    public void closeProducer() {
        sysLogProducer.shutdown();
    }

    
    public void sendMsgByteOneway(byte[] bytes) {
        if (bytes.length > MqConstant.MAX_MB) {
            logger.error(LogUtils.format("[发送消息过大, 无法发送!], msg=<{0}>", new String(bytes)));
            throw new SystemException(CommonExceptionEnum.PARAM_ERROR, "[sendMsgByteOneway] 发送消息大于4MB!");
        }
        Message message = new Message(getTopic(), getTag(), bytes);
        try {
            sysLogProducer.sendOneway(message);
        } catch (MQClientException | RemotingException | InterruptedException e) {
            logger.error(LogUtils.format("线程<{0}>, 生产者=<{1}>发送单向消息出错",
                    Thread.currentThread().getName(), this.getClass().getSimpleName()), e);
        }
    }

    
    protected abstract String getNameServer();

    
    protected abstract String getProducerGroup();

    
    protected abstract String getInstanceName();

    
    protected abstract String getTopic();

    
    protected abstract String getTag();

}

二. 生产者子类实现:

package com.canxiusi.web.online.producer;

import com.canxiusi.common.service.customize_mq.config.SysLogMqConfig;
import com.canxiusi.common.service.customize_mq.producer.AbstractSysLogProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component
public class SysLogProducer extends AbstractSysLogProducer {

    @Autowired
    private SysLogMqConfig sysLogMqConfig;

    @Override
    protected String getNameServer() {
        return sysLogMqConfig.getNameServerAddress();
    }

    @Override
    protected String getProducerGroup() {
        return sysLogMqConfig.getProducerGroup();
    }

    @Override
    protected String getInstanceName() {
        return sysLogMqConfig.getProducerInstance();
    }

    @Override
    protected String getTopic() {
        return sysLogMqConfig.getTopic();
    }

    @Override
    protected String getTag() {
        return sysLogMqConfig.getTag();
    }

}

三. 配置文件bean:

package com.canxiusi.common.service.customize_mq.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;


@Data
@Configuration
@PropertySource("classpath:syslogmq.properties")
@ConfigurationProperties(prefix = "syslog")
public class SysLogMqConfig {

    private String nameServerAddress;

    private String producerGroup;

    private String producerInstance;

    private String consumerGroup;

    private String consumerDbGroup;

    private String consumerMongoGroup;

    private String consumerDbInstance;

    private String consumerMongoInstance;

    private String topic;

    private String tag;

    private String dbTopic;

    private String mongoTag;

}

四. 消费者抽象实现:

package com.canxiusi.common.service.customize_mq.consumer;

import com.canxiusi.common.service.entity.logs.SysOperateLog;
import com.canxiusi.common.service.log4j2.spi.Logger;
import com.canxiusi.common.service.log4j2.spi.LoggerFactory;
import com.canxiusi.common.service.utils.JsonUtil;
import com.canxiusi.common.service.utils.LogUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.boot.context.event.ApplicationPreparedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;


public abstract class AbstractSysLogConsumer {

    
    protected static final Logger logger = LoggerFactory.getLogger(AbstractSysLogConsumer.class);

    private DefaultMQPushConsumer consumer;

    
    @PostConstruct
    public void initConsumer() {
        consumer = new DefaultMQPushConsumer(getConsumerGroup());
        consumer.setNamesrvAddr(getNameServer());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.setInstanceName(getInstanceName());
        consumer.setPullInterval(2000);
        consumer.setPullBatchSize(500);
        try {
            consumer.subscribe(getTopic(), getTag());
        } catch (MQClientException e) {
            logger.error("[SysLog消费者初始化异常]", e);
            System.exit(-1);
        }
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            if (CollectionUtils.isEmpty(msgs)) {
                logger.info(LogUtils.format("[消息列表为空], msgs=<{0}>", msgs));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            for (MessageExt msg : msgs) {
                try {
                    handlerMsg(msg);
                } catch (Exception e) {
                    logger.error("[日志消费处理异常]", e);
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        logger.info(LogUtils.format("[SysLog消费者初始化完成], consumer=<{0}>", consumer));
    }

    
    @EventListener(ApplicationPreparedEvent.class)
    public void startSysLogConsumer(ApplicationPreparedEvent event) {
        try {
            consumer.start();
        } catch (MQClientException e) {
            logger.error(LogUtils.format("[SysLog消费者start异常], nameServerAddress={}", getNameServer()), e);
            System.exit(-1);
        }
    }

    @PreDestroy
    public void closeConsumer() {
        consumer.shutdown();
    }

    
    protected final void handlerMsg(MessageExt msg) {
        SysOperateLog sysOperateLog;
        try {
            sysOperateLog = JsonUtil.readValue(msg.getBody(), SysOperateLog.class);
        } catch (Exception e) {
            logger.error("[字节消息转换出错]", e);
            throw e;
        }
        handlerLogs(sysOperateLog);
    }

    
    protected abstract void handlerLogs(SysOperateLog sysOperateLog);

    
    protected abstract String getNameServer();

    
    protected abstract String getConsumerGroup();

    
    protected abstract String getInstanceName();

    
    protected abstract String getTopic();

    
    protected abstract String getTag();

}

五. 消费者子类实现:

5.1 模拟mysql消费者

package com.canxiusi.web.service.consumer;

import com.canxiusi.common.service.customize_mq.config.SysLogMqConfig;
import com.canxiusi.common.service.customize_mq.consumer.AbstractSysLogConsumer;
import com.canxiusi.common.service.entity.logs.SysOperateLog;
import com.canxiusi.common.service.log4j2.spi.Logger;
import com.canxiusi.common.service.log4j2.spi.LoggerFactory;
import com.canxiusi.common.service.utils.LogUtils;
import com.canxiusi.web.service.mapper.operate.SysOperateLogMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component
public class SysLogDbConsumer extends AbstractSysLogConsumer {

    
    private static final Logger logger = LoggerFactory.getLogger(SysLogDbConsumer.class);

    @Autowired
    private SysLogMqConfig sysLogMqConfig;

    @Autowired
    private SysOperateLogMapper operateLogMapper;

    @Override
    protected void handlerLogs(SysOperateLog sysOperateLog) {
        logger.info(LogUtils.format("当前线程<{0}>, <{1}>消费了消息, sysLog=<{2}>",
                Thread.currentThread().getName(), this.getClass().getSimpleName(), sysOperateLog));
        operateLogMapper.insertSelective(sysOperateLog);
    }

    @Override
    protected String getNameServer() {
        return sysLogMqConfig.getNameServerAddress();
    }

    @Override
    protected String getConsumerGroup() {
        return sysLogMqConfig.getConsumerGroup();
    }

    @Override
    protected String getInstanceName() {
        return sysLogMqConfig.getConsumerDbInstance();
    }

    @Override
    protected String getTopic() {
        return sysLogMqConfig.getTopic();
    }

    @Override
    protected String getTag() {
        return sysLogMqConfig.getTag();
    }

}

5.2 模拟mongo消费者:

package com.canxiusi.web.service.consumer;

import com.canxiusi.common.service.customize_mq.config.SysLogMqConfig;
import com.canxiusi.common.service.customize_mq.consumer.AbstractSysLogConsumer;
import com.canxiusi.common.service.entity.logs.SysOperateLog;
import com.canxiusi.common.service.log4j2.spi.Logger;
import com.canxiusi.common.service.log4j2.spi.LoggerFactory;
import com.canxiusi.common.service.utils.LogUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


@Component
public class SysLogMongoConsumer extends AbstractSysLogConsumer {

    private static final Logger logger = LoggerFactory.getLogger(SysLogMongoConsumer.class);

    @Autowired
    private SysLogMqConfig sysLogMqConfig;

    @Override
    protected void handlerLogs(SysOperateLog sysOperateLog) {
        logger.info(LogUtils.format("当前线程<{0}>, <{1}>消费了消息, sysLog=<{2}>",
                Thread.currentThread().getName(), this.getClass().getSimpleName(), sysOperateLog));
    }

    @Override
    protected String getNameServer() {
        return sysLogMqConfig.getNameServerAddress();
    }

    @Override
    protected String getConsumerGroup() {
        return sysLogMqConfig.getConsumerGroup();
    }

    @Override
    protected String getInstanceName() {
        return sysLogMqConfig.getConsumerMongoInstance();
    }

    
    @Override
    protected String getTopic() {
        return sysLogMqConfig.getTopic();
    }

    @Override
    protected String getTag() {
        return sysLogMqConfig.getTag();
    }

}

六. 具体发送消息实例:

6.1 我这里是配合spring事件监听机制以及策略模式又做了一层解耦合, 把消息的构建和发送消息的逻辑分离开, 而且使用了异步

package com.canxiusi.web.online.producer;

import com.canxiusi.common.service.entity.logs.SysOperateLog;
import com.canxiusi.common.service.log4j2.spi.Logger;
import com.canxiusi.common.service.log4j2.spi.LoggerFactory;
import com.canxiusi.common.service.spring_event.constant.EventConstant;
import com.canxiusi.common.service.strategy.anno.StrategyAnno;
import com.canxiusi.common.service.strategy.bean.Strategy;
import com.canxiusi.common.service.strategy.context.Flow;
import com.canxiusi.common.service.strategy.context.FlowContext;
import com.canxiusi.common.service.utils.JsonUtil;
import com.canxiusi.common.service.utils.LogUtils;
import com.canxiusi.common.service.utils.map_utils.CommonMapParam;
import com.canxiusi.common.service.utils.map_utils.ExecuteDataHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Optional;


@Component
@StrategyAnno(routingKey = EventConstant.SYS_LOG_SENDER, weight = "10010")
public class SysLogEventSender implements Strategy {

    
    private static final Logger logger = LoggerFactory.getLogger(SysLogEventSender.class);

    @Autowired
    private SysLogProducer sysLogProducer;

    @Override
    public void doSpecificHandler(Flow flow, FlowContext flowContext) {
        long now = System.currentTimeMillis();
        // 获取不为null的sysLog
        SysOperateLog operateLog = Optional.ofNullable(
                        ExecuteDataHelper.getParam(flowContext.getGlobalMap(), CommonMapParam.SYS_LOG_LOG))
                .orElse(new SysOperateLog());
        byte[] bytes = JsonUtil.toJsonAsBytes(operateLog);
        // 发送mq消息
        sysLogProducer.sendMsgByteOneway(bytes);
        long end = System.currentTimeMillis();
        logger.info(LogUtils.format("当前线程<{0}>, <{1}>发送消息, flowContext=<{2}>, 耗时=<{3}>毫秒",
                Thread.currentThread().getName(),
                this.getClass().getSimpleName(),
                flowContext.getGlobalMap(),
                end - now));
    }

}

七. 运行测试 :

7.1 首先是生产者发送消息: (消息内容可以忽略, 模拟的), 这里我自定了线程工厂和线程名称, 可以看到线程已经不再是spring默认分配的 http-nio-端口号, 而是我使用线程池开启的新线程

7.3 消费者消费消息情况, 可以看到2个消费者都消费了消息, 并且是异步消费, 但是这个线程是由rocketmq内部产生的

八. 之后登录控制台查看:

这里模拟的场景是发送一条消息被2个消费者消费

自定义对rocketmq生产者以及消费者的封装

Java相关栏目本月热门文章