1. AMQP 核心组件

RabbitAdmin

SpringAMQP声明

RabbitTemplate

SimpleMessageListenerContainer

MessageListenerAdapter

MessageConverter

2. RabbitAdmin

RabbitAdmin

注意：

autoStartUp必须要设置为true,否则Spring容器不会加载RabbitAdmin类

RabbitAdmin底层实现就是从Spring容器中获取Exchange、Bingding、RoutingKey以及Queue的@Bean声明

使用RabbitTemplate的execute方法执行对应的什么、修改、删除等一系列RabbitMQ基础功能操作

例如：添加一个交换机、删除一个绑定、清空一个队列里的消息等等

3. SpringAMQP声明和RabbitTemplate

SpringAMQP声明

以前，在Rabbit基础API里面声明一个Exchange、声明一个绑定、一个队列

ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.11.76");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
String exchangeName = "test_ack_exchange";
String queueName = "test_ack_queue";
String routingKey = "ack.#";
channel.exchangeDeclare(exchangeName, "topic", true, false, null);
channel.queueDeclare(queueName, true, false, false, null);
channel.queueBind(queueName, exchangeName, routingKey);
// 手工签收 必须要关闭 autoAck = false
channel.basicConsume(queueName, false, new MyConsumer(channel));

使用SpringAMQP去声明，就需要使用SpringAMQP的如下模式，即声明@Bean方式

 /**  
     * 针对消费者配置  
     * 1. 设置交换机类型  
     * 2. 将队列绑定到交换机  
        FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念  
        HeadersExchange ：通过添加属性key-value匹配  
        DirectExchange:按照routingkey分发到指定队列  
        TopicExchange:多关键字匹配  
     */  
    @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.*");  
    }

RabbitTemplate

RabbitTemplate，即消息模板

我们在与SpringAMQP整合的时候进行发送消息的关键词

该类提供了丰富的发送消息方法，包括可靠性投递消息方法、回调监听消息接口ConfirmCallback、返回值确认接口ReturnCallback等等。同样我们需要进行注入到Spring容器中，然后直接使用

在与Spring整合时需要实例化，但是在与SpringBoot整合时，在配置文件里添加配置即可

4. SimpleMessageListenerContainer

SimpleMessageListenerContainer

简单消息监听容器

这个类非常的强大，我们可以对它进行很多设置，对于消费者的配置项，这个类都可以满足

监听队列(多个队列)、自动启动、自动声明功能

设置事务特性、事务管理器、事务属性、事务容器(并发)、是否开启事务、回滚消息等

设置消费者数量、最小最大数量、批量消费

设置消息确认和自动确认模式、是否重回队列、异常捕捉handler函数

设置消费者标签生成策略、是否独占模式、消费者属性等

设置具体的监听器、消息转换器等等。

注意：

SimpleMessageListenerContainer可以进行动态设置，比如在运行中的应用可以动态的修改其消费者数量的大小、接收消息的模式等

很多机遇RabbitMQ的自制定话后端管控台在进行动态设置的时候，也是根据这一特性去实现的。所以可以看出SpringAMQP非常的强大

SimpleMessageListenerContainer原理：为什么可以动态感知配置变更？

SimpleMessageListenerContainer用一个List<String> 来存储所有queue，当要添加queue时，就往这个List添加，然后调用queuesChanged()方法，把所有的consumer都取消订阅，然后把他们都移除，然后有多少个consumer再重新new相同数量的consumer出来，然后每个consumer都把List中的每个都都订阅，到此就完成了。

5. MessageListenerAdapter

MessageListenerAdapter

默认方法名就是叫handleMessage。当然也可以自己去指定设置。通过messageListenerAdapter的代码我们可以看出如下核心属性

defaultListenerMethod默认监听方法名称：用于设置监听方法名称

Delegate 委托对象：实际真实的委托对象，用于处理消息

queueOrTagToMethodName 队列标识与方法名称组成集合

可以一一进行队列与方法名称的匹配

队列和方法名称绑定，即指定队列里的消息会被绑定的方法所接受处理

6. MessageConverter

MessageConverter

我们在进行发送消息的时候，正常情况下消息体为二进制的数据方式进行传输，如果希望内部帮我们进行转换，或者指定自定义的转换器，就需要用到MessageConverter

自定义常用转换器：MessageConverter，一般来讲都需要实现这个接口

重写下面两个方法：

toMessage:java对象转换为Message

fromMessage:Message对象转换为java对象

Json转换器：Jackson2JsonMessageConverter:可以进行Java对象的转换功能

DefaultJackson2JavaTypeMapper映射器：可以进行java对象的映射关系

自定义二进制转换器：比如图片类型、PDF、PPT、流媒体

7 代码演示

7.1 引入Pom文件

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
  </dependency>
  <!--amqp-->
  <dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>3.6.5</version>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
</dependencies>

7.2 配置类

