RabbitMQ消息中间件(第四章)第三部分

本章导航

• RabbitMQ整合Spring AMQP实战
• RabbitMQ整合Spring Boot实战
• RabbitMQ整合Spring Cloud实战

RabbitMQ整合Spring AMQP实战

• RabbitAdmin
• SpringAMQP声明
• SimpleMessageListenerContainer简单消息监听容器
• MessageListenerAdapter消息监听适配器
• MessageConverter 消息转换器，序列化和反序列化等操作
• 注意：autoStartup必须要设置为true，否则Spring容器不会加载RabbitAdmin类
• RabbitAdmin底层实现就是从Spring容器中获取Exchange、Bingding、RoutingKey以及Queue的@Bean声明
• 然后使用RabbitTemplate的execute方法执行对应的声明、修改、删除等一系列RabbitMQ基础功能操作
• 例如：添加一个交换机、删除一个绑定、清空一个队列里的消息等等

MessageConverter

MessageConverter消息转换器

• 我们在进行发送消息的时候，正常情况下消息体为二进制的数据方式进行传输，如果希望内部帮我们进行转换，或指定自定义的转换器，就需要用到MessageConverter
• 自定义常用转换器：MessageConverter，一般来讲都需要实现这个接口
• 重写下面两个方法：
• toMessage：java对象转换为Message
• fromMessage：Message对象转换为java对象
• Json转换器：Jackson2JsonMessageConverter：可以进行Java对象的转换功能
• DefaultJackson2JavaTypeMapper映射器：可以进行Java对象的映射关系
• 自定义二进制转换器：比如图片类型、PDF、PPT、流媒体

package com.cx.temp.common.rabbitmq.spring.convert;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.UUID;

public class ImageMessageConverter implements MessageConverter {

    @Override
    public Message toMessage(Object o, MessageProperties messageProperties) throws MessageConversionException {
        throw new MessageConversionException(" convert error ! ");
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        System.err.println("----------------PDF MessageConverter-------------------");

        Object _extName = message.getMessageProperties().getHeaders().get("extName");
        String extName = _extName == null ? "png" : _extName.toString();

        byte[] body = message.getBody();
        String fileName = UUID.randomUUID().toString();
        String path = "d:/010_test/" + fileName + "." + extName;
        File f = new File(path);
        try {
            Files.copy(new ByteArrayInputStream(body), f.toPath());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return f;
    }
}

package com.cx.temp.common.rabbitmq.spring.convert;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.UUID;

public class PDFMessageConverter implements MessageConverter {

    @Override
    public Message toMessage(Object o, MessageProperties messageProperties) throws MessageConversionException {
        throw new MessageConversionException(" convert error ! ");
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        System.err.println("----------------PDF MessageConverter-------------------");

        byte[] body = message.getBody();
        String fileName = UUID.randomUUID().toString();
        String path = "d:/010_test/" + fileName + ".pdf";
        File f = new File(path);
        try {
            Files.copy(new ByteArrayInputStream(body), f.toPath());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return f;
    }
}

package com.cx.temp.common.rabbitmq.spring.convert;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;

public class TextMessageConverter implements MessageConverter {

    @Override
    public Message toMessage(Object o, MessageProperties messageProperties) throws MessageConversionException {
        return new Message(o.toString().getBytes(), messageProperties);
    }

    @Override
    public Object fromMessage(Message message) throws MessageConversionException {
        String contentType = message.getMessageProperties().getContentType();
        if(null != contentType && contentType.contains("text")) {
            return new String(message.getBody());
        }
        return message.getBody();
    }

}

package com.cx.temp.common.rabbitmq.spring;

import com.cx.temp.common.rabbitmq.spring.adapter.MessageDelegate;
import com.cx.temp.common.rabbitmq.spring.convert.ImageMessageConverter;
import com.cx.temp.common.rabbitmq.spring.convert.PDFMessageConverter;
import com.cx.temp.common.rabbitmq.spring.convert.TextMessageConverter;
import com.cx.temp.common.rabbitmq.spring.entity.Order;
import com.cx.temp.common.rabbitmq.spring.entity.Packaged;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.DefaultJackson2JavaTypeMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 *
 */
@Configuration
@ComponentScan({"com.cx.temp.*"})
public class RabbitMQConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses("127.0.0.1:5672");
        connectionFactory.setUsername("root");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("/test001");
        return connectionFactory;
    }

    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    /**
     * 针对消费者配置
     * 1.设置交换机类型
     * 2.将队列绑定到交换机
     * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
     * HeadersExchange: 通过添加属性key-value匹配
     * DirectExchange: 按照routingkey分发到指定队列
     * TopicExchange: 多关键字匹配
     *
     * @return
     */
    @Bean
    public TopicExchange exchange001() {
        return new TopicExchange("topic001", true, false);
    }

