当前位置 : 主页 > 编程语言 > 其它开发 >

手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(九) - Sp

来源:互联网 收集:自由互联 发布时间:2022-06-07
手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(一) - 介绍 手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(二) - 数据库设计 手把手教你使用

手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(一) - 介绍
手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(二) - 数据库设计
手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(三) - 项目初始化
手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(四) - 日志 & 跨域配置
手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(五) - MyBatis-Plus & 代码生成器集成与配置
手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(六) - 本地缓存 Caffeine 和 分布式缓存 Redis 集成与配置
手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(七) - Elasticsearch 8.2 集成与配置
手把手教你使用 Spring Boot 3 开发上线一个前后端分离的生产级系统(八) - XXL-JOB 集成与配置

Spring AMQP 介绍

AMQP(高级消息队列协议)是一个异步消息传递所使用的应用层协议规范,为面向消息的中间件设计,不受产品和开发语言的限制. Spring AMQP 将核心 Spring 概念应用于基于 AMQP 消息传递解决方案的开发。

RabbitMQ 是基于 AMQP 协议的轻量级、可靠、可扩展、可移植的消息中间件,Spring 使用 RabbitMQ 通过 AMQP 协议进行通信。Spring Boot 为通过 RabbitMQ 使用 AMQP 提供了多种便利,包括 spring-boot-starter-amqp “Starter”。

Spring AMQP 集成与配置
  1. 可通过如下 Docker 命令 安装 RabbiMQ:
 docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.10-management
  1. 登录 RabbiMQ 的 web 管理界面,创建虚拟主机novel:

RabbitMQ

  1. 项目中加入如下的 maven 依赖:
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
  1. 在 application.yml 配置文件中加入 RabbitMQ 的连接配置:
spring:
  rabbitmq:
    addresses: "amqp://guest:guest@47.106.243.172"
    virtual-host: novel
    template:
      retry:
        # 开启重试
        enabled: true
        # 最大重试次数
        max-attempts: 3
        # 第一次和第二次重试之间的持续时间
        initial-interval: "3s"
  1. 此时已经可以在 Spring Beans 中注入 AmqpTemplate 发送消息了。
Spring AMQP 使用示例
  1. io.github.xxyopen.novel.core.constant包下创建 AMQP 相关常量类:
/**
 * AMQP 相关常量
 *
 * @author xiongxiaoyang
 * @date 2022/5/25
 */
public class AmqpConsts {

    /**
     * 小说信息改变 MQ
     * */
    public static class BookChangeMq{

        /**
         * 小说信息改变交换机
         * */
        public static final String EXCHANGE_NAME = "EXCHANGE-BOOK-CHANGE";

        /**
         * Elasticsearch book 索引更新的队列
         * */
        public static final String QUEUE_ES_UPDATE = "QUEUE-ES-BOOK-UPDATE";

        /**
         * Redis book 缓存更新的队列
         * */
        public static final String QUEUE_REDIS_UPDATE = "QUEUE-REDIS-BOOK-UPDATE";

        // ... 其它的更新队列

    }

}
  1. io.github.xxyopen.novel.core.config包下创建 AMQP 配置类,配置各个交换机、队列以及绑定关系:
/**
 * AMQP 配置类
 *
 * @author xiongxiaoyang
 * @date 2022/5/25
 */
@Configuration
public class AmqpConfig {

    /**
     * 小说信息改变交换机
     */
    @Bean
    public FanoutExchange bookChangeExchange() {
        return new FanoutExchange(AmqpConsts.BookChangeMq.EXCHANGE_NAME);
    }

    /**
     * Elasticsearch book 索引更新队列
     */
    @Bean
    public Queue esBookUpdateQueue() {
        return new Queue(AmqpConsts.BookChangeMq.QUEUE_ES_UPDATE);
    }

    /**
     * Elasticsearch book 索引更新队列绑定到小说信息改变交换机
     */
    @Bean
    public Binding esBookUpdateQueueBinding() {
        return BindingBuilder.bind(esBookUpdateQueue()).to(bookChangeExchange());
    }

    // ... 其它的更新队列以及绑定关系

}
  1. io.github.xxyopen.novel.manager.mq包下创建 AMQP 消息管理类,用来发送各种 AMQP 消息:
/**
 * AMQP 消息管理类
 *
 * @author xiongxiaoyang
 * @date 2022/5/25
 */
@Component
@RequiredArgsConstructor
public class AmqpMsgManager {

    private final AmqpTemplate amqpTemplate;

    @Value("${spring.amqp.enable}")
    private String enableAmqp;

    /**
     * 发送小说信息改变消息
     */
    public void sendBookChangeMsg(Long bookId) {
        if (Objects.equals(enableAmqp, CommonConsts.TRUE)) {
            sendAmqpMessage(amqpTemplate, AmqpConsts.BookChangeMq.EXCHANGE_NAME, null, bookId);
        }
    }

    private void sendAmqpMessage(AmqpTemplate amqpTemplate, String exchange, String routingKey, Object message) {
        // 如果在事务中则在事务执行完成后再发送,否则可以直接发送
        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    amqpTemplate.convertAndSend(exchange, routingKey, message);
                }
            });
            return;
        }
        amqpTemplate.convertAndSend(exchange, routingKey, message);
    }

}
  1. 在小说信息更新后,发送 AMQP 消息:
@Transactional(rollbackFor = Exception.class)
@Override
public RestResp<Void> saveBookChapter(ChapterAddReqDto dto) {
    // 1) 保存章节相关信息到小说章节表
    //  a) 查询最新章节号
   
    //  b) 设置章节相关信息并保存   

    // 2) 保存章节内容到小说内容表

    // 3) 更新小说表最新章节信息和小说总字数信息
    //  a) 更新小说表关于最新章节的信息
    
    //  b) 发送小说信息更新的 MQ 消息
    amqpMsgManager.sendBookChangeMsg(dto.getBookId());
    return RestResp.ok();
}
  1. io.github.xxyopen.novel.core.listener包下创建 Rabbit 队列监听器,监听各个 RabbitMQ 队列的消息并处理:
/**
 * Rabbit 队列监听器
 *
 * @author xiongxiaoyang
 * @date 2022/5/25
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class RabbitQueueListener {

    private final BookInfoMapper bookInfoMapper;

    private final ElasticsearchClient esClient;

    /**
     * 监听小说信息改变的 ES 更新队列,更新最新小说信息到 ES
     * */
    @RabbitListener(queues = AmqpConsts.BookChangeMq.QUEUE_ES_UPDATE)
    @SneakyThrows
    public void updateEsBook(Long bookId) {
        BookInfo bookInfo = bookInfoMapper.selectById(bookId);
        IndexResponse response = esClient.index(i -> i
                .index(EsConsts.BookIndex.INDEX_NAME)
                .id(bookInfo.getId().toString())
                .document(EsBookDto.build(bookInfo))
        );
        log.info("Indexed with version " + response.version());
    }

    // ... 监听其它队列,刷新其它副本数据

}

此时,如果需要更新其它小说副本数据,只需要配置更新队列和增加监听器,不需要在小说信息变更的地方增加任何业务代码,而且任意小说副本的数据刷新之间互不影响,真正实现了模块间的解耦。

网友评论