package com.bfxy.spring;
import java.util.UUID;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
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.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import com.bfxy.spring.adapter.MessageDelegate;
import com.bfxy.spring.convert.ImageMessageConverter;
import com.bfxy.spring.convert.PDFMessageConverter;
import com.bfxy.spring.convert.TextMessageConverter;
@Configuration
@ComponentScan({"com.bfxy.spring.*"})
public class RabbitMQConfig {
  @Bean
  public ConnectionFactory connectionFactory(){
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
    connectionFactory.setAddresses("192.168.11.76:5672");
    connectionFactory.setUsername("guest");
    connectionFactory.setPassword("guest");
    connectionFactory.setVirtualHost("/");
    return connectionFactory;
  }
  @Bean
  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:多关键字匹配  
     */  
    @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 Binding binding002() {  
        return BindingBuilder.bind(queue002()).to(exchange002()).with("rabbit.*");  
    } 
    @Bean  
    public Queue queue003() {  
        return new Queue("queue003", true); //队列持久  
    }
    @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.setExposeListenerChannel(true);
      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);
      }
    });
      */
      /**
       * 1 适配器方式. 默认是有自己的方法名字的：handleMessage
        // 可以自己指定一个方法的名字: consumeMessage
        // 也可以添加一个转换器: 从字节数组转换为String
      MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
      adapter.setDefaultListenerMethod("consumeMessage");
      adapter.setMessageConverter(new TextMessageConverter());
      container.setMessageListener(adapter);
      */
      /**
       * 2 适配器方式: 我们的队列名称 和 方法名称 也可以进行一一的匹配
       * 
      MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageDelegate());
      adapter.setMessageConverter(new TextMessageConverter());
      Map<String, String> queueOrTagToMethodName = new HashMap<>();
      queueOrTagToMethodName.put("queue001", "method1");
      queueOrTagToMethodName.put("queue002", "method2");
      adapter.setQueueOrTagToMethodName(queueOrTagToMethodName);
      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<String, Class<?>>();
    idClassMapping.put("order", com.bfxy.spring.entity.Order.class);
    idClassMapping.put("packaged", com.bfxy.spring.entity.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 textConvert = new TextMessageConverter();
    convert.addDelegate("text", textConvert);
    convert.addDelegate("html/text", textConvert);
    convert.addDelegate("xml/text", textConvert);
    convert.addDelegate("text/plain", textConvert);
    Jackson2JsonMessageConverter jsonConvert = new Jackson2JsonMessageConverter();
    convert.addDelegate("json", jsonConvert);
    convert.addDelegate("application/json", jsonConvert);
    ImageMessageConverter imageConverter = new ImageMessageConverter();
    convert.addDelegate("image/png", imageConverter);
    convert.addDelegate("image", imageConverter);
    PDFMessageConverter pdfConverter = new PDFMessageConverter();
    convert.addDelegate("application/pdf", pdfConverter);
    adapter.setMessageConverter(convert);
    container.setMessageListener(adapter);
      return container;
    }
}

7.3 消息转换器

图片转换器

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.UUID;
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 ImageMessageConverter implements MessageConverter {
  @Override
  public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
    throw new MessageConversionException(" convert error ! ");
  }
  @Override
  public Object fromMessage(Message message) throws MessageConversionException {
    System.err.println("-----------Image 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;
  }
}

PDF转换器

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.UUID;
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 PDFMessageConverter implements MessageConverter {
  @Override
  public Message toMessage(Object object, 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;
  }
}

文本转换器

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 object, MessageProperties messageProperties) throws MessageConversionException {
    return new Message(object.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();
  }
}

7.4 实体

public class Order {
  private String id;
  private String name;
  private String content;
  public Order() {
  }
  public Order(String id, String name, String content) {
    this.id = id;
    this.name = name;
    this.content = content;
  }
  public String getId() {
    return id;
  }
  public void setId(String id) {
    this.id = id;
  }
  public String getName() {
    return name;
  }
  public void setName(String name) {
    this.name = name;
  }
  public String getContent() {
    return content;
  }
  public void setContent(String content) {
    this.content = content;
  }
}

public class Packaged {
  private String id;
  private String name;
  private String description;
  public Packaged() {
  }
  public Packaged(String id, String name, String description) {
    this.id = id;
    this.name = name;
    this.description = description;
  }
  public String getId() {
    return id;
  }
  public void setId(String id) {
    this.id = id;
  }
  public String getName() {
    return name;
  }
  public void setName(String name) {
    this.name = name;
  }
  public String getDescription() {
    return description;
  }
  public void setDescription(String description) {
    this.description = description;
  }
}

7.5 测试类

import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.HashMap;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
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.SpringRunner;
import com.bfxy.spring.entity.Order;
import com.bfxy.spring.entity.Packaged;
import com.fasterxml.jackson.databind.ObjectMapper;
@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests {
  @Test
  public void contextLoads() {
  }
  @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.#")); //指定路由Key
    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() {
      @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.abc", "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.bfxy.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);
  }
}

RabbitMQ精讲6：与Spring AMQP整合实战

1. AMQP 核心组件

2. RabbitAdmin

3. SpringAMQP声明和RabbitTemplate

SpringAMQP声明

RabbitTemplate

4. SimpleMessageListenerContainer

SimpleMessageListenerContainer原理：为什么可以动态感知配置变更？

5. MessageListenerAdapter

6. MessageConverter

7 代码演示

7.1 引入Pom文件

7.2 配置类

7.3 消息转换器

图片转换器

PDF转换器

文本转换器

7.4 实体

7.5 测试类

热门文章

最新文章

相关课程

相关电子书

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

RabbitMQ精讲6：与Spring AMQP整合实战

1. AMQP 核心组件

2. RabbitAdmin

3. SpringAMQP声明和RabbitTemplate

SpringAMQP声明

RabbitTemplate

4. SimpleMessageListenerContainer

SimpleMessageListenerContainer原理：为什么可以动态感知配置变更？

5. MessageListenerAdapter

6. MessageConverter

7 代码演示

7.1 引入Pom文件

7.2 配置类

7.3 消息转换器

图片转换器

PDF转换器

文本转换器

7.4 实体

7.5 测试类

热门文章

最新文章

相关课程

相关电子书