SpringBoot高级-消息-RabbitTemplate发送接受消息&序列化机制[亲测有效]

大家好，欢迎来到IT知识分享网。

引入了spring-boot-starter-amqp模块,他引入了spring-messaging模块,包括引入了spring-rabbit模块,怎么配置使用呢,

<dependency>
	<groupId>org.springframework</groupId>
	<artifactId>spring-messaging</artifactId>
</dependency>

SpringBoot主要是引入了相关场景依赖,一切都是自动配置的,我们可以来分析一下原理,rabbitmq给我们自动配置了哪些

<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 
http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.learn</groupId>
  <artifactId>springboot-02-amqp</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>
  
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>1.5.12.RELEASE</version>
		<relativePath/> 
	</parent>
	
	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>1.8</java.version>
		<thymeleaf.version>3.0.9.RELEASE</thymeleaf.version>
		<thymeleaf-layout-dialect.version>2.2.2</thymeleaf-layout-dialect.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>
	</dependencies>
	
	<!-- 这个插件,可以将应用打包成一个可执行的jar包 -->
	<build>
	    <plugins>
	        <plugin>
	            <groupId>org.springframework.boot</groupId>
	            <artifactId>spring-boot-maven-plugin</artifactId>
	        </plugin>
	    </plugins>
	</build>
  
  
</project>

我们就找一下和rabbitmq相关的类,有一个叫RabbitAutoConfiguration,这个自动配置类里面帮我们做了什么,

我们在这记录一下,自动配置类,我们配了哪些,进来看,这一块我们配了连接工厂,这是人家帮我们配好的连接

工厂CachingConnectionFactory,有自动配置了连接工厂,从这里可以获取,Rabbitmq的连接,比如我们的主机地址,

用户名,密码,还有VisualHost等等

@Configuration
@ConditionalOnMissingBean(ConnectionFactory.class)
protected static class RabbitConnectionFactoryCreator {

	@Bean
	public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties config)
			throws Exception {
		RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean();
		if (config.determineHost() != null) {
			factory.setHost(config.determineHost());
		}
		factory.setPort(config.determinePort());
		if (config.determineUsername() != null) {
			factory.setUsername(config.determineUsername());
		}
		if (config.determinePassword() != null) {
			factory.setPassword(config.determinePassword());
		}
		if (config.determineVirtualHost() != null) {
			factory.setVirtualHost(config.determineVirtualHost());
		}
		if (config.getRequestedHeartbeat() != null) {
			factory.setRequestedHeartbeat(config.getRequestedHeartbeat());
		}
		RabbitProperties.Ssl ssl = config.getSsl();
		if (ssl.isEnabled()) {
			factory.setUseSSL(true);
			if (ssl.getAlgorithm() != null) {
				factory.setSslAlgorithm(ssl.getAlgorithm());
			}
			factory.setKeyStore(ssl.getKeyStore());
			factory.setKeyStorePassphrase(ssl.getKeyStorePassword());
			factory.setTrustStore(ssl.getTrustStore());
			factory.setTrustStorePassphrase(ssl.getTrustStorePassword());
		}
		if (config.getConnectionTimeout() != null) {
			factory.setConnectionTimeout(config.getConnectionTimeout());
		}
		factory.afterPropertiesSet();
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory(
				factory.getObject());
		connectionFactory.setAddresses(config.determineAddresses());
		connectionFactory.setPublisherConfirms(config.isPublisherConfirms());
		connectionFactory.setPublisherReturns(config.isPublisherReturns());
		if (config.getCache().getChannel().getSize() != null) {
			connectionFactory
					.setChannelCacheSize(config.getCache().getChannel().getSize());
		}
		if (config.getCache().getConnection().getMode() != null) {
			connectionFactory
					.setCacheMode(config.getCache().getConnection().getMode());
		}
		if (config.getCache().getConnection().getSize() != null) {
			connectionFactory.setConnectionCacheSize(
					config.getCache().getConnection().getSize());
		}
		if (config.getCache().getChannel().getCheckoutTimeout() != null) {
			connectionFactory.setChannelCheckoutTimeout(
					config.getCache().getChannel().getCheckoutTimeout());
		}
		return connectionFactory;
	}

}

都是从config里面得到的,config就是我们的RabbitProperties,我们点进来

