rabbitMQ教程 spring整合rabbitMQ代码实例

叁歲伎倆 2022-05-16 10:09 234阅读 0赞

转载自 [https://www.cnblogs.com/tohxyblog/p/7256554.html][https_www.cnblogs.com_tohxyblog_p_7256554.html]

## 一、开启rabbitMQ服务，导入MQ jar包和gson jar包(MQ默认的是jackson，但是效率不如Gson，所以我们用gson) ##

## ![1111496-20170729165504894-1809112620.png][]![1111496-20170729165602222-735748496.png][] ##

##  二、发送端配置，在spring配置文件中配置 ##

![复制代码][copycode.gif]

<?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-3.0.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
    
       
        <rabbit:connection-factory id="connectionFactory"  
            host="localhost" username="guest" password="guest" 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="sendMQ.Gson2JsonMessageConverter" />  
      
          
        <rabbit:template id="amqpTemplate" exchange="spring.queue.exchange"  routing-key="spring.queue.tag.key"  
            connection-factory="connectionFactory" message-converter="jsonMessageConverter" />

![复制代码][copycode.gif]

发送端代码：GSON配置

![复制代码][copycode.gif]

package sendMQ;
    
    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;
    
    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);  
            }  
    }

![复制代码][copycode.gif]

发送类接口：

![复制代码][copycode.gif]

public interface MQProducer {
        /**
         * 发送消息到指定队列
         * @param queueKey
         * @param object
         */
        public void sendDataToQueue(String queueKey, Object object);
    }

![复制代码][copycode.gif]

实现类：test是测试用的。

![复制代码][copycode.gif]

package sendMQ;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    import org.springframework.stereotype.Component;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    @RunWith(value = SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations = {
        "classpath:/spring-common.xml"})
    
    
    @Component
    public class MQProducerImpl implements MQProducer {
    
        @Autowired
        private  AmqpTemplate amqpTemplate;
    
            @Override
            public void sendDataToQueue(String queueKey, Object object) {
                System.out.println("--"+amqpTemplate);
                try {
                    amqpTemplate.convertAndSend(object);
                    System.out.println("------------消息发送成功");
                } catch (Exception e) {
                    System.out.println(e);
                }
    
            }
            
            @Test
          public  void test() {  
                Map<String,Object> msg = new HashMap<>();
                msg.put("data","hello,456");
                while(true){
                amqpTemplate.convertAndSend(msg); 
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    // TODO 自动生成的 catch 块
                    e.printStackTrace();
                }
                }
               
            }  
    
    }

![复制代码][copycode.gif]

接收端配置：

![复制代码][copycode.gif]

<rabbit:connection-factory id="connectionFactory" host="localhost" username="guest"  
            password="guest" 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="receiveMQ.QueueListenter" />  
           
          
        <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" >  
            <rabbit:listener queues="spring.queue.tag" ref="receiveMessageListener" />  
        </rabbit:listener-container>

![复制代码][copycode.gif]

接收端代码：

![复制代码][copycode.gif]

package receiveMQ;
    
    import java.io.UnsupportedEncodingException;
    
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    public class QueueListenter implements MessageListener{
        @Override
        public void onMessage(Message msg) {
             try {
                System.out.print("-------------------"+new String(msg.getBody(),"UTF-8"));
            } catch (UnsupportedEncodingException e) {
                // TODO 自动生成的 catch 块
                e.printStackTrace();
            }
        }
    
    }

![复制代码][copycode.gif]

接收端测试启动：

![复制代码][copycode.gif]

package receiveMQ;
    
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    public class ConsumerMain {
        public static void main(String[] args) {  
            new ClassPathXmlApplicationContext("spring-common.xml");    
        }  
    }

![复制代码][copycode.gif]

上面代码均有注释，应该不难看懂，复制即可使用，实现了MQ的简单功能。

说明：可以配置多个接收端，spring默认的是负载均衡机制，每个接收端接收一条的来，这些扩展功能待后面有时间再讲解

[https_www.cnblogs.com_tohxyblog_p_7256554.html]: https://www.cnblogs.com/tohxyblog/p/7256554.html
[1111496-20170729165504894-1809112620.png]: https://images2017.cnblogs.com/blog/1111496/201707/1111496-20170729165504894-1809112620.png
[1111496-20170729165602222-735748496.png]: /images/20220516/7544a5d26bfe47d19e0877864cc8d3ef.png
[copycode.gif]: /images/20220516/f3347e44bc7a4c59bd431b6bb8a20b7e.png