目录
一、简介
二、代码展示
父pom文件
pom文件
配置文件
config
生产者
消费者
测试
结果
一、简介
扇型(广播)交换机,这个交换机没有路由键概念,就算你绑了路由键也是无视的。 这个交换机在接收到消息后,会直接转发到绑定到它上面的所有队列。
广播交换机是不需要RoutingKey的;
二、代码展示
父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.7.1</version>
<!-- <version>2.2.5.RELEASE</version>-->
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.chensir</groupId>
<artifactId>spring-boot-rabbitmq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>spring-boot-rabbitmq</name>
<properties>
<java.version>8</java.version>
<hutool.version>5.8.3</hutool.version>
<lombok.version>1.18.24</lombok.version>
</properties>
<description>spring-boot-rabbitmq</description>
<packaging>pom</packaging>
<modules>
<module>direct-exchange</module>
<module>fanout-exchange</module>
<module>topic-exchange</module>
<module>game-exchange</module>
<module>dead-letter-queue</module>
<module>delay-queue</module>
<module>delay-queue2</module>
</modules>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>${hutool.version}</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>${lombok.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
</project>
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>com.chensir</groupId>
<artifactId>spring-boot-rabbitmq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>fanout-exchange</artifactId>
<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>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</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>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
</project>
配置文件
server.port=8082
#host
spring.rabbitmq.host=121.40.100.52
#默认5672
spring.rabbitmq.port=5672
#用户名
spring.rabbitmq.username=guest
#密码
spring.rabbitmq.password=guest
#连接到代理时用的虚拟主机
spring.rabbitmq.virtual-host=/
#每个消费者每次可最大处理的nack消息数量 默认250个
spring.rabbitmq.listener.simple.prefetch=1
#表示消息确认方式,其有三种配置方式,分别是none、manual(手动)和auto(自动);默认auto
spring.rabbitmq.listener.simple.acknowledge-mode=auto
#监听重试是否可用
spring.rabbitmq.listener.simple.retry.enabled=true
#最大重试次数
#spring.rabbitmq.listener.simple.retry.max-attempts=5
#最大重试时间间隔
spring.rabbitmq.listener.simple.retry.max-interval=20000ms
#第一次和第二次尝试传递消息的时间间隔
spring.rabbitmq.listener.simple.retry.initial-interval=3000ms
#应用于上一重试间隔的乘数
spring.rabbitmq.listener.simple.retry.multiplier=2
#决定被拒绝的消息是否重新入队;默认是true(与参数acknowledge-mode有关系)
spring.rabbitmq.listener.simple.default-requeue-rejected=false
config
package com.chensir.config;
import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.security.Key;
@Configuration
public class RabbitConfig {
//解决对象类型乱码
@Bean
public Jackson2JsonMessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
/**
* 广播交换机(扇形交换机)
* @return
*/
@Bean
public FanoutExchange fanoutExchange(){
// DirectExchange directExchange=new DirectExchange("DirectExchange-01",true,false);
FanoutExchange fanoutExchange = new FanoutExchange("FanoutExchange-01", true, false);
return fanoutExchange;
}
/**
* 队列a
* @return
*/
@Bean
public Queue fanoutQueueA(){
Queue queue = QueueBuilder.durable("FanoutQueue-01").build();
return queue;
}
/**
* 队列b
* @return
*/
@Bean
public Queue fanoutQueueB(){
// Queue queue=QueueBuilder.durable("DirectQueue-01").build();
Queue queue = QueueBuilder.durable("FanoutQueue-02").build();
return queue;
}
/**
* 绑定
* @return
*/
@Bean
public Binding bindingA() {
// Binding binding=BindingBuilder.bind(directQueue()).to(directExchange).with("key-01");
Binding binding = BindingBuilder.bind(fanoutQueueA()).to(fanoutExchange());
return binding;
}
@Bean
public Binding bindingB() {
// Binding binding=BindingBuilder.bind(directQueue()).to(directExchange).with("key-01");
Binding binding = BindingBuilder.bind(fanoutQueueB()).to(fanoutExchange());
return binding;
}
}
生产者
@Component
public class FanoutProvider {
@Resource
private RabbitTemplate rabbitTemplate;
/**
* 无需绑定routingkey
*/
public void send(){
for (int i = 0; i < 10; i++) {
User user=new User();
user.setId(i);
user.setName("陈sir");
user.setSex("男");
// rabbitTemplate.convertAndSend("DirectExchange-01","key-01",user);
rabbitTemplate.convertAndSend("FanoutExchange-01",null,user);
}
}
}
消费者
@Component
public class FanoutConsumer {
@RabbitHandler
@RabbitListener(queues = "FanoutQueue-01")
public void processA(User user) {
System.out.println("A 接受到消息,并正常处理结束"+ JSONUtil.toJsonStr(user));
}
@RabbitHandler
@RabbitListener(queues = "FanoutQueue-02")
public void processB(User user) {
System.out.println("B 接受到消息,并正常处理结束"+ JSONUtil.toJsonStr(user));
}
}
测试
结果