@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitProperties {

这里面封装了Rabbitmq的所有配置,那我们就来配置他们,找到我们的配置文件,

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.port=5672
#spring.rabbitmq.virtual-host=

首先rabbitmq的主机地址,我们写上我们的ip地址,http不用写,http是web的访问地址,还有rabbitmq的用户名,我们也写上,

guest,guest用户,还有我们的密码,也叫guest,还有rabbitmq的端口,这个端口呢,不写默认就是5672,这个可以不写,virtual-host

属性我们之前有说过,是一定要指定的,我们也可以不指定,virtual-host如果我们不写,如果是空字符串也是访问"/"

/**
 * If addresses have been set and the first address has a virtual host it is returned.
 * Otherwise returns the result of calling {@code getVirtualHost()}.
 * @return the virtual host or {@code null}
 * @see #setAddresses(String)
 * @see #getVirtualHost()
 */
public String determineVirtualHost() {
	if (CollectionUtils.isEmpty(this.parsedAddresses)) {
		return getVirtualHost();
	}
	Address address = this.parsedAddresses.get(0);
	return address.virtualHost == null ? getVirtualHost() : address.virtualHost;
}

都可以通过配置文件来知道,这个配置我们基本写完,但是我们如何给rabbitmq发送以及接收消息呢,我们可以看自动配置,

这个自动配置除了看工厂,它还会给我们放一个RabbitTemplate,给容器中加一个他,给RabbitMQ接收和发送消息的,就像以前

的JdbcTemplate,RedisTemplate,那么他还有什么呢,还可以来这里继续来看,除了RabbitTemplate,他还放了AmqpAdmin,这

AmqpAdmin呢,它是Rabbitmq的管理组件,Rabbitmq系统管理组件,他不来发消息和收消息,可以帮我们声明一个队列,创建

一个交换器,这个是自动配置给我们放的组件,我们就用RabbitTemplate,我在这个测试类里面

我们尝试给消息队列里面发送一些内容,我们来测试几个消息,第一个单播下的点对点消息,

我们将一个消息发送队列里面,我们可以使用rabbitTemplate,两个方法,send传入exchange,就是交换器,

消息先给交换器,根据路由件放到哪个队列里面,后面这个就是我们要发的消息,这个是需要我们自己来

编辑这个消息,第一个传一个交换器,第二个是传一个路由件routekey,第三个传一个message,他的好处呢就是,

特别是message需要自己定义,Message需要自己定义,需要自己构造一个,自己构造的时候呢,我们来看一下这个

Message,我们要发的消息内容,amqp这个核心包里面

org.springframework.amqp.core.Message

你要把你发的消息序列号成一个数组,还有你消息的头信息,

public Message(byte[] body, MessageProperties messageProperties) { //NOSONAR
	this.body = body; //NOSONAR
	this.messageProperties = messageProperties;
}

可以定制消息体内容和消息头,这是rabbitTemplate.send,我们常见的是convertAndSend,转化并发送,我们只需要

传入交换器,路由件,还有数据对象,这个对象默认就会被当成消息体,而且会自动转换,我们只需要传入要发送的对象,

只需要传入发送的对象,自动序列化给rabbitmq,这个object是被当成消息体的,默认当成消息体,我们就可以用这个方法,

我们就直接用下面这个方法,我们要发点对点消息,我们就要按照交换器的规则,direct能够直接派发给一个队列,我们就连上

这个交换器,我们就把消息给exchange.direct,我们连上这个交换器,这个交换器我们绑定了很多的路由件,比如china,

路由件是他,交给这个队列,我们就用china.news,我们用这个路由件,要发送的内容是什么,就随便写一个map,我可以写一个

map,；里面存一些数据放进去,比如msg存一些提示消息,这是我们要发的数据,我来发送测试一下,我们看这个数据能不能到达,

这个运行完成了,我们来我们的消息队列来看一下

localhost:15672

我们刚才发的是china.news

消息是这个样子的

是因为序列号默认是JAVA的序列化方式,给我们序列化的对象,解码以后的样子,对象呗默认序列化以后,

发送出去,发送点对点消息,我们要接收到这个消息,怎么办呢,可以用这个API,我们同样用rabbitTemplate,

有这个receive方法可以接收,包括还有receiveAndConverter,接收并转化,如果用这个接收,你收来自哪个消息

队列的消息,会给你把这个消息转成Messager,里面就只有消息头和消息体,如果用receiveAndConverter,可以把消息

体转成我们想要的对象,我们最终的消息是发到china.news,这个里面了,所以我们要收数据,也是收这个消息队列的,

在这里写上消息队列的名称,然后把收过来的数据拿过来,这个数据的类型是什么,再来打印的数据是什么

package com.learn.springboot;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

import org.junit.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;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringBootAmqpApplicationTests {
	
	@Autowired
	RabbitTemplate rabbitTemplate;
	
	/**
	 * 1.单播(点对点)
	 */
	@Test
	public void contextLoads() {
		// Message需要自己构造一个,定义消息体内容和消息头
//		rabbitTemplate.send(exchange, routingKey, message);
		
		// object默认当成消息体,只需要传入要发送的对象,自动序列化发送给rabbitmq
//		rabbitTemplate.convertAndSend(routingKey, message, messagePostProcessor);
		
		Map<String,Object> map = new HashMap<String,Object>();
		map.put("msg", "这是第一个消息");
		map.put("data", Arrays.asList("helloworld",1231,true));		
		// 对象被默认序列化以后发送出去
		rabbitTemplate.convertAndSend("exchange.direct", "china.news",map);
	}
	
	@Test
	public void receive() {
		Object o = rabbitTemplate.receiveAndConvert("china.news");
		System.out.println(o.getClass());
		System.out.println(o);
	}
}

class java.util.HashMap
{msg=这是第一个消息, data=[helloworld, 1231, true]}

数据收到的类型是HashMap,拿到这些数据没什么问题,所以我们也能准确的收到数据,但是这个数据收到以后,

来看这个消息队列里面,这个消息队列就没有了

确实已经没有了,刚才已经收到了,接收数据,有时候我想把数据序列化成JSON的格式,那种序列化的结果有点不好看,

如何将数据自动的转为JSON发送出去,其实非常的简单,我们到RabbitTemplate这里来看,为什么刚才是那种序列化结果,

原因我们来往下翻,原因有一个消息转换器

private volatile MessageConverter messageConverter = new SimpleMessageConverter();

这个转换器呢,默认是SimpleMessageConverter,这个和MessageConverter呢,我们来序列化数据的时候,就是按照JDK的

序列化规则,

content = SerializationUtils.deserialize(
	createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));

