springBoot整合rabbitMQ的方法详解

更新时间：2021年04月20日 09:59:08 作者：喜羊羊love红太狼

这篇文章主要介绍了springBoot整合rabbitMQ的方法详解,本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下

引入pom

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.4.5</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.wxy</groupId>
	<artifactId>test-rabbitmq</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>test-rabbitmq</name>
	<description>Demo project for Spring Boot</description>
	<properties>
		<java.version>1.8</java.version>
	</properties>
	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
 
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.amqp</groupId>
			<artifactId>spring-rabbit-test</artifactId>
			<scope>test</scope>
		</dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
 
	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>
 
</project>

测试

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<String,Object> 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);
                }
            }
        }

将默认消息类型转化成自定义json格式

第一：上面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);
    }

配置一个messageConversert(org.springframework.amqp.support.converter包中的)

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();
    }
}

再次发送消息体json格式

使用注解@RabbitListener监听

监听多个队列

@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<String, Object> arguments
		amqpAdmin.declareBinding(new Binding("amqp.queue", Binding.DestinationType.QUEUE,
				"amqpexchange.direct","wxy.news", null));
 
	}

到此这篇关于springBoot整合rabbitMQ的方法详解的文章就介绍到这了,更多相关springBoot整合rabbitMQ内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Java 多线程并发AbstractQueuedSynchronizer详情
这篇文章主要介绍了Java 多线程并发AbstractQueuedSynchronizer详情，文章围绕主题展开想象的内容介绍，具有一定的参考价值，感兴趣的小伙伴可以参考一下
2022-06-06
详解SpringBoot可执行Jar包运行原理
SpringBoot有一个很方便的功能就是可以将应用打成可执行的Jar，那么大家有没想过这个Jar是怎么运行起来的呢，本篇博客就来介绍下 SpringBoot可执行Jar包的运行原理,需要的朋友可以参考下
2023-05-05
Java中的List与Set转换方式
Java中,List和Set是两种基本的集合类型,它们在允许重复元素、元素顺序、实现类以及性能方面有着明显的区别,List允许重复元素并保持元素插入的顺序,常见实现有ArrayList、LinkedList和Vector；Set不允许重复元素
2024-11-11
详解非spring框架下使用querydsl的方法
Querydsl是一个采用API代替拼凑字符串来构造查询语句，可跟 Hibernate 和 JPA 等框架结合使用。本文介绍的是非spring环境下querydsl JPA整合使用，感兴趣的小伙伴们可以参考一下
2019-01-01
Java中的CountDownLatch同步工具类使用解析
这篇文章主要介绍了Java中的CountDownLatch使用解析,CountDownLatch初始化的时候必须指定一个count,await方法会一直阻塞直到调用countdown方法,count为0,当count为0时,所有的等待线程都会被释放,需要的朋友可以参考下
2023-12-12
Java FastJson使用教程
这篇文章主要介绍了如何使用FastJson，帮助大家将 Java 对象转换为 JSON 格式，感兴趣的朋友可以了解下
2020-10-10
Java中final关键字和final的四种用法实例
final关键字代表最终的、不可改变的,下面这篇文章主要给大家介绍了关于Java中final关键字和final的四种用法实例,文中通过图文以及实例代码介绍的非常详细,需要的朋友可以参考下
2023-02-02
使用@ControllerAdvice同时配置过滤多个包
这篇文章主要介绍了使用@ControllerAdvice同时配置过滤多个包的操作，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-06-06
浅谈Spring-boot事件监听
这篇文章主要介绍了浅谈Spring-boot事件监听，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-09-09
95%的Java程序员人都用不好Synchronized详解
这篇文章主要为大家介绍了95%的Java程序员人都用不好Synchronized详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2023-03-03

springBoot整合rabbitMQ的方法详解

相关文章

最新评论

大家感兴趣的内容

最近更新的内容

常用在线小工具