Java 集成阿里云消息队列，日志消息存储

JSON 2018-04-24 23:35:57 5212

根据地区ECS分布，选择相同地区的消息队列组件，对阿里云消息队列我做了简单的压测，效果还是不错的，费用比较低。最近一个项目中实际应用到了阿里云的消息队列。用来处理日志和业务数据。测试可以选择公网topic。生产环境一定要选择与自己ECS相同的低于，比如你ECS是华南1，华北2，那么你的队列实例也选择相同的即可。

这篇可能给你要使用，或者已经使用阿里云队列的人员会有所帮助，其他请略过。

Springboot 集成阿里云队列：https://www.sojson.com/blog/293.html

阿里云集群/广播消费：https://help.aliyun.com/document_detail/43163.html

Spring 集成消息队列：https://help.aliyun.com/document_detail/29553.html

Java 环境准备：https://help.aliyun.com/document_detail/29546.html

事物消息队列：https://help.aliyun.com/document_detail/29548.html

日志队列消费者

消费我采用多节点 Springboot 微服务消费。直接上代码了。

@Configuration
@Slf4j
public class QueueConsumer {


    @Value("${aliyun.accessKey}")
    private String accessKey;
    @Value("${aliyun.accessSecretKey}")
    private String secretKey;



    @Value("${aliyun.queue.stock.consumerId}")
    private String stockConsumerId;
    @Value("${aliyun.queue.stock.topic}")
    private String stockTopic;
    @Value("${aliyun.queue.log.consumerId}")
    private String logConsumerId;
    @Value("${aliyun.queue.log.topic}")
    private String logTopic;
    @Autowired
    AppleCheckUtils appleCheckUtils;
    @Autowired
    StockDetailsService stockDetailsService;
    @Autowired
    StockService stockService;
    @Autowired
    LogRepository logRepository;
    @Autowired
    LogUtils logUtils;

    @Bean(name="stockConsumerBean")
    public ConsumerBean initStockConsumerBean(){
        ConsumerBean consumerBean = new ConsumerBean();
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ConsumerId, stockConsumerId);
        properties.put(PropertyKeyConst.AccessKey, accessKey);
        properties.put(PropertyKeyConst.SecretKey, secretKey);
        properties.put(PropertyKeyConst.ConsumeThreadNums, 10);

        consumerBean.setProperties(properties);

        //监听消息
        MessageListener messageListener = (message, context) -> {

            String json = new String(message.getBody());
            SOMap<String,String> map = new Gson().fromJson(json, SOMap.class);
            String transactionId = map.get("transactionId");
            String checksum = map.get("checksum");


            log.info("队列消费开始:[]。transactionId:{} , checksum:{}",message.getMsgID(),transactionId,checksum);
            logUtils.init(1L)
                    .begin().log("队列消费开始:[]。transactionId:{} , checksum:{}",message.getMsgID(),transactionId,checksum).businessId(checksum);
            try {
                StockDetails stockDetails = stockDetailsService.findById(transactionId);
                if(null == stockDetails){
                    log.error("队列消费出现查询不到stockDetails对象:[{}]。transactionId:{} , checksum:{}",message.getMsgID(),transactionId,checksum);
                    logUtils.error("队列消费出现查询不到stockDetails对象:[{}]。transactionId:{} , checksum:{}",message.getMsgID(),transactionId,checksum)
                            .end().send();
                    return Action.ReconsumeLater;
                }
                if(!new Integer(0).equals(stockDetails.getStatus())){
                    log.error("队列消费出现stockDetails.status != 0:[{}]，:[{}]。transactionId:{} , checksum:{}",stockDetails.getStatus(),message.getMsgID(),transactionId,checksum);
                    logUtils.error("队列消费出现stockDetails.status != 0:[{}]，:[{}]。transactionId:{} , checksum:{}",stockDetails.getStatus(),message.getMsgID(),transactionId,checksum)
                            .end().send();
                    return Action.CommitMessage;
                }
                Boolean result = appleCheckUtils.check(stockDetails.getReceiptDataOne(), transactionId, checksum);
                if(result){
                    stockDetailsService.updateStatus1ByQueue(stockDetails,checksum);
                    return Action.CommitMessage;
                }
                logUtils.log("{}：队列成功 :transactionId:{}",checksum,transactionId)
                        .end().send();
                return Action.ReconsumeLater;
            }catch (Exception e) {
                log.error("{}：队列执行失败 :transactionId:{}",checksum,transactionId,e);
                logUtils.error("{}：队列执行失败 :transactionId:{}",checksum,transactionId)
                        .end().send();
                return Action.ReconsumeLater;
            }
        };
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>();

