RabbitMQ的常用交换机在springboot中的使用

一、导入依赖


<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-amqpartifactId>
dependency>

二、编写配置


下面配置都是spring:下的
 
rabbitmq:
  host: 44.98.18
  username: guest
  password: guest
  # 虚拟主机
  virtual-host: /
  port: 5672
  listener:
    simple:
      # 消费者最小数量
      concurrency: 10
      # 消费者最大数量
      max-concurrency: 10
      # 限制消费者每次只能处理一条消息，处理完再继续处理下一条消息
      prefetch: 1
      # 启动时是否默认启动容器，默认true
      auto-startup: true
      # 拒绝策略：被拒绝时是否重新进入队列
      default-requeue-rejected: true
  template:
    retry:
      # 发布重试，默认false
      enabled: true
      # 重试时间默认1000ms
      initial-interval: 1000ms
      # 重试次数,默认3
      max-attempts: 3
      # 重试最大间隔时间，默认10000ms
      max-interval: 10000ms
      # 重试的间隔乘数，比如配置2.0，第一次等10s，第二次等20s，第三次等40秒
      multiplier: 1

一个简单队列

需要创建一个配置类，然后创建一个生产者，再创建一个消费者，之后在具体的业务中调用生产者就欧克了

创建一个队列 RabbitMQConfig.java


/**
 * 
 * RabbitMQ配置类
 * 
 *
 * @author wangmh
 * @since 2022/9/20 17:01
 */
@Configuration
public class RabbitMQConfig {
 
    @Bean
    public Queue queue(){
        // 名字，是否持久化
        return new Queue("queue",true);
    }
 
}

生产者：MQSender.java


@Service
@Slf4j
public class MQSender {
 
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    public void send(Object msg){
        log.info("发送消息："+msg);
        rabbitTemplate.convertAndSend("queue",msg);
    }
 
}

消费者：MQReceiver.java


@Service
@Slf4jd
public class MQReceiver {
 
    // 监听这个队列
    @RabbitListener(queues = "queue")
    public void receive(Object msg){
        log.info("接收消息"+msg);
    }
 
}

输出结果


结果：
2022-09-20 19:43:49.720  INFO 11180 --- [nio-8080-exec-1] com.example.rabbitmq.MQSender            : 发送消息：hello
2022-09-20 19:43:49.853  INFO 11180 --- [ntContainer#0-9] com.example.rabbitmq.MQReceiver          : 接收消息(Body:'hello' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=, receivedRoutingKey=queue, deliveryTag=1, consumerTag=amq.ctag-rr1oDI2wtJb1Np0jucHeGw, consumerQueue=queue])

交换机

一、fanout

广播模式，把队列和交换机进行绑定，转发消息是最快的，因为他不需要去处理路由键

发送到交换机的消息，会被转发到和这个交换机绑定的所有队列上（发布订阅模式

在配置文件中配置2个队列，1个交换机，把这个交换机和2个队列做绑定


@Configuration
public class RabbitMQConfig {
 
    private static final String QUEUE01 = "queue_fanout01";
    private static final String QUEUE02 = "queue_fanout02";
    private static final String EXCHANGE = "fanoutExchange";
 
    @Bean
    public Queue queue01(){
        return new Queue(QUEUE01);
    }
 
    @Bean
    public Queue queue02(){
        return new Queue(QUEUE02);
    }
 
 
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange(EXCHANGE);
    }
 
    // 绑定交换机和队列
    @Bean
    public Binding binding01(){
        return BindingBuilder.bind(queue01()).to(fanoutExchange());
    }
 
    @Bean
    public Binding binding02(){
        return BindingBuilder.bind(queue02()).to(fanoutExchange());
    }
 
}

发送：


@Autowired
private RabbitTemplate rabbitTemplate;
 
public void send01(Object msg){
    log.info("发送消息："+msg);
    rabbitTemplate.convertAndSend("fanoutExchange","",msg);
}

接收 :


@RabbitListener(queues = "queue_fanout01")
public void receive01(Object msg){
    log.info("queue01："+msg);
}
 
