springBoot整合rabbitMQ的方法详解

<&＃63;xml version="1.0" encoding="UTF-8"&＃63;>

	4.0.0
	
		org.springframework.boot
		spring-boot-starter-parent
		2.4.5
		 
	
	com.wxy
	test-rabbitmq
	0.0.1-SNAPSHOT
	test-rabbitmq
	Demo project for Spring Boot
	
		1.8
	
	
		
			org.springframework.boot
			spring-boot-starter-amqp
		
		
			org.springframework.boot
			spring-boot-starter-web
		
 
		
			org.springframework.boot
			spring-boot-starter-test
			test
		
		
			org.springframework.amqp
			spring-rabbit-test
			test
		
        
            junit
            junit
            test
        
    
 
	
		
			
				org.springframework.boot
				spring-boot-maven-plugin
			
		
	
 

package com.wxy.rabbit;
 
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
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 java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
 
@RunWith(SpringRunner.class)
@SpringBootTest
class TestRabbitmqApplicationTests {
 
	@Autowired
	RabbitTemplate rabbitTemplate;
 
	@Test
	public  void sendmessage() {
		String exchange  = "exchange.direct";
		String routingkey  = "wxy.news";
		//object为消息发送的消息体，可以自动实现消息的序列化
		Map msg = new HashMap<>();
		msg.put("msg","使用mq发送消息");
		msg.put("data", Arrays.asList("helloword",123456,true));
		rabbitTemplate.convertAndSend(exchange, routingkey,msg);
	}
 
 
	@Test
	public  void receive() {
		Object object  = rabbitTemplate.receiveAndConvert("wxy.news");
		System.out.println(object);
	}
 
}

 ###############在RabbitTemplate默认使用的是SimpleMessageConverter#######
  private MessageConverter messageCOnverter= new SimpleMessageConverter();
 
 
  ###############源码：使用SerializationUtils.deserialize###############
   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.startsWith("text")) {
                String encoding = properties.getContentEncoding();
                if (encoding == null) {
                    encoding = this.defaultCharset;
                }
 
                try {
                    cOntent= new String(message.getBody(), encoding);
                } catch (UnsupportedEncodingException var8) {
                    throw new MessageConversionException("failed to convert text-based Message content", var8);
                }
            } else if (contentType != null && contentType.equals("application/x-java-serialized-object")) {
                try {
                    cOntent= SerializationUtils.deserialize(this.createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
                } catch (IllegalArgumentException | IllegalStateException | IOException var7) {
                    throw new MessageConversionException("failed to convert serialized Message content", var7);
                }
            }
        }

第一：上面SimpleMessageConverter是org.springframework.amqp.support.converter包下MessageConverter接口的一个实现类
 
第二：查看该接口MessageConverter下支持哪些消息转化
ctrl+H查看该接口中的所有实现类
 
第三步：找到json相关的convert

RabbitTemplateConfigurer中定义if (this.messageConverter != null)则使用配置的messageConverter

 ################## if (this.messageConverter != null)则使用配置的messageConverter 
public void configure(RabbitTemplate template, ConnectionFactory connectionFactory) {
        PropertyMapper map = PropertyMapper.get();
        template.setConnectionFactory(connectionFactory);
        if (this.messageConverter != null) {
            template.setMessageConverter(this.messageConverter);
        }
 
        template.setMandatory(this.determineMandatoryFlag());
        Template templateProperties = this.rabbitProperties.getTemplate();
        if (templateProperties.getRetry().isEnabled()) {
            template.setRetryTemplate((new RetryTemplateFactory(this.retryTemplateCustomizers)).createRetryTemplate(templateProperties.getRetry(), Target.SENDER));
        }
 
        templateProperties.getClass();
        map.from(templateProperties::getReceiveTimeout).whenNonNull().as(Duration::toMillis).to(template::setReceiveTimeout);
        templateProperties.getClass();
        map.from(templateProperties::getReplyTimeout).whenNonNull().as(Duration::toMillis).to(template::setReplyTimeout);
        templateProperties.getClass();
        map.from(templateProperties::getExchange).to(template::setExchange);
        templateProperties.getClass();
        map.from(templateProperties::getRoutingKey).to(template::setRoutingKey);
        templateProperties.getClass();
        map.from(templateProperties::getDefaultReceiveQueue).whenNonNull().to(template::setDefaultReceiveQueue);
    }

package com.wxy.rabbit.config;
 
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class MessageConverConfig {
 
    @Bean
    public MessageConverter getMessageConvert(){
        return  new Jackson2JsonMessageConverter();
    }
}

@RabbitListener(queues = {"wxy.news","wxy.emps"})

@RabbitListener(queues = "wxy.news")

package com.wxy.rabbit.service;
 
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
 
@Service
public class RabbitMqReceiveService {
 
    @RabbitListener(queues = {"wxy.news","wxy.emps"})
    public void  getReceiveMessage(){
        System.out.println("监听到性的消息");
    }
 
 
    @RabbitListener(queues =  {"wxy.news","wxy.emps"})
    public void  getReceiveMessageHead(Message message){
        System.out.println(message.getBody());
        System.out.println( message.getMessageProperties());
    }
 
}

@Test
	public  void create() {
		//创建一个点对点的交换器
		amqpAdmin.declareExchange(new DirectExchange("amqpexchange.direct"));
		//创建一个队列
		// String name,：队列名称
		// boolean durable :持久化
		amqpAdmin.declareQueue(new Queue("amqp.queue",true));
		//绑定
		//String destination, Binding.DestinationType destinationType, String exchange, String routingKey
		//  @Nullable Map arguments
		amqpAdmin.declareBinding(new Binding("amqp.queue", Binding.DestinationType.QUEUE,
				"amqpexchange.direct","wxy.news", null));
 
	}