        Subscription subscription = new Subscription();
        subscription.setTopic(stockTopic);
        subscription.setExpression("*");
        subscriptionTable.put(subscription,messageListener);
        consumerBean.setSubscriptionTable(subscriptionTable);
        consumerBean.start();




        return consumerBean;
    }

    @Bean(name="logConsumerBean")
    public ConsumerBean initLogConsumerBean(){
        ConsumerBean consumerBean = new ConsumerBean();
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ConsumerId, logConsumerId);
        properties.put(PropertyKeyConst.AccessKey, accessKey);
        properties.put(PropertyKeyConst.SecretKey, secretKey);
        properties.put(PropertyKeyConst.ConsumeThreadNums, 10);

        consumerBean.setProperties(properties);
        //监听消息
        MessageListener messageListener = (message, context) -> {


            String body = new String(message.getBody());
            Log logx = new Gson().fromJson(body, Log.class);

            try {
                logRepository.save(logx);
                return Action.CommitMessage;
            }catch (Exception e) {
                log.error("日志队列出错",e);
                return Action.ReconsumeLater;
            }
        };
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>();

        Subscription subscription = new Subscription();
        subscription.setTopic(logTopic);
        subscription.setExpression("*");
        subscriptionTable.put(subscription,messageListener);
        consumerBean.setSubscriptionTable(subscriptionTable);
        consumerBean.start();

        return consumerBean;
    }
}

消息生产者

消息生产者是用普通的SSM项目， XML 配置的，也是直接贴代码了。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean id="producer" class="com.aliyun.openservices.ons.api.bean.ProducerBean" init-method="start" destroy-method="shutdown">
        <!-- Spring 接入方式支持 Java SDK 支持的所有配置项 -->
        <property name="properties" > <!--生产者配置信息-->
            <props>
                <prop key="ProducerId">${queue_log_producerId}</prop> <!--请替换 XXX-->
                <prop key="AccessKey">${aliyun_accessKey}</prop>
                <prop key="SecretKey">${aliyun_accessSecretKey}</prop>
            </props>
        </property>
    </bean>

    <!--<bean class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">-->
        <!--<property name="staticMethod" value="com.sojson.common.utils.LogUtils.setProducer"/>-->
        <!--<property name="arguments" ref="producer"/>-->
    <!--</bean>-->

    <bean id="logUtils" class="com.sojson.common.utils.LogUtils" init-method="init">

        <property name="producer" ref="producer" />
        <property name="logTypeService" ref="logTypeService" />
        <property name="logTopic" value="${queue_log_topic}" />
        <property name="logPriducerId" value="${queue_log_producerId}" />
    </bean>

</beans>

另外贴一下我的日志工具类。就是会同步到队列里去，然后消费日志入库的操作。

@Slf4j
@Data
@Component
public final class LogUtils {

    private Log logEntity;

    private LogUtils(){}


    public String logTopic;

    public String logPriducerId;
    private int id;

    private ProducerBean producer;

    private LogTypeService logTypeService;


    /**
     * 初始化log工具类，
     * @param type  日志的类型，如果没有必须要先添加，填写ID
     * @return
     */
    public  LogUtils init(int type){
        //保存上一次的，里面会判断是否为null
        send();
        i(type);
        return this;
    }

    void i(int id){
        Log elog = new Log();
        this.id = id;
        elog.setLogType(id);
        elog.setLogTypeName(LogType.get(id));
        elog.setId(MathUtil.uuid());
        this.logEntity = elog;
    }
    public LogUtils clear(){
        return init(this.id);
    }

    /**
     * 操作之前描述
     * @param obj
     * @return
     */
    public LogUtils before(Object obj){
        if(obj !=null){
            String before = toJSON(obj);
            this.logEntity.setBeforex(before);
        }
        return this;
    }
    /**
     * 改变之后描述
     * @param obj
     * @return
     */
    public LogUtils after(Object obj){
        if(obj !=null){
            String after = toJSON(obj);
            this.logEntity.setAfterx(after);
        }
        return this;
    }
    //业务ID
    public LogUtils businessId(Object id){
        if(id !=null){
            String businessId = id.toString();
            this.logEntity.setBusinessId(businessId);
        }
        return this;
    }
    //操作用户
    public LogUtils memberName(String name){
        if(name !=null){
            this.logEntity.setMemberName(name);
        }
        return this;
    }
    /**
     * 用户信息id，可以是操作id，可以是数据相关的用户ID
     * @param id
     * @return
     */
    public LogUtils memberId(Long id){
        if(id !=null){
            this.logEntity.setMemberId(id);
        }
        return this;
    }
    public LogUtils member(){
        Member token = TokenManager.getToken();
        if(token !=null){
            this.logEntity.setMemberName(token.getUserName());
            this.logEntity.setMemberId(token.getId());
        }
        return this;
    }
    //直接给一个用户对象
    public LogUtils member(Member token){

        if(token !=null){
            this.logEntity.setMemberName(token.getUserName());
            this.logEntity.setMemberId(token.getId());
        }
        return this;
    }

