RabbitMQ spring 使用总结

柔情只为你懂 2022-09-25 15:18 211阅读 0赞

rabbitMQ相关概念不在本文介绍范围，rabbitMQ官网和其他博客都有大量介绍。

本文重点内容是spring和rabbit环境搭建以及使用中注意事项总结。

### 1.1   rabbitMQ服务器搭建 ###

下载安装官网最新版本服务器

### 1.2   rabbitMQ开启服务管理 ###

rabbitMQ start 启动

### 1.3   spring pom配置 ###

<spring-rabbit.version>1.3.9.RELEASE</spring-rabbit.version>  <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>${rabbitmq-client.version}</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>${spring-rabbit.version}</version> </dependency>

### 1.4   spring config配置 ###

在D:\\workspace\\sps\\src\\main\\resources\\spring-rabbitmq.xml

配置如下：

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xmlns="http://www.springframework.org/schema/beans" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd"> <mvc:annotation-driven /> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.master.ip}" port="${rabbitmq.master.port}" username="${rabbitmq.master.username}" password="${rabbitmq.master.password}" /> <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="order_topic_exchange" message-converter="gsonConverter" /> <rabbit:admin connection-factory="connectionFactory" /> <rabbit:queue name="orderQueue" durable="true" /> <rabbit:queue name="orderPayQueryQueue" durable="true" auto-delete="false" exclusive="false"> <rabbit:queue-arguments> <entry key="x-message-ttl"> <value type="java.lang.Long">600000</value> </entry> <entry key="x-dead-letter-exchange" value="pay_delay_exchange"/> </rabbit:queue-arguments> </rabbit:queue> <rabbit:queue name="orderPayDelayQueryQueue" durable="true"/> <rabbit:topic-exchange name="pay_delay_exchange"> <rabbit:bindings> <rabbit:binding queue="orderPayDelayQueryQueue" pattern="orderPay.#"/> </rabbit:bindings> </rabbit:topic-exchange> <rabbit:topic-exchange name="order_topic_exchange"> <rabbit:bindings> <rabbit:binding queue="orderQueue" pattern="sps.#"/> <rabbit:binding queue="orderPayQueryQueue" pattern="orderPay.#"/> </rabbit:bindings> </rabbit:topic-exchange> <rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" concurrency="10"> <rabbit:listener queues="orderQueue" ref="orderQueueListener"/> </rabbit:listener-container> <bean id="orderQueueListener" class="com.supuy.sps.services.queue.OrderQueueListener" /> <bean id="gsonConverter" class="com.supuy.core.mq.Gson2JsonMessageConverter"/> </beans>

### 1.5  延迟消息队列 ###

有时候，因为各种原因，我们想实现延迟消费的目的，但是rabbitMQ并没有提供这个功能，这时候，可以通过x-message-ttl和x-dead-letter-exchange实现。

<rabbit:queue name="orderPayQueryQueue" durable="true" auto-delete="false" exclusive="false">
            <rabbit:queue-arguments>
                <entry key="x-message-ttl">
                    <value type="java.lang.Long">600000</value>
                </entry>
                <entry key="x-dead-letter-exchange" value="pay_delay_exchange"/>
            </rabbit:queue-arguments>
        </rabbit:queue>

### 1.6   生产者 ###

@Override
    public void orderBuilder(int type,String orderCode) {
        String key = "tps."+orderCode;
        orderCode = type+"."+orderCode;
        amqpMaster.convertAndSend(key, orderCode);
        logger.info("订单加入消息队列，订单编码：{}", key);
    }

### 1.7   消费者 ###