@RabbitListener(queues = "queue_fanout02")
public void receive02(Object msg){
    log.info("queue02 :"+msg);
}

运行后就可以看到控制台的变化：

结果：

二、direct

所有被发送到direct交换机上的消息，都会被转发到路由key中指定的一个queue

和fanout相比，就是多了一个key，匹配的时候根据key进行匹配

发送的时候要带有路由key,如果匹配不到默认上会丢失，但是我们在yaml中配置了，发送失败重新进入队列

但是：

当项目使用的时间越来越就，路由key就会越来越多，就会变得越来越难管

RabbitMQConfig.java


@Configuration
public class RabbitMQConfig {
 
    private static final String QUEUE01 = "queue_direct01";
    private static final String QUEUE02 = "queue_direct02";
    private static final String EXCHANGE = "directExchange";
    private static final String ROUTINGKEY01 = "queue.red";
    private static final String ROUTINGKEY02 = "queue.green";
 
    @Bean
    public Queue queue01(){
        return new Queue(QUEUE01);
    }
 
    @Bean
    public Queue queue02(){
        return new Queue(QUEUE02);
    }
 
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange(EXCHANGE);
    }
 
    // 绑定队列和交换机，这种模式需要指定路由key
    @Bean
    public Binding binding01(){
        return BindingBuilder.bind(queue01()).to(directExchange()).with(ROUTINGKEY01);
    }
 
    @Bean
    public Binding binding02(){
        return BindingBuilder.bind(queue02()).to(directExchange()).with(ROUTINGKEY02);
    }
 
}

发送者：


@Service
@Slf4j
public class MQSender {
 
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    public void send02(Object msg){
        log.info("发送红色的消息：："+msg);
        rabbitTemplate.convertAndSend("directExchange","queue.red",msg);
    }
 
    public void send03(Object msg){
        log.info("发送绿色的消息：："+msg);
        rabbitTemplate.convertAndSend("directExchange","queue.green",msg);
    }
 
}

接收者：


@Service
@Slf4j
public class MQReceiver {
 
    // 监听这个队列
    @RabbitListener(queues = "queue_direct01")
    public void receive02(Object msg){
        log.info("queue_direct01："+msg);
    }
 
    @RabbitListener(queues = "queue_direct02")
    public void receive03(Object msg){
        log.info("queue_direct02 :"+msg);
    }
 
}

测试


@RequestMapping("/direct")
@ResponseBody
public void mq01(){
    mqSender.send02("hello+red");
    mqSender.send03("hello+green");
}

运行结果：

看到管控台可以看到，对应的交换机，队列，和队列对应的key

三、topic

*只能替代1个词

#可以替代0个或多个词、

可以说是 direct模式的延申，方便我们管理路由key

所有发送到主机交换机的消息都会被转发到所有的路由ley中，所指定的topic的queue上去，交换机会将路由key和topic模糊匹配，此时，队列需要绑定一个topic

RabbitMQConfig.java


@Configuration
public class RabbitMQConfig {
 
    private static final String QUEUE01 = "queue_topic01";
    private static final String QUEUE02 = "queue_topic02";
    private static final String EXCHANGE = "topicExchange";
    private static final String ROUTINGKEY01 = "#.queue.#";
    private static final String ROUTINGKEY02 = "*.queue.#";
 
    @Bean
    public Queue queue01(){
        return new Queue(QUEUE01);
    }
 
    @Bean
    public Queue queue02(){
        return new Queue(QUEUE02);
    }
 
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(EXCHANGE);
    }
 
    // 绑定队列和交换机，这种模式需要指定路由key
    @Bean
    public Binding binding01(){
        return BindingBuilder.bind(queue01()).to(topicExchange()).with(ROUTINGKEY01);
    }
 
    @Bean
    public Binding binding02(){
        return BindingBuilder.bind(queue02()).to(topicExchange()).with(ROUTINGKEY02);
    }
 
}

发送消息：


@Service
@Slf4j
public class MQSender {
 
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    public void send01(Object msg){
        log.info("发送消息Queue1：："+msg);
        rabbitTemplate.convertAndSend("topicExchange","queue.red.message",msg);
    }
 
