rabbitmq的使用介绍

news2025/5/29 8:35:11

一.队列工作模式介绍

1.WorkQueues模型

生产者直接把消息发送给队列，然后消费者订阅队列

特点: 消息不会重复, 分配给不同的消费者.

代码实现：

消费者代码：

@Component
@Slf4j
public class SpringRabbitListener {
    @RabbitListener(queues = "work.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }
}

生产者代码：

@Test
public void testWorkQueue() throws InterruptedException {
    // 队列名称
    String queueName = "work.queue";
    // 消息
    String message = "hello, message_";
    for (int i = 0; i < 50; i++) {
        // 发送消息，每20毫秒发送一次，相当于每秒发送50条消息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}

2.含有交换机的模型

Publisher：生产者，不再发送消息到队列中，而是发给交换机
Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
Consumer：消费者，与以前一样，订阅队列，没有变化

交换机的四种类型：

①Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
②Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
③Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
④Headers：头匹配，基于MQ的消息头匹配（用的较少，这里不讲）

RoutingKey: 路由键.⽣产者将消息发给交换器时, 指定的⼀个字符串, ⽤来告诉交换机应该如何处理这个消息.

Binding Key:绑定. RabbitMQ中通过Binding(绑定)将交换器与队列关联起来, 在绑定的时候⼀般会指定⼀个Binding Key, 这样RabbitMQ就知道如何正确地将消息路由到队列了.

①fanout交换机

交换机和队列的关系图：

1）可以有多个队列
2）每个队列都要绑定到Exchange（交换机）
3）生产者发送的消息，只能发送到交换机
4）交换机把消息发送给绑定过的所有队列
5）订阅队列的消费者都能拿到消息

代码实现：

消费者代码：

@Component
@Slf4j
public class SpringRabbitListener {
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) {
        System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) {
        System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
    }
}

生产者代码：

@Test
public void testFanoutExchange() {
    // 交换机名称
    String exchangeName = "hmall.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

②direct交换机

交换和队列的关系图：交换机和队列之间多了一个绑定关系

1）队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）。
2）消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
3）Exchange不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key进行判断，只有队列的 Routingkey 与消息的 Routing key 完全一致，才会接收到消息。

代码实现：

消费者代码：

@Component
@Slf4j
public class SpringRabbitListener {
    @RabbitListener(queues = "direct.queue1")
    public void listenDirectQueue1(String msg) {
        System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
    }

    @RabbitListener(queues = "direct.queue2")
    public void listenDirectQueue2(String msg) {
        System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
    }
}

生产者代码：

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "hmall.direct";
    // 消息
    String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "yellow", message);
}

③topic交换机

交换机和队列的关系图：交换机和队列之间的绑定的字符串含有通配符

通配符规则：

# ：匹配一个或多个词
* ：匹配不多不少恰好1个词

假如此时publisher发送的消息使用的`RoutingKey`共有四种：

china.news：可以路由到队列1和2
china.weather：可以路由到队列1
japan.news：可以路由到队列2

代码实现：

消费者代码：

@Component
@Slf4j
public class SpringRabbitListener {
    @RabbitListener(queues = "topic.queue1")
    public void listenTopicQueue1(String msg){
        System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
    }

    @RabbitListener(queues = "topic.queue2")
    public void listenTopicQueue2(String msg){
        System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
    }
}

生产者代码：

@Test
public void testSendTopicExchange() {
    // 交换机名称
    String exchangeName = "hmall.topic";
    // 消息
    String message = "喜报！孙悟空大战哥斯拉，胜!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

二.使用java代码声明交换机，队列，绑定交换机队列

1.基于Bean声明：写成一个配置类

@Configuration
public class FanoutConfiguration {
    /**
     * 声明交换机
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        //return ExchangeBuilder.fanoutExchange("hmall.fanout").build();
        return new FanoutExchange("hmall.fanout");
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue fanoutQueue1(){
        //return QueueBuilder.durable("fanout.queue1").build();
        return new Queue("fanout.queue1");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue fanoutQueue2(){
        //return QueueBuilder.durable("fanout.queue2").build();
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

2.使用注解声明，把声明写在消费者的注解上

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue1"),
        exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
    System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue2"),
        exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "yellow"}
))

三.消息转换器

上面 convertAndSend 方法的第三个参数是一个 Object 对象，在数据传输时，spring 会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。
只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

① 数据体积过大
② 有安全漏洞
③ 可读性差

所以我们需要配置json转换器

①引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

②添加配置

@Bean
public MessageConverter messageConverter(){
    // 1.定义消息转换器
    Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
    // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
    jackson2JsonMessageConverter.setCreateMessageIds(true);  // 这个代码是从幂等性添加进来的
    return jackson2JsonMessageConverter;
}