前言
目的主要是学习RabbitMQ其中一种Direct交换机，大概会简单介绍学习为主：毕竟还是要来演示Springboot整合RabbitMQ注解的方式来使用
一.Direct交换机模式
1.旁白
Direct是直接交换机模式，也可以说是一对一的关系。
生产者和消费者，具有相同的交换机名称（Exchange）、交换机类型和相同的密匙（routingKey），那么消费者即可成功获取到消息。
用更直接话来讲就是direct交换机: 通过routingKey和exchange决定的那个唯一的queue可以接收消息。
当然也可以用更官方的话来说，消息中的路由键(routing key)如果和Binding中的binding key 一致，交换器就将消息发到对应的队列中，路由键与队列完全匹配，单播模式。一对一绑定。容易配置和使用
2.图说
红色：Producer代表着生产者：也就是发消息的一端,Consumer代表着消费生产者的消息
绿色： 声明一个Exchange(交换机),Queues声明多个队列,Bindings声明绑定交换机和队列的
黄色：声明的路由键，只对应相同routingKey才能被消息
直接大概可以这样说 生产者生产消息到绑定好的交换机和队列，消费者根据对应的routingkey去消费消息
image.png
3.举例
image.png
二.Springboot整合Rabbimq实现direct
准备创建工程项目，目录结构如下：
image.png
2.1 统一配置pom.xml依赖
- 父工程
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.4.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.mi</groupId>
<artifactId>springboot-rabbitmq-day1</artifactId>
<packaging>pom</packaging>
<version>1.0-SNAPSHOT</version>
<modules>
<module>springboot-send-rabbitmq</module>
<module>springboot-recive-rabbitmq</module>
</modules>
- 发送工程和接受工程一样
<parent>
<artifactId>springboot-rabbitmq-day1</artifactId>
<groupId>com.mi</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.mi</groupId>
<artifactId>springboot-send-rabbitmq</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>2.1.2.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
</dependencies>
2.2 统一配置 application.properties
#发送端8082，接受端8081
server.port=8082
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/food?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=GMT%2B8
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.rabbitmq.address=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
#消息确认机制
spring.rabbitmq.listener.direct.acknowledge-mode=auto
2.3 统一配置 application启动器
发送端和接受端基本一样
@SpringBootApplication
@MapperScan(value = "com.xxx.xxx",annotationClass = Mapper.class)
@ComponentScan("com.xxx.xxx")
public class ReciveApplication {
public static void main(String args[]) {
SpringApplication.run(ReciveApplication.class, args);
}
}
3 Send发送端工程
3.1 config包
1）配置连接Rabbit连接
两种配置：以防漏掉些什么
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
connectionFactory.setPort(5672);
connectionFactory.setPassword("guest");
connectionFactory.setUsername("guest");
connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
connectionFactory.setPublisherReturns(true);
connectionFactory.createConnection();
return connectionFactory;
}
第二种:在之前application.properties里面配置连接地址和端口，然后RabbitListenerContainerFactory 去自动去连接配置里面地址和端口
@Bean
public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
return factory;
}
- 声明direct交换机，队列，路由键
@Bean
public Exchange directExchange(){
return new DirectExchange("dircet.exchange.test");
}
@Bean
public Queue directQueue(){
return new Queue("direct.queue.test");
}
@Bean
public Binding directBinding(){
return new Binding("direct.queue.test",
Binding.DestinationType.QUEUE,
"dircet.exchange.test",
"direct.key",null);
}
- 配置RabbitTemplate回调方法和发送方法确认方法
@Bean
@Qualifier("rabbitTemplate")
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//开启mandatory模式（开启失败回调）
rabbitTemplate.setMandatory(true);
//添加失败回调方法
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
log.info("message:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}",
message, replyCode, replyText, exchange, routingKey);
});
// 添加发送方确认模式方法
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) ->
log.info("correlationData:{}, ack:{}, cause:{}",
correlationData.getId(), ack, cause));
return rabbitTemplate;
}
4):整个config
@Component
@Slf4j
public class RabbitListenerConfig {
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
connectionFactory.setPort(5672);
connectionFactory.setPassword("guest");
connectionFactory.setUsername("guest");
connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
connectionFactory.setPublisherReturns(true);
connectionFactory.createConnection();
return connectionFactory;
}
@Bean
public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
return factory;
}
@Bean
@Qualifier("rabbitTemplate")
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//开启mandatory模式（开启失败回调）
rabbitTemplate.setMandatory(true);
//添加失败回调方法
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
log.info("message:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}",
message, replyCode, replyText, exchange, routingKey);
});
// 添加发送方确认模式方法
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) ->
log.info("correlationData:{}, ack:{}, cause:{}",
correlationData.getId(), ack, cause));
return rabbitTemplate;
}
/***声明 direct 队列 一对一***/
@Bean
public Exchange directExchange(){
return new DirectExchange("dircet.exchange.test");
}
@Bean
public Queue directQueue(){
return new Queue("direct.queue.test");
}
@Bean
public Binding directBinding(){
return new Binding("direct.queue.test",
Binding.DestinationType.QUEUE,
"dircet.exchange.test",
"direct.key",null);
}
}
3.2 dto包
@Getter
@Setter
@ToString
public class OrderMessageDTO {
private Integer orderId;
private BigDecimal price;
private Integer productId;
}
3.3 service包
public interface DirectService {
public void sendMessage();
}
@Slf4j
@Service
public class DirectServiceImpl implements DirectService {
@Autowired
private RabbitTemplate rabbitTemplate;
ObjectMapper objectMapper = new ObjectMapper();
@Override
public void sendMessage() {
log.info("==========发送Direct类型消息=======");
try {
String directStr = "Hello,我是directMesage";
// 第一种方式
OrderMessageDTO orderMessageDTO = new OrderMessageDTO();
orderMessageDTO.setOrderId(1);
orderMessageDTO.setPrice(new BigDecimal("20"));
orderMessageDTO.setProductId(100);
String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);
// 发送端确认是否确认消费
CorrelationData correlationData = new CorrelationData();
// 唯一ID
correlationData.setId(orderMessageDTO.getOrderId().toString());
// 发送
rabbitTemplate.convertAndSend("dircet.exchange.test","direct.queue.test",messageToSend,correlationData);
log.info("发送成功");
} catch (JsonProcessingException e) {
e.printStackTrace();
}
}
}
3.4 controller包
@RestController
@Slf4j
@RequestMapping("/api")
public class SendController {
@Autowired
private DirectService directService;
@GetMapping
public void sendOrder(){
for (int i = 0; i < 9000; i++) {
directService.sendMessage();
}
}
}
4. Receive 接受端工程
#######4.1config包
同上，选择一个连接RabbitMQ工厂
@Component
@Slf4j
public class RabbitListenerConfig {
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
connectionFactory.setPort(5672);
connectionFactory.setPassword("guest");
connectionFactory.setUsername("guest");
connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
connectionFactory.setPublisherReturns(true);
connectionFactory.createConnection();
return connectionFactory;
}
@Bean
public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
return factory;
}
}
#######4.2 service包
public interface DirectReciveService {
public void DirectRecive(Message message);
}
@Service
@Slf4j
public class DirectReciveServiceImpl implements DirectReciveService{
@RabbitListener(
containerFactory = "rabbitListenerContainerFactory",
bindings = {
@QueueBinding(
value = @Queue(name = "direct.queue.test"),
exchange = @Exchange(name = "dircet.exchange.test",
type = ExchangeTypes.DIRECT),
key = "direct.queue.test"
)
}
)
@Override
public void DirectRecive(@Payload Message message) {
log.info("========direct接受消息===========");
String messageBody = new String(message.getBody());
log.info(" body = {} " ,messageBody);
}
}
5. 启动发送端和接收端工程
1)访问：发送端地址http://localhost:8082/api
2)发送端：
correlationData：1 确认返回的标认ID，
ack：true 确认发送端的发出被消息返回的确认
cause:暂时未知
2021-05-01 01:43:43.958 INFO 9596 --- [nio-8082-exec-1] c.m.send.service.impl.DirectServiceImpl : ==========发送Direct类型消息=======
2021-05-01 01:43:43.959 INFO 9596 --- [nio-8082-exec-1] c.m.send.service.impl.DirectServiceImpl : 发送成功
2021-05-01 01:43:43.980 INFO 9596 --- [nectionFactory2] com.mi.send.config.RabbitListenerConfig : correlationData:1, ack:true, cause:null
2021-05-01 01:43:43.984 INFO 9596 --- [nectionFactory2] com.mi.send.config.RabbitListenerConfig : correlationData:1, ack:true, cause:null
3):接收端
消费发送端的Message
2021-05-01 01:43:43.984 INFO 13516 --- [ntContainer#0-1] c.m.r.s.Impl.DirectReciveServiceImpl : ========direct接受消息===========
2021-05-01 01:43:43.984 INFO 13516 --- [ntContainer#0-1] c.m.r.s.Impl.DirectReciveServiceImpl : body = {"orderId":1,"price":20,"productId":100}
6.结语
Springboot与RabbitMQ上手学习之Direct模式就到此为止