     public void send02(Object msg){
        log.info("发送消息，被queue1+queue2接收：："+msg);
        rabbitTemplate.convertAndSend("topicExchange","message.queue.green",msg);
    }
 
}

接收消息


@Service
@Slf4j
public class MQReceiver {
 
    // 监听这个队列
    @RabbitListener(queues = "queue_topic01")
    public void receive02(Object msg){
        log.info("queue_topic01："+msg);
    }
 
    @RabbitListener(queues = "queue_topic02")
    public void receive03(Object msg){
        log.info("queue_topic02 :"+msg);
    }
 
}

测试


@RequestMapping("/topic")
@ResponseBody
public void mq01(){
    mqSender.send02("hello+red");
    mqSender.send02("hello+green");
}

结果

和刚刚一样，可以去管控太去查看响应的交换机和队列的信息

三、Headers模式

并不依赖于路由key

whereAll() 多个键值对都要满足

whereAny() 任意一个键值对满足即

RabbitMQConfig.java


@Configuration
public class RabbitMQConfig {
 
    private static final String QUEUE01 = "queue_headers01";
    private static final String QUEUE02 = "queue_headers02";
    private static final String EXCHANGE = "headerExchange";
 
    @Bean
    public Queue queue01(){
        return new Queue(QUEUE01);
    }
 
    @Bean
    public Queue queue02(){
        return new Queue(QUEUE02);
    }
 
    @Bean
    public HeadersExchange headersExchange(){
        return new HeadersExchange(EXCHANGE);
    }
 
    // 绑定队列和交换机，这种模式需要指定路由key
    @Bean
    public Binding binding01(){
        Map map = new HashMap<>();
        map.put("color","red");
        map.put("speed","low");
        // 两个只需要匹配一个
        return BindingBuilder.bind(queue01()).to(headersExchange()).whereAny(map).match();
    }
 
    @Bean
    public Binding binding02(){
        Map map = new HashMap<>();
        map.put("color","red");
        map.put("speed","fast");
        // 必须要两个同时匹配上才可以
        return BindingBuilder.bind(queue02()).to(headersExchange()).whereAll(map).match();
    }
 
}

发送消息：


public void send01(String msg){
    log.info("发送两个都能接收的：："+msg);
    MessageProperties properties = new MessageProperties();
    properties.setHeader("color","red");
    properties.setHeader("speed","fast");
    Message message = new Message(msg.getBytes(),properties);
    rabbitTemplate.convertAndSend("headerExchange","",message);
}
 
public void send02(String msg){
    log.info("发送消息，queue1接收：："+msg);
    MessageProperties properties = new MessageProperties();
    properties.setHeader("color","red");
    properties.setHeader("speed","normal");
    Message message = new Message(msg.getBytes(),properties);
    rabbitTemplate.convertAndSend("headerExchange","",message);
}

接收消息：


@RabbitListener(queues = "queue_headers01")
public void receive01(Message message){
    log.info("queue01接收Message对象："+message);
    log.info("queue01接收消息："+ new String(message.getBody()));
}
 
@RabbitListener(queues = "queue_headers02")
public void receive02(Message message){
    log.info("queue02接收Message对象："+message);
    log.info("queue02接收消息："+ new String(message.getBody()));
}

测试：


@RequestMapping("/direct")
@ResponseBody
public void mq01(){
    mqSender.send01("王萌虎");
    mqSender.send02("wangmenghu");
}

结果：

相关阅读:
ArcGIS：如何利用站点数据(例如臭氧)进行克里金插值得到连续臭氧表面？
Zabbix监控组件及流程
 Django(二)精美博客搭建(13)实现留言页面及留言功能
 论文精读GAN: Generative Adversarial Nets
QML元素定位器：Row、Colum、Grid、Flow、定位器嵌套以及Repeater用法
 436. 寻找右区间--LeetCode_二分
 量化交易之日内回转策略：如何利用MACD指标实现盈利？
【机器学习】Kmeans聚类算法
 全域营销怎么做？统一客户数据是起点
 神奇的MappedByteBuffer
原文地址：https://blog.csdn.net/w13966597931/article/details/127105242