一、概述
四、SpringBoot+RabbitMQ整合
- 使用docker安装RabbitMQ
1）、使用docker pull 下载rabbitmq
[root@localhost ~]# docker pull registry.docker-cn.com/library/rabbitmq:3-management
3-management: Pulling from library/rabbitmq
683abbb4ea60: Already exists
30a58d97bcb5: Pull complete
...
- 启动rabbitMQ
[root@localhost ~]# docker images
REPOSITORY TAG IMAGE ID CREATED SIZE
registry.docker-cn.com/library/redis latest 71a81cb279e3 9 days ago 83.4MB
registry.docker-cn.com/library/rabbitmq 3-management 500d74765467 9 days ago 149MB
mysql 5.7 66bc0f66b7af 9 days ago 372MB
[root@localhost ~]# docker run -d -p 5672:5672 -p 15672:15672 --name myrabbitmq 500d74765467
7599303175cb42287d0f58c0b9d0db67070199670cad4f680f6348e41d6e2240
-
在浏览器输入：http://主机地址:15672 进入rabbitMQ登录页面
-
输入默认用户名：guest/guest 进入管理界面
- 搭建springboot+rabbit工程
<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>
引入spring-boot-starter-amqp包后springboot帮我们自动配置：
- 执行RabbitAutoConfiguration
- 自动配置类会帮我们自动配置连接工厂、RabbitTemplate（给RabbitMQ发送和接受消息）、AmqpAdmin（RabbitMQ系统管理功能组件）
@Configuration
@ConditionalOnMissingBean({ConnectionFactory.class})
protected static class RabbitConnectionFactoryCreator {
protected 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().intValue());
}
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().intValue());
}
factory.afterPropertiesSet();
CachingConnectionFactory connectionFactory = new CachingConnectionFactory((com.rabbitmq.client.ConnectionFactory)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().intValue());
}
if(config.getCache().getConnection().getMode() != null) {
connectionFactory.setCacheMode(config.getCache().getConnection().getMode());
}
if(config.getCache().getConnection().getSize() != null) {
connectionFactory.setConnectionCacheSize(config.getCache().getConnection().getSize().intValue());
}
if(config.getCache().getChannel().getCheckoutTimeout() != null) {
connectionFactory.setChannelCheckoutTimeout(config.getCache().getChannel().getCheckoutTimeout().longValue());
}
return connectionFactory;
}
}
-------------------------------------------------------------------------------------
@Bean
@ConditionalOnSingleCandidate(ConnectionFactory.class)
@ConditionalOnMissingBean({RabbitTemplate.class})
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
MessageConverter messageConverter = (MessageConverter)this.messageConverter.getIfUnique();
if(messageConverter != null) {
rabbitTemplate.setMessageConverter(messageConverter);
}
rabbitTemplate.setMandatory(this.determineMandatoryFlag());
Template templateProperties = this.properties.getTemplate();
Retry retryProperties = templateProperties.getRetry();
if(retryProperties.isEnabled()) {
rabbitTemplate.setRetryTemplate(this.createRetryTemplate(retryProperties));
}
if(templateProperties.getReceiveTimeout() != null) {
rabbitTemplate.setReceiveTimeout(templateProperties.getReceiveTimeout().longValue());
}
if(templateProperties.getReplyTimeout() != null) {
rabbitTemplate.setReplyTimeout(templateProperties.getReplyTimeout().longValue());
}
return rabbitTemplate;
}
------------------------------------------------------------------------------------------------
@Bean
@ConditionalOnSingleCandidate(ConnectionFactory.class)
@ConditionalOnProperty(
prefix = "spring.rabbitmq",
name = {"dynamic"},
matchIfMissing = true
)
@ConditionalOnMissingBean({AmqpAdmin.class})
public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
return new RabbitAdmin(connectionFactory);
}
- RabbitProperties 封装了RabbitMQ的所有配置
@ConfigurationProperties(
prefix = "spring.rabbitmq"
)
public class RabbitProperties {
private String host = "localhost";
private int port = 5672;
private String username;
private String password;
private final RabbitProperties.Ssl ssl = new RabbitProperties.Ssl();
private String virtualHost;
private String addresses;
private Integer requestedHeartbeat;
private boolean publisherConfirms;
private boolean publisherReturns;
private Integer connectionTimeout;
private final RabbitProperties.Cache cache = new RabbitProperties.Cache();
private final RabbitProperties.Listener listener = new RabbitProperties.Listener();
private final RabbitProperties.Template template = new RabbitProperties.Template();
private List<RabbitProperties.Address> parsedAddresses;
...
- application.properties配置
spring.rabbitmq.host=192.168.43.53
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
- 编写测试类
- 自定义MessageConverter
package com.pyy.rabbitmq.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;
/**
* Created by Administrator on 2018/7/7 0007.
*/
@Configuration
public class MyAMQPConfig {
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
}
- Exchange.direct：点对点（单播模式）
发送消息：
/**
* 1、单播（点对点）
*/
@Test
public void direct() {
//Message需要自定构造一个:定义消息体内容和消息头：
//rabbitTemplate.send(exchange, routeKey, message);
// object默认当成消息体，只需要传入需要发送的内容，自动序列化发送给rabbitmq
// rabbitTemplate.convertAndSend(exchange, routeKey, object);
User user = new User();
user.setUsername("张三");
user.setPassword("12312321");
// msg 对象被默认（SimpleMessageConverter -- byte[]）序列化后发送出去
// 通过自定义：MessageConverter 实现JSON序列化
rabbitTemplate.convertAndSend("exchange.direct", "pyy.news", user);
}
只有Queue：pyy.news 该消息队列接收到消息，其他队列不能接收到该消息。
消息内容如下：
接受消息：
/**
* 接受数据
*/
@Test
public void receve() {
Object obj = rabbitTemplate.receiveAndConvert("pyy.news");
System.out.println(obj.getClass());
System.out.println(obj);
if(obj instanceof User){
User u = (User) obj;
System.out.println(u.getUsername());
System.out.println(u.getPassword());
}
}
结果：
class com.pyy.rabbitmq.User
com.pyy.rabbitmq.User@4c27d39d
张三
12312321
- Exchange.fanout：广播模式
/**
* 2、广播
*/
@Test
public void fanout() {
//Message需要自定构造一个:定义消息体内容和消息头：
//rabbitTemplate.send(exchange, routeKey, message);
// object默认当成消息体，只需要传入需要发送的内容，自动序列化发送给rabbitmq
// rabbitTemplate.convertAndSend(exchange, routeKey, object);
User user = new User();
user.setUsername("李四");
user.setPassword("123456");
// fanout模式路由键不用指定，所有绑定到这个交换机的消息队列都能接受到该消息
rabbitTemplate.convertAndSend("exchange.fanout", "", user);
}
注意：fanout模式路由键不用指定，所有绑定到这个交换机的消息队列都能接受到该消息
- Exchange.topic 主题模式
/**
* 3、topic
*/
@Test
public void topic() {
//Message需要自定构造一个:定义消息体内容和消息头：
//rabbitTemplate.send(exchange, routeKey, message);
// object默认当成消息体，只需要传入需要发送的内容，自动序列化发送给rabbitmq
// rabbitTemplate.convertAndSend(exchange, routeKey, object);
User user = new User();
user.setUsername("王五");
user.setPassword("123456");
// topic模式路由键只有和exchange绑定的路由键规则匹配，对应的消息队列就能收到消息
rabbitTemplate.convertAndSend("exchange.topic", "*.news", user);
}
SpringBoot高级-消息-@RabbitListener&@EnableRabbit监听
- 开启RabbitListener注解支持
@EnableRabbit // 开启基于注解的RabbitMQ
@SpringBootApplication
public class Springboot02RabbitmqApplication {
public static void main(String[] args) {
SpringApplication.run(Springboot02RabbitmqApplication.class, args);
}
}
- 使用@RabbitListener注解完成消息接收
package com.pyy.rabbitmq.service;
import com.pyy.rabbitmq.model.User;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
/**
* Created by Administrator on 2018/7/8 0008.
*/
@Service
public class UserService {
/**
* 自动监听消息队列：pyy.news
* @param user
*/
@RabbitListener(queues = {"pyy.news"})
public void receive(User user) {
System.out.println("收到消息："+user);
}
//结果： 收到消息：User{username='张三', password='12312321'}
/**
* 自动监听消息队列：pyy.news
* @param message
*/
@RabbitListener(queues = {"pyy"})
public void receive(Message message) {
System.out.println("消息头："+message.getMessageProperties());
System.out.println("消息体："+message.getBody());
}
}
系统会自动监听指定名称的消息队列，只有有消息自动消费。
AmqpAdmin 管理rabbitmq相关操作：
package com.pyy.rabbitmq;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.*;
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 Springboot02RabbitmqApplicationTests1 {
@Autowired
RabbitTemplate rabbitTemplate;
@Autowired
AmqpAdmin amqpAdmin;
/**
* 创建交换机：exchange
*/
@Test
public void createExchange() {
amqpAdmin.declareExchange(new DirectExchange("amqpadmin.DirectExchange"));
amqpAdmin.declareExchange(new FanoutExchange("amqpadmin.FanoutExchange"));
amqpAdmin.declareExchange(new TopicExchange("amqpadmin.TopicExchange"));
System.out.println("创建完毕");
}
/**
* 创建队列：queye
*/
@Test
public void createQueue() {
amqpAdmin.declareQueue(new Queue("amqpadmin.queue", true));
System.out.println("创建完毕");
}
/**
* 创建绑定规则：binding
*/
@Test
public void createBinding() {
/**
String destination: 目的地
String exchange： 交换机
String routingKey： 路由键
Map<String, Object> arguments：
Binding.DestinationType destinationType:目的地类型
*/
amqpAdmin.declareBinding(new Binding("amqpadmin.queue", Binding.DestinationType.QUEUE, "amqpadmin.DirectExchange", "amqp.haha", null));
System.out.println("创建完毕");
}
}