我们可以给他换一个MessageConverter

org.springframework.amqp.support.converter.MessageConverter

我们来写上配置,比如我们在config包下写一个MyAMQPConfig,这个Config我们来写一个@Configuration,

我们就自己来放一个MessageConverter,这是我们的MessageConverter,来放哪个MessageConverter呢,就来看他有哪些,

抽象AbstractJsonMessageConverter有跟JSON有关的,Jackson2JsonMessageConverter,我们换成他,我们自动配置也会生效,

当我们来配置RabbitTemplate的时候,如果我们有自己定义的MessageConverter,也会给我们设置进来,

@Bean
@ConditionalOnSingleCandidate(ConnectionFactory.class)
@ConditionalOnMissingBean(RabbitTemplate.class)
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
	RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
	MessageConverter messageConverter = this.messageConverter.getIfUnique();
	if (messageConverter != null) {
		rabbitTemplate.setMessageConverter(messageConverter);
	}

我们再来测试这个消息,我们把消息发出去,来看还是不是原来的消息

package com.learn.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 MyAMQPConfig {

	@Bean
	public MessageConverter messageConvert() {
		return new Jackson2JsonMessageConverter();
	}
}

消息就是我们的JSON数据,而且还有定义的消息头,消息头有我们key的类型,我们整个消息的类型,

application/json,序列化能过去,存起来,能不能取出消息,我们看到这块也是没有问题的,

class java.util.HashMap
{msg=这是第一个消息, data=[helloworld, 1231, true]}

比如我们发送Book对象,接下来我们来发一个图书的对象,我们来new一个book

package com.learn.bean;

public class Book {

	private String bookName;
	
	private String author;
	
	public Book() {
		super();
	}

	public Book(String bookName, String author) {
		super();
		this.bookName = bookName;
		this.author = author;
	}

	public String getBookName() {
		return bookName;
	}

	public void setBookName(String bookName) {
		this.bookName = bookName;
	}

	public String getAuthor() {
		return author;
	}

	public void setAuthor(String author) {
		this.author = author;
	}

	@Override
	public String toString() {
		return "Book [bookName=" + bookName + ", author=" + author + "]";
	}
	
}

package com.learn.springboot;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

import org.junit.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 com.learn.bean.Book;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringBootAmqpApplicationTests {
	
	@Autowired
	RabbitTemplate rabbitTemplate;
	
	/**
	 * 1.单播(点对点)
	 */
	@Test
	public void contextLoads() {
		// Message需要自己构造一个,定义消息体内容和消息头
//		rabbitTemplate.send(exchange, routingKey, message);
		
		// object默认当成消息体,只需要传入要发送的对象,自动序列化发送给rabbitmq
//		rabbitTemplate.convertAndSend(routingKey, message, messagePostProcessor);
		
		Map<String,Object> map = new HashMap<String,Object>();
		map.put("msg", "这是第一个消息");
		map.put("data", Arrays.asList("helloworld",1231,true));		
		// 对象被默认序列化以后发送出去
		rabbitTemplate.convertAndSend("exchange.direct", "china.news",new Book("西游记","吴承恩"));
	}
	
	@Test
	public void receive() {
		Object o = rabbitTemplate.receiveAndConvert("china.news");
		System.out.println(o.getClass());
		System.out.println(o);
	}
}

能不能反序列化获取出来呢,我们先用Object对象来接收,

class com.learn.bean.Book
Book [bookName=西游记, author=吴承恩]

这个Class拿到的是Book对象,强转也可以在这里强转,消息序列化和反序列化,我们测试的是一个单播,那如果想

广播,不管是单播还是广播,我们发给对应的exchange,我们再来测试一下,广播类型的转换器,然后路由件就不用写了,

因为广播是不用管路由件的,我们来测试一下广播

	/**
	 * 广播
	 */
	@Test
	public void sendMsg() {
		rabbitTemplate.convertAndSend("exchange.fanout","",new Book("三国演义","罗贯中"));
	}