监听mq消息(用于监听DTO获取请求参数，结合spring配置)

梦里梦外; 2022-03-29 03:22 285阅读 0赞

> ### 监听mq消息(用于监听DTO获取请求参数，结合spring配置) ###

**【spring-rabbitmq.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"
    	   xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    
    	
        <rabbit:connection-factory id="rabbitmqConnectionFactory" host="${rabbitmq.address}" port="${rabbitmq.port}"
                                   username="${rabbitmq.username}" password="${rabbitmq.password}"
                                   virtual-host="${rabbitmq.vhost}" connection-factory="refConnectionFactory" />
    
    	<bean id="refConnectionFactory" class="com.rabbitmq.client.ConnectionFactory">
    		
    		<property name="requestedHeartbeat" value="240" />
    		
    		<property name="connectionTimeout" value="2000" />
    	</bean>
    
    	<rabbit:admin connection-factory="rabbitmqConnectionFactory" />
    
    	
    	<rabbit:queue id="message_receive_mq" name="com.caox.model.mq.ParamsDTO"
    				  durable="true" auto-delete="false" exclusive="false" />
    	
    	<rabbit:listener-container connection-factory="rabbitmqConnectionFactory" acknowledge="manual"
    							   prefetch="0" concurrency="5" max-concurrency="5">
    
    	<rabbit:listener queues="message_receive_mq" ref="messageListener" />
    	</rabbit:listener-container>
    
    </beans>

**【监听实例，处理逻辑】： **

package com.caox.rabbitmq.listener;
    
    import com.alibaba.fastjson.JSON;
    import com.caox.model.mq.ParamsDTO;
    import com.caox.rabbitmq.MqManualUtil;
    import com.rabbitmq.client.Channel;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
    import org.springframework.stereotype.Component;
    
    /**
     * 监听获取mq消息
     *
     * @author : nazi 2019/01/07
     */
    @Slf4j
    @Component
    public class MessageListener implements ChannelAwareMessageListener {
    
        @Override
        public void onMessage(Message message, Channel channel) {
            try {
                long startTime = System.currentTimeMillis();
                String msg = new String(message.getBody(), "UTF-8");
                log.info("mq消息队列刷新接受消息 | PARAMS：{}", msg);
                
                // msg 格式必须是jsonStr格式 如：{"message":"javaBean To jsonStr"} 否则 JSON.parseObject报错
    
                ParamsDTO paramsDTO = JSON.parseObject(msg, ParamsDTO.class);
    
                log.info("mq消息队列刷新接受消息 | RESULT :{}" ,paramsDTO.getMessage());
    
                // 手动移除消息
                MqManualUtil.manualRemoveMS(message, channel);
                log.info("mq消息队列刷新接受消息，执行成功，耗时：{}", System.currentTimeMillis()-startTime);
            } catch (Throwable e){
                log.error("mq消息队列刷新接受消息，异常信息：", e);
                // 手动移除消息
                MqManualUtil.manualRemoveMS(message, channel);
            }
        }
    
    }

**【javaBean对象】： **

package com.caox.model.mq;
    
    import lombok.Getter;
    import lombok.Setter;
    import lombok.ToString;
    
    import java.io.Serializable;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/7 16:57
     */
    @Setter
    @Getter
    @ToString
    public class ParamsDTO implements Serializable {
        /** 消息 */
        private String message;
    }

**【mq手动工具类】：**

package com.caox.rabbitmq;
    
    import com.rabbitmq.client.Channel;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    
    /**
     * MQ手动工具类
     *
     * @author : nazi 2019/01/07
     */
    @Slf4j
    public class MqManualUtil {
    
        /**
         * 手动移除消息
         *
         * @param message MQ消息对象
         * @param channel MQ通道对象
         */
        public static void manualRemoveMS(Message message, Channel channel){
            try {
                long startTime = System.currentTimeMillis();
                log.info("手动移除MQ消息开始");
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                log.info("手动移除MQ消息结束，耗时：{}", System.currentTimeMillis()-startTime);
            } catch (Throwable e) {
                log.error("手动移除MQ消息异常，异常信息：", e);
            }
        }
    
    }

**【rabbitmq发送消息 javaBean转jsonStr】：**[【参考文献】：json与javabean之间的转化][json_javabean]

ParamsDTO paramsDTO = new ParamsDTO();
     paramsDTO.setMessage("javaBean To jsonStr");
     //1、使用JSONObject
     Object json = JSONObject.toJSON(paramsDTO);
     json.toString ==> {"message":"javaBean To jsonStr"}

[json_javabean]: http://www.cnblogs.com/ya-qiang/p/9009506.html