    @Bean
    public Queue queue001() {
        return new Queue("queue001", true); //队列持久
    }

    @Bean
    public Binding binding001() {
        return BindingBuilder.bind(queue001()).to(exchange001()).with("spring.*");
    }

    @Bean
    public TopicExchange exchange002() {
        return new TopicExchange("topic002", true, false);
    }

    @Bean
    public Queue queue002() {
        return new Queue("queue002", true); //队列持久
    }

    @Bean
    public Queue queue003() {
        return new Queue("queue003", true); //队列持久
    }

    @Bean
    public Binding binding002() {
        return BindingBuilder.bind(queue002()).to(exchange002()).with("rabbit.*");
    }

    @Bean
    public Binding binding003() {
        return BindingBuilder.bind(queue003()).to(exchange001()).with("mq.*");
    }

    @Bean
    public Queue queue_image() {
        return new Queue("image_queue", true); //队列持久
    }

    @Bean
    public Queue queue_pdf() {
        return new Queue("pdf_queue", true); //队列持久
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

    @Bean
    public SimpleMessageListenerContainer messageContainer(ConnectionFactory connectionFactory) {

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(queue001(), queue002(), queue003(), queue_image(), queue_pdf());
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(5);
        container.setDefaultRequeueRejected(false); //不进行重回队列
        container.setAcknowledgeMode(AcknowledgeMode.AUTO); //自动签收机制
        container.setConsumerTagStrategy(new ConsumerTagStrategy() {
            @Override
            public String createConsumerTag(String queue) {
                return queue + "_" + UUID.randomUUID().toString();
            }
        });
//        container.setMessageListener(new ChannelAwareMessageListener() {
//            @Override
//            public void onMessage(Message message, Channel channel) throws Exception {
//                String msg = new String(message.getBody());
//                System.err.println("-----------消费者：" + msg);
//            }
//        });


        //第一种
//        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//        container.setMessageListener(adapter);

        //第二种
//        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//        adapter.setDefaultListenerMethod("consumeMessage"); //也可以通过这个方法设置默认查找适配器对象的方法，默认的方法名是handleMessage

        //第三种
        /** 1.适配器方式，默认是有自己的方法名称的：handleMessage
         //可以自己指定一个方法的名字：consumeMessage
         //也可以添加一个转换器：从字节数组转换String
         MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
         adapter.setDefaultListenerMethod("consumeMessage");
         adapter.setMessageConverter(new TextMessageConverter());
         container.setMessageListener(adapter);
         */

        /**
         * 2.适配器方式：我们的队列名称和方法名称也可以也进行一一匹配
         */
//        Map<String, String> queueOrTagToMethodName = new HashMap<>();
//        queueOrTagToMethodName.put("queue001", "method1");
//        queueOrTagToMethodName.put("queue002", "method2");
//
//        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//        adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
//        adapter.setMessageConverter(new TextMessageConverter());
//        container.setMessageListener(adapter);

          // 1.1 支持JSON格式的转换器
//        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//        adapter.setDefaultListenerMethod("consumeMessage");
//        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
//        adapter.setMessageConverter(jackson2JsonMessageConverter);
//        container.setMessageListener(adapter);

          // 1.2 DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持Java对象转换
//          MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//          adapter.setDefaultListenerMethod("consumeMessage");
//          Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
//          DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
//          jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);
//          adapter.setMessageConverter(jackson2JsonMessageConverter);
//          container.setMessageListener(adapter);

          // 1.3 DefaultJackson2JavaTypeMapper & Jackson2JsonMessageConverter 支持Java对象多映射转换
//          MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
//          adapter.setDefaultListenerMethod("consumeMessage");
//          Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
//          DefaultJackson2JavaTypeMapper javaTypeMapper = new DefaultJackson2JavaTypeMapper();
//
//          Map<String, Class<?>> idClassMapping = new HashMap<>();
//          idClassMapping.put("order", Order.class);
//          idClassMapping.put("packaged", Packaged.class);
//          javaTypeMapper.setIdClassMapping(idClassMapping);
//          jackson2JsonMessageConverter.setJavaTypeMapper(javaTypeMapper);
//          adapter.setMessageConverter(jackson2JsonMessageConverter);
//          container.setMessageListener(adapter);

          //1.4 ext convert
        MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
        adapter.setDefaultListenerMethod("consumeMessage");

        //全局转换器：
        ContentTypeDelegatingMessageConverter convert = new ContentTypeDelegatingMessageConverter();
        TextMessageConverter textMessageConverter = new TextMessageConverter();
        convert.addDelegate("text",  textMessageConverter);
        convert.addDelegate("html/text",  textMessageConverter);
        convert.addDelegate("xml/text",  textMessageConverter);
        convert.addDelegate("text/plan",  textMessageConverter);

        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
        convert.addDelegate("json",  jackson2JsonMessageConverter);
        convert.addDelegate("application/json",  jackson2JsonMessageConverter);

        ImageMessageConverter imageMessageConverter = new ImageMessageConverter();
        convert.addDelegate("image/png",  imageMessageConverter);
        convert.addDelegate("image",  imageMessageConverter);

        PDFMessageConverter pdfMessageConverter = new PDFMessageConverter();
        convert.addDelegate("application/pdf",  pdfMessageConverter);

        adapter.setMessageConverter(convert);
        container.setMessageListener(adapter);

         return container;
    }


}

  

package com.cx.temp.rabbitmq;

import com.cx.temp.admin.AdminApplication;
import com.cx.temp.common.rabbitmq.spring.entity.Order;
import com.cx.temp.common.rabbitmq.spring.entity.Packaged;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.io.File;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.HashMap;

@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = AdminApplication.class)
public class RabbitMQTest {

    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Test
    public void testAdmin() throws Exception {

        //第一种声明与绑定方式
        rabbitAdmin.declareExchange(new DirectExchange("test.direct", false, false));

        rabbitAdmin.declareExchange(new TopicExchange("test.topic", false, false));

        rabbitAdmin.declareExchange(new FanoutExchange("test.fanout", false, false));

        rabbitAdmin.declareQueue(new Queue("test.direct.queue", false));

        rabbitAdmin.declareQueue(new Queue("test.topic.queue", false));

        rabbitAdmin.declareQueue(new Queue("test.fanout.queue", false));

        rabbitAdmin.declareBinding(new Binding("test.direct.queue", Binding.DestinationType.QUEUE,
                "test.direct", "direct", new HashMap<>()));

        //第二种 支持链式声明与绑定
        rabbitAdmin.declareBinding(BindingBuilder
                .bind(new Queue("test.topic.queue", false))
                .to(new TopicExchange("test.topic", false, false))
                .with("user.#"));

        rabbitAdmin.declareBinding(BindingBuilder
                .bind(new Queue("test.fanout.queue", false))
                .to(new FanoutExchange("test.fanout", false, false)));


        //清空队列数据
        rabbitAdmin.purgeQueue("test.topic.queue", false);

    }

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage() throws Exception {
        //1 创建消息
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.getHeaders().put("desc", "信息描述..");
        messageProperties.getHeaders().put("type", "自定义消息类型..");
        Message message = new Message("Hello RabbitMQ".getBytes(), messageProperties);

        rabbitTemplate.convertAndSend("topic001", "spring.amqp", message, new MessagePostProcessor() {
            //消息发送之后在对这个message进行设置
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                System.err.println("----------添加额外的设置-----------");
                message.getMessageProperties().getHeaders().put("desc", "额外修改的信息描述");
                message.getMessageProperties().getHeaders().put("attr", "额外新加的属性");
                return message;
            }
        });

    }

    @Test
    public void testSendMessage2() throws Exception {
        //1 创建消息
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text/plain");
        Message message = new Message("mq 消息1234".getBytes(), messageProperties);

        rabbitTemplate.send("topic001", "spring.abc",message);

        rabbitTemplate.convertAndSend("topic001", "spring.amqp", "hello object message send!");
        rabbitTemplate.convertAndSend("topic002", "rabbit.amqp", "hello object message send!");

    }

    @Test
    public void testSendMessage4Text() throws Exception {
        //1 创建消息
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text/plain");
        Message message = new Message("mq 消息1234".getBytes(), messageProperties);

        rabbitTemplate.send("topic001", "spring.abc",message);
        rabbitTemplate.send("topic002", "rabbit.abc",message);
    }

    @Test
    public void testSendJsonMessage() throws Exception {
        Order order = new Order();
        order.setId("001");
        order.setName("消息订单");
        order.setContent("描述信息");

        ObjectMapper mapper = new ObjectMapper();
        String json = mapper.writeValueAsString(order);

        System.err.println("order 4 json" + json);

        MessageProperties messageProperties = new MessageProperties();
        //这里注意一定要修改contentType为application/json
        messageProperties.setContentType("application/json");

        Message message = new Message(json.getBytes(), messageProperties);

        rabbitTemplate.send("topic001", "spring.order",message);
    }

    @Test
    public void testSendJavaMessage() throws Exception {
        Order order = new Order();
        order.setId("001");
        order.setName("消息订单");
        order.setContent("描述信息");

        ObjectMapper mapper = new ObjectMapper();
        String json = mapper.writeValueAsString(order);

        System.err.println("order 4 json" + json);

        MessageProperties messageProperties = new MessageProperties();
        //这里注意一定要修改contentType为application/json
        messageProperties.setContentType("application/json");
        messageProperties.getHeaders().put("__TypeId__", "com.cx.temp.common.rabbitmq.spring.entity.Order");

        Message message = new Message(json.getBytes(), messageProperties);

        rabbitTemplate.send("topic001", "spring.order",message);
    }

    @Test
    public void testSendMappingMessage() throws Exception {

        ObjectMapper mapper = new ObjectMapper();

        Order order = new Order();
        order.setId("001");
        order.setName("消息订单");
        order.setContent("描述信息");

        String json1 = mapper.writeValueAsString(order);
        System.err.println("order 4 json" + json1);

        MessageProperties messageProperties1 = new MessageProperties();
        //这里注意一定要修改contentType为application/json
        messageProperties1.setContentType("application/json");
        messageProperties1.getHeaders().put("__TypeId__", "order");

        Message message1 = new Message(json1.getBytes(), messageProperties1);

        rabbitTemplate.send("topic001", "spring.order", message1);

        Packaged pack = new Packaged();
        pack.setId("002");
        pack.setName("包裹消息");
        pack.setDescription("包裹描述消息");

        String json2 = mapper.writeValueAsString(pack);
        System.err.println("pack 4 json" + json2);

        MessageProperties messageProperties2 = new MessageProperties();
        //这里注意一定要修改contentType为application/json
        messageProperties2.setContentType("application/json");
        messageProperties2.getHeaders().put("__TypeId__", "packaged");

        Message message2 = new Message(json2.getBytes(), messageProperties2);

        rabbitTemplate.send("topic001", "spring.pack", message2);
    }

　　@Test
    public void testSendExtConverterMessage() throws Exception {
        byte[] body = Files.readAllBytes(Paths.get("d://002_books", "picture.png"));
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("image/png");
        messageProperties.getHeaders().put("extName", "png");
        Message message = new Message(body, messageProperties);

        rabbitTemplate.send("", "image_queue",message);

//        byte[] body = Files.readAllBytes(Paths.get("d://002_books", "mysql.pdf"));
//        MessageProperties messageProperties = new MessageProperties();
//        messageProperties.setContentType("application/pdf");
//
//        Message message = new Message(body, messageProperties);
//
//        rabbitTemplate.send("", "pdf_queue",message);
    }

}