RabbitMQ项目实战(一）

news2024/11/23 3:26:00

文章目录

RabbitMQ项目实战
- 选择客户端
- 基础实战

前情提要：我们了解了消息队列，RabbitMQ的入门，交换机，以及核心特性等知识，现在终于来到了激动人心的项目实战环节！本小节主要介绍通过Spring Boot RabbitMQ Starter 在SpringBoot项目中跑通测试RabbitMQ，话不多说，我们马上开始！

RabbitMQ项目实战

选择客户端

怎么在项目中使用RabbitMQ?

使用官方的客户端（类比jdbc）

优点：兼容性好，换语言成本低，比较灵活
缺点：太灵活，要自己去处理一些事情，比如要自己维护管理链接，很麻烦

使用封装好的客户端，比如Spring Boot RabbitMQ Starter（类比mybatis）

优点：简单易用，直接配置直接用，更方便地去管理链接
缺点：不够灵活，被框架限制

基础实战

我们使用Spring Boot RabbitMQ Starter
https://spring.io/guides/gs/messaging-rabbitmq/

引入依赖
注意：使用的版本一定要和你的springboot版本一致，去maven中心仓库中找版本一致的

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>2.7.2</version>
</dependency>

2）在yml中引入配置

spring:
    rabbitmq:
        host: localhost
        port: 5672
        password: guest
        username: guest

3）创建交换机和队列，一般在项目启动之前执行创建一次即可

package com.yupi.springbootinit.bimq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * 用于创建测试程序用到的交换机和队列（只用在程序启动前执行一次）
 */
public class MqInitMain {
    public static void main(String[] args) {
        try{
            ConnectionFactory factory  = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            String EXCHANGE_NAME = "code_exchange";
            channel.exchangeDeclare(EXCHANGE_NAME,"direct");
            String queueName = "code_queue";
            channel.queueDeclare(queueName,true,false,false,null);
            channel.queueBind(queueName,EXCHANGE_NAME,"my_routingKey");
        }catch (Exception e){

        }
    }
}

4）生产者代码

package com.yupi.springbootinit.bimq;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class MyMessageProducer {
    @Resource
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String exchange,String routingKey,String message){
        rabbitTemplate.convertAndSend(exchange,routingKey,message);
    }
}

5）消费者代码

package com.yupi.springbootinit.bimq;

import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MyMessageConsumer {


    //使用@SneakyThrows注解简化异常处理
    //使得你可以在不声明抛出异常的方法中抛出受检异常，而无需捕获它们。这在一些特定情况下可能会很有用，但通常不建议频繁使用，因为它可能会破坏代码的可读性和健壮性。
    @SneakyThrows
    //使用@RabbitListener注解指定要监听的队列名称为"code_queue"，并设置消息的确认机制为手动确认
    @RabbitListener(queues = {"code_queue"},ackMode = "MUNAL")
    // // 在RabbitMQ中,每条消息都会被分配一个唯一的投递标签，用于标识该消息在通道中的投递状态和顺序。通过使用@Header(AmqpHeaders.DELIVERY_TAG)注解,可以从消息头中提取出该投递标签,并将其赋值给long deliveryTag参数。
    public void reciveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliverttag){
        log.info("receiveMessage message = {}", message);
        //手动确认消息的接收
        channel.basicAck(deliverttag,false);
    }
}

测试类测试

package com.yupi.springbootinit.bimq;

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;

import static org.junit.jupiter.api.Assertions.*;
@SpringBootTest
class MyMessageProducerTest {

    @Resource
    private MyMessageProducer myMessageProducer;
    @Test
    void sendMessage() {
        myMessageProducer.sendMessage("code_exchange", "my_routingKey", "你好呀");
    }
}