RabbitMQ：RabbitMQ + Spring配置文件rabbit标签

太过爱你忘了你带给我的痛 2022-05-18 05:14 171阅读 0赞

> # **RabbitMQ：RabbitMQ + Spring配置文件rabbit标签** #

### 1.消费者配置文件和启动类： ###

** 【Consumer.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:context="http://www.springframework.org/schema/context"
           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/context
                    http://www.springframework.org/schema/context/spring-context.xsd
                http://www.springframework.org/schema/rabbit
                    http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
    
        
        <rabbit:connection-factory id="connectionFactory" host="127.0.0.1" username="caoxia"
                                   password="caoxia123456" port="5672" virtual-host="/"  channel-cache-size="5" />
    
        <rabbit:admin connection-factory="connectionFactory"/>
    
        
        <rabbit:queue durable="true" auto-delete="false" exclusive="false" name="spring.queue.tag"/>
    
    
        
        <rabbit:direct-exchange name="spring.queue.exchange" durable="true" auto-delete="false">
            <rabbit:bindings>
                <rabbit:binding queue="spring.queue.tag" key="spring.queue.tag.key"/>
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        <bean id="receiveMessageListener"
              class="com.caox.rabbitmq.demo._13_spring_rabbitmq_label_xml.ReceiveMessageListener" />
    
        
        <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" >
            <rabbit:listener queues="spring.queue.tag" ref="receiveMessageListener" />
        </rabbit:listener-container>
    
        
        
                         
    
        
              
    </beans>

**【ConsumerMain.java】：**

package com.caox.rabbitmq.demo._13_spring_rabbitmq_label_xml;
    
    /**
     * Created by nazi on 2018/7/30.
     */
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class ConsumerMain {
    
        public static void main(String[] args) {
            new ClassPathXmlApplicationContext("spring-rabbit-label/Consumer.xml");
    
        }
    }

### **2.生产者配置文件和启动类： ** ###

**【Producer.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:context="http://www.springframework.org/schema/context"
           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/context
                    http://www.springframework.org/schema/context/spring-context.xsd
                http://www.springframework.org/schema/rabbit
                    http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
    
        
        <rabbit:connection-factory id="connectionFactory"
                                   host="127.0.0.1" username="caoxia" password="caoxia123456" port="5672"
                                   virtual-host="/" channel-cache-size="5" />
    
        <rabbit:admin connection-factory="connectionFactory" />
    
        
        <rabbit:queue  durable="true"
                       auto-delete="false" exclusive="false" name="spring.queue.tag" />
    
        
        
        <rabbit:direct-exchange name="spring.queue.exchange"
                                durable="true" auto-delete="false">
            <rabbit:bindings>
                <rabbit:binding queue="spring.queue.tag" key="spring.queue.tag.key" />
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        
        <bean id="jsonMessageConverter"
              class="com.caox.rabbitmq.demo._13_spring_rabbitmq_label_xml.Gson2JsonMessageConverter" />
    
        
        <rabbit:template id="amqpTemplate" exchange="spring.queue.exchange"  routing-key="spring.queue.tag.key"
                         connection-factory="connectionFactory" message-converter="jsonMessageConverter" />
    
        
        
              
            
            
            
        
    </beans>

** 【ProducerMain.java】：**

package com.caox.rabbitmq.demo._13_spring_rabbitmq_label_xml;
    
    /**
     * Created by nazi on 2018/7/30.
     */
    
    import com.caox.sharding.entity.User;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class ProducerMain {
    
        public static void main(String[] args) {
            ApplicationContext context = new ClassPathXmlApplicationContext("spring-rabbit-label/Producer.xml");
            AmqpTemplate amqpTemplate = context.getBean(RabbitTemplate.class);
            User user = new User();
            user.setName("niuniu");
            amqpTemplate.convertAndSend(user);
        }
    }

### **3.**实现一个消息监听器ReceiveMessageListener.java ###

package com.caox.rabbitmq.demo._13_spring_rabbitmq_label_xml;
    
    import org.springframework.amqp.core.Message;
    
    /**
     * Listener interface to receive asynchronous delivery of Amqp Messages.
     *
     * Created by nazi on 2018/7/30.
     */
    public interface MessageListener {
    
        void onMessage(Message message);
    
    }

### **4.messageconver插件实现(Gson)** ###

**spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列，由于Gson的速度快于jackson,这里替换为Gson的一个实现**

**【pom.xml引入jar包】：**

<dependency>
           <groupId>com.google.code.gson</groupId>
           <artifactId>gson</artifactId>
           <version>2.3</version>
    </dependency>

**【Gson2JsonMessageConverter.java 】：**

package com.caox.rabbitmq.demo._13_spring_rabbitmq_label_xml;
    
    import java.io.IOException;
    import java.io.UnsupportedEncodingException;
    
    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 com.google.gson.Gson;
    
    /**
     * Created by nazi on 2018/7/30.
     * 由于考虑到效率，如下使用Gson实现消息转换。
     */
    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);
        }
    }