package com.supuy.tps.service.queue;
    
    import com.alibaba.fastjson.JSON;
    import com.rabbitmq.client.Channel;
    import com.supuy.tps.common.mq.Gson2JsonMessageConverter;
    import com.supuy.tps.dto.bean.WmsOrderParam;
    import com.supuy.tps.service.IOrderShopService;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
    import org.springframework.beans.factory.annotation.Autowired;
    
    /** * Created by bill on 2016/5/31. */
    public class OrderSendQueueListener implements ChannelAwareMessageListener {
        private static Logger logger = LoggerFactory.getLogger(OrderSendQueueListener.class);
        @Autowired
        private Gson2JsonMessageConverter messageConverter;
        @Autowired
        private IOrderShopService orderShopService;
        @Override
        public void onMessage(Message message, Channel channel) throws Exception {
            channel.basicQos(100);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            String data=(String)messageConverter.fromMessage(message);
            if (data!=null){
                WmsOrderParam wmsOrderParam= JSON.parseObject(data,WmsOrderParam.class);
                if (wmsOrderParam != null){
                    wmsOrderParam.setOrderCode(wmsOrderParam.getOrderCode().substring(1));
                    orderShopService.pushOrderLogInfo(wmsOrderParam);
                }
            }
        }
    }

附加类Gson2JsonMessageConverter实现如下，

package com.supuy.tps.common.mq;
    
    import com.google.gson.Gson;
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.support.converter.AbstractJsonMessageConverter;
    import org.springframework.amqp.support.converter.ClassMapper;
    import org.springframework.amqp.support.converter.DefaultClassMapper;
    import org.springframework.amqp.support.converter.MessageConversionException;
    
    import java.io.IOException;
    import java.io.UnsupportedEncodingException;
    
    public class Gson2JsonMessageConverter extends AbstractJsonMessageConverter {
          
        private static Log log = LogFactory.getLog(Gson2JsonMessageConverter.class);
          
        private static ClassMapper classMapper =  new DefaultClassMapper();
      
        private static Gson gson = new Gson();
      
        public Gson2JsonMessageConverter() {  
            super();  
        }  
    
    
        @Override  
        protected Message createMessage(Object object, MessageProperties messageProperties) {
            byte[] bytes = null;  
            try {  
                String jsonString = gson.toJson(object);  
                bytes = jsonString.getBytes(getDefaultCharset());  
            }  
            catch (IOException e) {  
                throw new MessageConversionException(
                        "Failed to convert Message content", e);  
            }  
            messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
            messageProperties.setContentEncoding(getDefaultCharset());  
            if (bytes != null) {  
                messageProperties.setContentLength(bytes.length);  
            }  
            classMapper.fromClass(object.getClass(), messageProperties);
            return new Message(bytes, messageProperties);
        }  
      
        @Override  
        public Object fromMessage(Message message) throws MessageConversionException {
            Object content = null;  
            MessageProperties properties = message.getMessageProperties();
            if (properties != null) {  
                String contentType = properties.getContentType();  
                if (contentType != null && contentType.contains("json")) {  
                    String encoding = properties.getContentEncoding();  
                    if (encoding == null) {  
                        encoding = getDefaultCharset();  
                    }  
                    try {  
                            Class<?> targetClass = getClassMapper().toClass(
                                    message.getMessageProperties());
                            content = convertBytesToObject(message.getBody(),  
                                    encoding, targetClass);  
                    }  
                    catch (IOException e) {  
                        throw new MessageConversionException(
                                "Failed to convert Message content", e);  
                    }  
                }  
                else {  
                    log.warn("Could not convert incoming message with content-type ["  
                            + contentType + "]");  
                }  
            }  
            if (content == null) {  
                content = message.getBody();  
            }  
            return content;  
        }  
      
        private Object convertBytesToObject(byte[] body, String encoding, Class<?> clazz) throws UnsupportedEncodingException {  
            String contentAsString = new String(body, encoding);  
            return gson.fromJson(contentAsString, clazz);  
        }
    
        @Override
        public ClassMapper getClassMapper() {
            return new DefaultClassMapper();
    
        }
    }

### 1.8   Q&A ###

1 ttl设置之后，下次修改时间，会报错，这时候，需要先删除该队列，重启项目。

2 接受消息之后，出现错误，该消息就会被持续占有，无法消费。所以，要活用消息的ack，nack，reject。