    /**
     * 描述信息
     * @param str   第一个是字符串，后面可以多个参数，占位用“{}”
     * @return
     */
    public LogUtils log(Object...str){
        return out(str);
    }
    public LogUtils error(Object...str){
        if(null != this.logEntity){
            this.logEntity.setType(2);
        }
        return out(str);
    }
    LogUtils out(Object...str){
        if(str !=null && str.length !=0){
            //默认log类型
            if(null != this.logEntity && null == this.logEntity.getType()){
                this.logEntity.setType(1);
            }

            String logstr = (String)str[0];
            if(str.length > 1){
                for(int i=1;i<str.length;i++){
                    String element = toJSON(str[i]);
                    logstr = logstr.replaceFirst("\\{\\s*?}",element);
                }
            }
            String old = StringUtils.trim(this.logEntity.getLogIntro());
            if(null == old){
                this.logEntity.setLogIntro(logstr);
            }else{
                this.logEntity.setLogIntro(old +"\n"+logstr);
            }
        }
        return this;
    }
    /**
     * 参数信息
     * @param str   第一个是字符串，后面可以多个参数，占位用“{}”
     * @return
     */
    public LogUtils parms(Object...str){
        if(str !=null && str.length !=0){
            String logstr = (String) str[0];
            if(str.length > 1){
                for(int i=1;i<str.length;i++){
                    String element = toJSON(str[i]);
                    logstr = logstr.replaceFirst("\\{\\s*?}",element);
                }
            }
            String old = this.logEntity.getParms();
            if(null == old){
                this.logEntity.setParms(logstr);
            }else{
                this.logEntity.setParms(old +" \n "+logstr);
            }
        }
        return this;
    }

    /**
     * 开始时间
     * @return
     */
    public LogUtils begin(){
        return begin(new Date());
    }
    public LogUtils begin(Date date){
        this.logEntity.setBeginTime(date);
        return this;
    }

    /**
     * 结束时间
     * @return
     */
    public LogUtils end(){
        return end(new Date());
    }
    public LogUtils end(Date date){
        this.logEntity.setEndTime(date);

        //时间计算
        if(null != this.logEntity.getBeginTime()){
            this.logEntity.setTime(date.getTime() - this.logEntity.getBeginTime().getTime());
        }
        return this;
    }


    public void send(){
        if(null != logEntity && null != logEntity.getLogIntro() && null != logEntity.getId()){
            try{
                Message m = new Message();
                //创建时间
                this.logEntity.setCreateTime(new Date());

                m.setBody(new Gson().toJson(this.logEntity).getBytes());
                m.setTopic(logTopic);
                m.setTag("log");
                producer.send(m);
            }catch (Exception e){
                log.error("添加队列失败，：{}",e);
            }
        }
        i(id);
    }

    String toJSON(Object obj){
        try {
            String str;
            if(obj instanceof String || obj instanceof Integer|| obj instanceof Boolean || obj instanceof Short || obj instanceof Float || obj instanceof  Double|| obj instanceof Long || obj instanceof Byte){
                str =  String.valueOf(obj);
            }else{
                str = new Gson().toJson(obj);
            }
            return str;
        } catch (Exception e) {
            return obj.toString();
        }
    }



    public LogUtils debug(){
        if(null != this.logEntity){
            this.logEntity.setType(1);
        }
        return this;
    }
    public LogUtils warn(){
        if(null != this.logEntity){
            this.logEntity.setType(3);
        }
        return this;
    }


    public void setProducer(ProducerBean producer){
        this.producer = producer;
    }
    public void setLogPriducerId(String logPriducerId){
        this.logPriducerId = logPriducerId;
    }
    public void setLogTopic(String logTopic){
        this.logTopic = logTopic;
    }
    public void setLogTopic(LogTypeService logTypeService){
        this.logTypeService = logTypeService;
    }

    private void init() {
        LogType.init(logTypeService.findAll());
    }
}