仿今日头条项目

自媒体文章上下架

自媒体文章上下架的实现：

流程实现

消息通知的其他需求：

消息通知的其他需求

Kafka 概述

消息中间件的对比：

特性 ActiveMQ RabbitMQ RocketMQ Kafka
开发语言 java erlang java scala
单机吞吐量 万级 万级 10万级 100万级
时效性 ms us ms ms级以内
可用性 高（主从） 高（主从） 非常高（分布式） 非常高（分布式）
功能特性 成熟的产品、较全的文档、各种协议支持好 并发能力强、性能好、延迟低 MQ功能比较完善，扩展性佳 只支持主要的MQ功能，主要应用于大数据领域

消息中间件对比 – 选择建议：

消息中间件 建议
Kafka 追求高吞吐量，适合产生大量数据的互联网服务的数据收集业务
RocketMQ 可靠性要求很高的金融互联网领域，稳定性高，经历了多次阿里双11考验
RabbitMQ 性能较好，社区活跃度高，数据量没有那么大，优先选择功能比较完备的RabbitMQ

Kafka 介绍：Kafka 是一个分布式流媒体平台，类似于 消息队列 或 企业消息传递系统。

kafka 官网：http://kafka.apache.org/

Kafka

Kafka 名称解释：

Kafka结构图

  • producer：发布消息的对象 称之为 主题生产者（Kafka topic producer）

  • topic：Kafka将消息分门别类，每一类的消息称之为一个主题（Topic）

  • consumer：订阅消息并处理发布的消息的对象 称之为 主题消费者（consumers）

  • broker：已发布的消息保存在一组服务器中，称之为 Kafka集群。集群中的每一个 服务器 都是一个 代理 （Broker），消费者可以订阅一个或多个 主题（topic），并从Broker拉数据，从而消费这些已发布的消息。

Kafka 安装配置

Kafka 对于 zookeeper 是强依赖，保存 kafka 相关的节点数据，所以安装Kafka之前必须先安装 zookeeper

  1. Docker 安装 zookeeper

    1
    docker pull zookeeper

  2. 创建容器

    1
    docker run -d --name zookeeper -p 2181:2181 zookeeper

  3. Docker 安装 kafka

    1
    docker pull wurstmeister/kafka

  4. 创建容器

    1
    2
    3
    4
    5
    6
    7
    docker run -d --name kafka \
    --env KAFKA_ADVERTISED_HOST_NAME=192.168.88.129 \
    --env KAFKA_ZOOKEEPER_CONNECT=192.168.88.129:2181 \
    --env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.88.129:9092 \
    --env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
    --env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
    --net=host wurstmeister/kafka
    • --net=host，直接使用容器宿主机的网络命名空间， 即没有独立的网络环境。它使用宿主机的ip和端口

kafka 入门案例

案例流程

  • 生产者 发送消息，多个消费者 只能有 一个消费者 接收到消息
  • 生产者 发送消息，多个消费者 都可以 接收到消息

流程：

  1. 创建 kafka-demo 项目，导入依赖

    1
    2
    3
    4
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
    </dependency>

  2. 编写 消息生产者类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    /**
     * 生产者
     */
    public class ProducerQuickStart {

        public static void main(String[] args) {

            // 1. kafka链接配置信息
            Properties prop = new Properties();

            // 1.1 设置kafka的链接地址
            prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.88.129:9092");

            // 1.2 设置key和value的序列化
            prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

            // 2. 创建kafka生产者对象
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);

            // 3. 发送消息
            /**
             * 第一个参数：topic
             * 第二个参数：消息的key
             * 第三个参数：消息的value
             */
            ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<String, String>("topic-first", "key-001","hello kafka");
            producer.send(kvProducerRecord);

            // 4. 关闭消息通道
            // 必须要关闭，否则发送不成功
            producer.close();
        }
    }

  3. 编写 消息消费者类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    /**
     * 消费者
     */
    public class ConsumerQuickStart {
        public static void main(String[] args) {

            // 1. kafka的配置信息
            Properties prop = new Properties();

            // 1.1 链接地址
            prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.88.129:9092");

            // 1.2 key和value的反序列化器
            prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

            // 1.3 设置消费者组
            prop.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");

            // 2. 创建消费者对象
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(prop);

            // 3. 订阅主题（topic）
            consumer.subscribe(Collections.singletonList("topic-first"));

            // 4. 拉去消息

            while (true) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    System.out.println(consumerRecord.key());
                    System.out.println(consumerRecord.value());
                }
            }
        }
    }

总结：

  • 生产者发送消息，多个消费者订阅同一个主题，只能有 一个消费者 收到消息（一对一）。是在同一个group 下

    同一个group下

  • 生产者发送消息，多个消费者订阅同一个主题，所有消费者 都能收到消息（一对多）。是在不同的 group 下

    不同的group下

Kafka 分区机制

Kafka 中的 分区机制 指的是：将每个 主题 划分成 多个分区（Partition）

可以处理更多的消息，不受单台服务器的限制，可以不受限的处理更多的数据

kafka分区机制

topic剖析

  • 每一个 分区 都是一个 顺序的、不可变 的 消息队列， 并且可以 持续的添加
  • 分区中的消息都被分了一个 序列号，称之为 偏移量(offset)，在每个分区中此 偏移量 都是 唯一的

kafka分区策略

Kafka 高可用设计

集群

Kafka集群

  • Kafka 的服务器端由被称为 Broker 的 服务进程 构成，即一个 Kafka 集群由多个 Broker 组成
  • 如果 集群 中某一台 机器宕机，其他机器上的 Broker 也依然能够对外提供服务。这其实就是 Kafka 提供高可用的手段之一

备份机制 (Replication）

Kafka备份机制

Kafka 中消息的备份又叫做 副本（Replica）

Kafka 定义了两类副本：

  • 领导者副本（Leader Replica）
  • 追随者副本（Follower Replica）

同步方式：

Kafka备份机制-同步方式

  • ISR（in-sync replica）：需要 同步复制保存 的 follower
  • 如果 leader 失效后，需要选出新的 leader
    • 选举的原则如下：
      • 第一：选举时优先从 ISR 中选定，因为这个列表中 follower 的数据是与 leader 同步的
      • 第二：如果 ISR 列表中的 follower 都不行了，就只能从其他 follower 中选取
  • 极端情况，就是 所有副本 都 失效 了
    • 这时有两种方案：
      • 第一：等待 ISR 中的一个活过来，选为 Leader，数据可靠，但活过来的时间不确定
      • 第二：选择第一个活过来的 副本（Replication），不一定是 ISR 中的，选为 leader，以最快速度恢复可用性，但数据不一定完整

Kafka 生产者详解

发送类型

同步发送

使用 send() 方法发送，它会返回一个 Future 对象，调用 get() 方法进行等待，就可以知道消息是否发送成功

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
 * 第一个参数：topic
 * 第二个参数：消息的key
 * 第三个参数：消息的value
 */
ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<String, String>("topic-first", "key-001","hello kafka");

// 同步发送消息
RecordMetadata recordMetadata = null;
try {
    recordMetadata = producer.send(kvProducerRecord).get();
} catch (Exception e) {
    e.printStackTrace();
} 
System.out.println(recordMetadata.offset()); // 打印偏移量

异步发送

调用 send() 方法，并指定一个回调函数，服务器在返回响应时调用函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/**
 * 第一个参数：topic
 * 第二个参数：消息的key
 * 第三个参数：消息的value
 */
ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<String, String>("topic-first", "key-001","hello kafka");

// 异步消息发送
producer.send(kvProducerRecord, new Callback() {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e != null) {
            // 这里可以异步记录异常信息到日志表中
            System.out.println("记录异常信息到日志表中");
        }
        System.out.println(recordMetadata.offset());
    }
}

参数说明

ack

ack配置

参数表格：

确认机制 说明
acks=0 生产者在成功写入消息之前 不会等待任何来自服务器 的响应，消息有丢失的风险，但是速度最快
acks=1（默认值） 只要集群首领节点收到消息，生产者就会收到一个来自服务器的成功响应
acks=all 只有当所有参与赋值的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应

eg：

1
2
3
// ack配置 消息确认机制
// 只有当所有的副本同步之后，消息才会发送成功
prop.put(ProducerConfig.ACKS_CONFIG, "all");

retries

retries配置

生产者从服务器收到的错误有可能是 临时性错误，在这种情况下，retries 参数的值决定了生产者可以重发消息的次数，如果达到这个次数，生产者会放弃重试返回错误，默认情况下，生产者会在每次重试之间等待100ms

eg：

1
2
3
// 重试次数
// 当生产者发送消息，首领没有确认之后，会重试进行发送，一直发送10次
prop.put(ProducerConfig.RETRIES_CONFIG, 10);

消息压缩

默认情况下， 消息发送时 不会被压缩

压缩算法 说明
snappy 占用较少的 CPU， 却能提供较好的性能和相当可观的压缩比， 如果看重性能和网络带宽，建议采用
lz4 占用较少的 CPU， 压缩和解压缩速度较快，压缩比也很客观
gzip 占用较多的 CPU，但会提供更高的压缩比，网络带宽有限，可以使用这种算法

使用 压缩 可以 降低网络传输开销 和 存储开销，而这往往是向 Kafka 发送消息的瓶颈所在

eg：

1
2
// 数据压缩
prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

Kafka 消费者详解

消费者组

消费者组

  • 消费者组（Consumer Group） ：指的就是由 一个 或 多个 消费者 组成的 群体

  • 一个发布在 topic 上消息被分发给此 消费者组 中的 一个消费者

    • 所有的 消费者 都在 一个组（group） 中，那么这就变成了 queue 模型

    • 所有的 消费者 都在 不同的组（group） 中，那么就完全变成了 发布-订阅 模型

消费有序性

应用场景：

  • 即时消息中的 单对单聊天 和 群聊，保证 发送方消息 发送顺序 与 接收方的顺序 一致

  • 充值转账 两个渠道 在 同一个时间 进行 余额变更，短信通知 必须要 有顺序

  • …

举例

  • kafka集群 托管 4 个分区（P0-P3），2个 消费者 组，消费组A 有 2 个 消费者，消费组B 有 4 个
  • topic分区 中 消息 只能由 消费者组 中的 唯一一个消费者 处理，所以消息肯定是按照 先后顺序 进行处理的
  • 但是它也仅仅是保证 Topic 的 一个分区 顺序处理，不能保证跨分区的消息先后处理顺序。 所以，如果想要顺序的处理Topic的所有消息，那就只提供一个分区

提交 和 偏移量

kafka不会像其他 JMS 队列那样需要得到 消费者的确认，消费者可以使用 kafka 来追踪消息在 分区的位置（偏移量）

消费者会往一个叫做 _consumer_offset 的 特殊主题 发送消息，消息里包含了每个分区的 偏移量

如果消费者发生 崩溃 或 有新的消费者加入群组，就会 触发再均衡

再均衡

如果 消费者2 挂掉以后，会发生再均衡，消费者2负责的分区会被其他消费者进行消费，再均衡后不可避免会出现一些问题：

  • 问题1：如果提交偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理

    问题1

  • 问题2：如果提交的偏移量大于客户端的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失

    问题2

如果想要解决这些问题，还要知道目前 kafka 提交偏移量的方式：

  • 提交偏移量的方式有两种，分别是 自动提交偏移量 和 手动提交偏移量

    • 自动提交偏移量

      • 当 enable.auto.commit 被设置为 true，提交方式就是让 消费者 自动提交偏移量，每隔 5 秒消费者会自动把从 poll() 方法接收的 最大偏移量 提交上去

    • 手动提交偏移量

      • 手动提交 ，当 enable.auto.commit 被设置为 false 可以有以下三种提交方式：

        • 提交当前偏移量（同步提交）

          1
          2
          3
          4
          5
          6
          7
          8
          9
          10
          // 手动添加偏移量
          prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

          // 在拉去消息代码中添加
          try {
              // 同步提交偏移量
              consumer.commitSync();
          } catch (CommitFailedException e) {
              System.out.println("记录提交失败的异常: " + e);
          }

        • 异步提交

          1
          2
          3
          4
          5
          6
          7
          8
          9
          10
          11
          12
          13
          // 手动添加偏移量
          prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

          // 异步方式提交偏移量
          consumer.commitAsync(new OffsetCommitCallback() {
              @Override
              public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                  
                  if (e != null) {
                      System.out.println("记录错误的提交偏移量：" + map + "，异常信息为：" + e);
                  }
              }
          });

        • 同步和异步组合提交

          1
          2
          3
          4
          5
          6
          7
          8
          9
          10
          11
          12
          13
          14
          15
          16
          17
          18
          19
          20
          21
          22
          // 手动添加偏移量
          prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

          try {
              while (true) {
                  ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                  for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                      System.out.println(consumerRecord.key());
                      System.out.println(consumerRecord.value());
                  }
                  
                  // 异步提交偏移量
                  consumer.commitAsync();
              }
          } catch (Exception e) {
              e.printStackTrace();
              System.out.println("记录错误的信息：" + e);
          } finally {
              
              // 同步提交偏移量
              consumer.commitSync();
          }

SpringBoot集成Kafka收发消息

流程：

  1. 导入依赖

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- kafkfa -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
    </dependencies>

  2. 在 resources 中创建文件 application.yml

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    server:
      port: 9991
    spring:
      application:
        name: kafka-demo
      kafka:
        bootstrap-servers: 192.168.88.129:9092
        producer:
          retries: 10
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:
          group-id: ${spring.application.name}-test
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

  3. 消费生产者

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    // 在controller包下

    @RestController
    public class HelloController {

        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;

        @GetMapping("/hello")
        public String hello() {
            kafkaTemplate.send("itcast-topic", "程序员");
            return "ok";
        }
    }

  4. 消费消费者

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    // 在listener包下

    @Component
    public class HelloListener {

        @KafkaListener(topics = "itcast-topic")
        public void onMessage(String message) {
            if (!StringUtils.isEmpty(message)) {
                System.out.println(message);
            }
        }
    }

传递消息为对象

流程：

1、2 同上

消费生产者：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@RestController
public class HelloController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/hello")
    public String hello() {
        // kafkaTemplate.send("itcast-topic", "程序员");
        User user = new User();
        user.setUsername("xiaowang");
        user.setAge(18);

        kafkaTemplate.send("user-topic", JSON.toJSONString(user));

        return "ok";
    }
}

消费消费者：

1
2
3
4
5
6
7
8
9
10
11
@Component
public class HelloListener {

    @KafkaListener(topics = "user-topic")
    public void onMessage(String message) {
        if (!StringUtils.isEmpty(message)) {
            User user = JSON.parseObject(message, User.class);
            System.out.println(user);
        }
    }
}

自媒体文章上下架功能完成

自媒体端

自媒体文章上下架功能完成

接口定义：

说明
接口路径 /api/v1/news/down_or_up
请求方式 POST
参数 DTO
响应结果 ResponseResult

DTO

1
2
3
4
5
6
7
8
9
10
@Data
public class WmNewsDto {
    
    private Integer id;
    /**
    * 是否上架  0 下架  1 上架
    */
    private Short enable;
                       
}

流程：

  1. 在 heima-leadnews-wemedia 工程下的 WmNewsController 新增方法

    1
    2
    3
    4
    @PostMapping("/down_or_up")
    public ResponseResult downOrUp(@RequestBody WmNewsDto dto){
        return null;
    }

  2. 在 WmNewsDto 中新增 enable 属性

  3. 在 WmNewsService 新增方法

    1
    2
    3
    4
    5
    6
    /**
     * 文章的上下架
     * @param dto
     * @return
     */
    public ResponseResult downOrUp(WmNewsDto dto);

  4. 实现方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    /**
     * 文章的上下架
     *
     * @param wmNewsDto
     * @return
     */
    @Override
    public ResponseResult downOrUp(WmNewsDto wmNewsDto) {
        
        // 1. 检查参数
        if (wmNewsDto.getId() == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        
        // 2. 查询文章
        WmNews wmNews = getById(wmNewsDto.getId());
        if (wmNews == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.DATA_NOT_EXIST, "文章不存在");
        }
        
        // 3. 判断文章是否已发布
        if (!wmNews.getStatus().equals(WmNews.Status.PUBLISHED.getCode())) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID, "当前文章不是发布状态，不能上下架");
        }
        
        // 4. 修改文章 enable
        if (wmNewsDto.getEnable() != null && wmNewsDto.getEnable() > -1 && wmNewsDto.getEnable() < 2) {
            update(Wrappers.<WmNews>lambdaUpdate().set(WmNews::getEnable, wmNewsDto.getEnable())
                    .eq(WmNews::getId, wmNewsDto.getId()));
        }
        
        return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
    }

article端

流程：

  1. 导入 Kafka 依赖（在 heima-leadnews-common模块中 ）

    1
    2
    3
    4
    5
    6
    7
    8
    9
    <!-- kafkfa -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
    </dependency>

  2. 在 自媒体端 的 nacos 配置中心配置 kafka 的生产者（在 heima-leadnews-wemedia模块中 ）

    1
    2
    3
    4
    5
    6
    7
    spring:
      kafka:
        bootstrap-servers: 192.168.88.129:9092
        producer:
          retries: 10
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer

  3. 在 自媒体端 文章上下架后 发送消息

    1
    2
    3
    4
    5
    6
    7
    8
    // 4. 修改文章 enable
    // ...
        if (wmNews.getArticleId() != null) {
            // 发送消息，通知article修改文章的配置
            Map<String, Object> map = new HashMap<>();
            map.put("articleId", wmNews.getArticleId());
            map.put("enable", wmNewsDto.getEnable());
            kafkaTemplate.send(WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC, JSON.toJSONString(map));

  4. 定义一个 常量类

    1
    2
    3
    public class WmNewsMessageConstants {
        public static final String WM_NEWS_UP_OR_DOWN_TOPIC="wm.news.up.or.down.topic";
    }

  5. 在 article 端的 nacos 配置中心配置 kafka 的消费者（在 heima-leadnews-article 模块中）

    1
    2
    3
    4
    5
    6
    7
    spring:
      kafka:
        bootstrap-servers: 192.168.88.129:9092
        consumer:
          group-id: ${spring.application.name}
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

  6. 在 article 端编写监听，接收数据

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    @Component
    @Slf4j
    public class ArticleDownListener {

        @Autowired
        private ApArticleConfigService apArticleConfigService;

        @KafkaListener(topics = WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC)
        public void onMessage(String message) {
            if (StringUtils.isNotBlank(message)) {
                Map map = JSON.parseObject(message, Map.class);
                apArticleConfigService.updateByMap(map);
            }
        }
    }

  7. 修改 ap_article_config 表的数据

  8. 新建 ApArticleConfigService

    1
    2
    3
    4
    5
    6
    7
    8
    public interface ApArticleConfigService extends IService<ApArticleConfig> {
        /**
         * 修改文章
         *
         * @param map
         */
        void updateByMap(Map map);
    }

  9. 实现类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    @Service
    @Transactional
    public class ApArticleConfigServiceImpl extends ServiceImpl<ApArticleConfigMapper, ApArticleConfig> implements ApArticleConfigService {

        /**
         * 修改文章
         *
         * @param map
         */
        @Override
        public void updateByMap(Map map) {

            // 0 下架 1 上架
            Object enable = map.get("enable");
            boolean isDown = true;
            if (enable.equals(1)) {
                isDown = false;
            }

            // 修改文章
            update(Wrappers.<ApArticleConfig>lambdaUpdate()
                    .eq(ApArticleConfig::getArticleId, map.get("articleId"))
                    .set(ApArticleConfig::getIsDown, isDown));

        }
    }

app端文章搜索

搭建 ElasticSearch 环境

流程：

  1. 拉取镜像

    1
    docker pull elasticsearch:7.4.0

  2. 创建容器

    1
    docker run -id --name elasticsearch -d --restart=always -p 9200:9200 -p 9300:9300 -v /usr/share/elasticsearch/plugins:/usr/share/elasticsearch/plugins -e "discovery.type=single-node" elasticsearch:7.4.0

  3. 配置中文分词器 ik

    1
    2
    3
    4
    5
    6
    cd /usr/share/elasticsearch/plugins

    sudo mkdir analysis-ik

    # 如果没有rz命令，执行 sudo apt install lrzsz
    rz
    • 这部分安装详情见 ElasticSearch 的笔记部分

  4. Postman 测试

app端文章搜索功能实现

功能实现：

  • 用户输入关键可搜索文章列表

  • 关键词高亮显示

  • 文章列表展示与 home 展示一样，当用户点击某一篇文章，可查看文章详情

思路分析：

思路分析

创建对应的索引和映射：

可以在 kibana 创建索引库，具体操作说明见之前的 ES 笔记

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
PUT /app_info_article
{
  "mappings": {
    "properties": {
      "id": {
        "type": "long"
      },
      "publishTime": {
        "type": "date"
      },
      "layout": {
        "type": "integer"
      },
      "images": {
        "type": "keyword",
        "index": false
      },
      "staticUrl": {
        "type": "keyword",
        "index": false
      },
      "authorId": {
        "type": "long"
      },
      "authorName": {
        "type": "text"
      },
      "title": {
        "type": "text",
        "analyzer": "ik_smart"
      },
      "content": {
        "type": "text",
        "analyzer": "ik_smart"
      }
    }
  }
}

数据初始化到索引库

流程：

  1. 导入 es-init 到 heima-leadnews-test 工程下，用于数据的导入

  2. 查询所有的文章信息，批量导入到 es 索引库中

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    @SpringBootTest
    @RunWith(SpringRunner.class)
    public class ApArticleTest {

        /**
         * 注意：数据量的导入，如果数据量过大，需要分页导入
         * @throws Exception
         */

        @Autowired
        private ApArticleMapper apArticleMapper;

        @Autowired
        private RestHighLevelClient restHighLevelClient;

        @Test
        public void init() throws Exception {

            // 1. 查询所有符合条件的文章数据
            List<SearchArticleVo> list = apArticleMapper.loadArticleList();

            // 2. 批量导入到es索引库

            BulkRequest request = new BulkRequest("app_info_article");

            for (SearchArticleVo searchArticleVo : list) {

                // IndexRequest 表示新增
                IndexRequest indexRequest = new IndexRequest()
                        .id(searchArticleVo.getId().toString())
                        .source(JSON.toJSONString(searchArticleVo), XContentType.JSON);

                // 批量添加数据
                request.add(indexRequest);
            }

            restHighLevelClient.bulk(request, RequestOptions.DEFAULT);

        }

    }

  3. 其中 ApArticleMapper 中 loadArticleList 的实现为

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    <mapper namespace="com.heima.es.mapper.ApArticleMapper">

        <resultMap id="resultMap" type="com.heima.es.pojo.SearchArticleVo">
            <id column="id" property="id"/>
            <result column="title" property="title"/>
            <result column="author_id" property="authorId"/>
            <result column="author_name" property="authorName"/>
            <result column="layout" property="layout"/>
            <result column="images" property="images"/>
            <result column="publish_time" property="publishTime"/>
            <result column="static_url" property="staticUrl"/>
            <result column="content" property="content"/>
        </resultMap>
        <select id="loadArticleList" resultMap="resultMap">
            SELECT
                aa.*, aacon.content
            FROM
                `ap_article` aa,
                ap_article_config aac,
                ap_article_content aacon
            WHERE
                aa.id = aac.article_id
              AND aa.id = aacon.article_id
              AND aac.is_delete != 1
              AND aac.is_down != 1

        </select>

    </mapper>

文章搜索功能实现

实现接口的定义：

文章搜索功能实现

UserSearchDto 的定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@Data
public class UserSearchDto {

    /**
    * 搜索关键字
    */
    String searchWords;
    /**
    * 当前页
    */
    int pageNum;
    /**
    * 分页条数
    */
    int pageSize;
    /**
    * 最小时间
    */
    Date minBehotTime;

    public int getFromIndex(){
        if(this.pageNum<1) return 0;
        if(this.pageSize<1) this.pageSize = 10;
        return this.pageSize * (pageNum-1);
    }
}

实现流程：

  1. 创建 heima-leadnews-search 的模块

  2. 在 heima-leadnews-service 的 pom 中添加依赖

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    <!--elasticsearch-->
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
        <version>7.12.1</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-client</artifactId>
        <version>7.12.1</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch</artifactId>
        <version>7.12.1</version>
    </dependency>

  3. 在 nacos 配置中心 leadnews-search

    1
    2
    3
    4
    5
    6
    7
    spring:
      autoconfigure:
      # 搜索微服务不需要连接数据库，要把该字段忽略
        exclude: org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
    elasticsearch:
      host: 192.168.88.129
      port: 9200

  4. 搜索接口的定义

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    @RestController
    @RequestMapping("/api/v1/article/search")
    public class ArticleSearchController {

        @Autowired
        private ArticleSearchService articleSearchService;

        @PostMapping("/search")
        public ResponseResult search(@RequestBody UserSearchDto dto) throws IOException {
            return articleSearchService.search(dto);
        }
    }

  5. 创建业务层接口：ApArticleSearchService

    1
    2
    3
    4
    5
    6
    7
    8
    9
    public interface ArticleSearchService {

        /**
         * es文章分页检索
         * @param dto
         * @return
         */
        public ResponseResult search(UserSearchDto dto) throws IOException;
    }

  6. 实现类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    @Service
    @Slf4j
    public class ArticleSearchServiceImpl implements ArticleSearchService {
        /**
         * es文章分页检索
         *
         * @param dto
         * @return
         */

        @Autowired
        private RestHighLevelClient restHighLevelClient;

        @Override
        public ResponseResult search(UserSearchDto dto) throws IOException {

            // 1. 检查参数
            if (dto == null || StringUtils.isBlank(dto.getSearchWords())) {
                return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
            }

            // 2. 设置查询条件
            SearchRequest request = new SearchRequest("app_info_article");

            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            // boolean查询
            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

            // 2.1 关键字的分词之后查询
            QueryStringQueryBuilder queryStringQueryBuilder = QueryBuilders.queryStringQuery(dto.getSearchWords())
                    .field("title")
                    .field("content")
                    .defaultOperator(Operator.OR);

            boolQueryBuilder.must(queryStringQueryBuilder);

            // 2.2 查询小于mindate的数据
            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("publishTime")
                    .lt(dto.getMinBehotTime().getTime());

            boolQueryBuilder.filter(rangeQueryBuilder);

            // 2.3 分页查询
            searchSourceBuilder.from(0);
            searchSourceBuilder.size(dto.getPageSize());

            // 2.4 按照发布时间倒序查询
            searchSourceBuilder.sort("publishTime", SortOrder.DESC);

            // 2.5 设置高亮 title
            HighlightBuilder highlightBuilder = new HighlightBuilder();
            highlightBuilder.field("title");
            // 设置高亮的前缀
            highlightBuilder.preTags("<font style='color: red; font-size: inherit;'>");
            // 设置高亮的后缀
            highlightBuilder.postTags("</font>");
            searchSourceBuilder.highlighter(highlightBuilder);


            searchSourceBuilder.query(boolQueryBuilder);
            request.source(searchSourceBuilder);
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);


            // 3. 结果封装返回

            List<Map> list = new ArrayList<>();

            SearchHit[] hits = response.getHits().getHits();

            for (SearchHit hit : hits) {
                String json = hit.getSourceAsString();
                Map map = JSON.parseObject(json, Map.class);

                // 处理高亮
                if (hit.getHighlightFields() != null && hit.getHighlightFields().size() > 0) {
                    Text[] titles = hit.getHighlightFields().get("title").getFragments();
                    String title = StringUtils.join(titles);
                    // 高亮标题
                    map.put("h_title", title);
                } else {
                    // 原始标题
                    map.put("h_title", map.get("title"));
                }

                list.add(map);
            }

            return ResponseResult.okResult(list);
        }
    }

  7. 在 app 的网关中添加搜索微服务的路由配置

    1
    2
    3
    4
    5
    6
    7
    #搜索微服务
    - id: leadnews-search
     uri: lb://leadnews-search
     predicates:
       - Path=/search/**
     filters:
       - StripPrefix= 1

新增文章创建索引

思路分析：

思路分析

实现流程：

  1. 在 model 模块下定义 SearchArticleVo

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    @Data
    public class SearchArticleVo {

        // 文章id
        private Long id;
        // 文章标题
        private String title;
        // 文章发布时间
        private Date publishTime;
        // 文章布局
        private Integer layout;
        // 封面
        private String images;
        // 作者id
        private Long authorId;
        // 作者名词
        private String authorName;
        // 静态url
        private String staticUrl;
        // 文章内容
        private String content;
    }

  2. 在文章微服务的 ArticleFreemarkerService 中的 buildArticleToMinIO 方法中收集数据并发送消息

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    @Service
    @Slf4j
    @Transactional
    public class ArticleFreemarkerServiceImpl implements ArticleFreemarkerService {
        /**
         * 生成静态文件上传到minio中
         *
         * @param apArticle
         * @param content
         */

        @Autowired
        private Configuration configuration;

        @Autowired
        private FileStorageService fileStorageService;

        @Autowired
        private ApArticleService apArticleService;

        @Async
        @Override
        public void buildArticleToMinIO(ApArticle apArticle, String content) {
            // 已知文章id
            // 1. 获取文章内容
            if (StringUtils.isNotBlank(content)) {

                // 2. 文章内容通过freemarker生成html文件
                Template template = null;
                StringWriter out = null;
                try {
                    template = configuration.getTemplate("article.ftl");

                    // 数据模型
                    Map<String, Object> contentDataModel = new HashMap<>();

                    contentDataModel.put("content", JSONArray.parseArray(content));

                    out = new StringWriter();

                    // 合成
                    template.process(contentDataModel, out);

                } catch (Exception e) {
                    e.printStackTrace();
                }


                // 3. 把html文件上传到minio中
                InputStream in = new ByteArrayInputStream(out.toString().getBytes());
                String path = fileStorageService.uploadHtmlFile("", apArticle.getId() + ".html", in);

                // 4. 修改ap_article表，保存static_url字段
                apArticleService.update(Wrappers.<ApArticle>lambdaUpdate().eq(ApArticle::getId, apArticle.getId())
                        .set(ApArticle::getStaticUrl, path));

                // 发送消息，创建索引
                createArticleESIndex(apArticle, content, path);
            }
        }

        /**
         * 发送消息，创建索引
         *
         * @param apArticle
         * @param content
         * @param path
         */

        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;

        private void createArticleESIndex(ApArticle apArticle, String content, String path) {
            SearchArticleVo vo = new SearchArticleVo();
            BeanUtils.copyProperties(apArticle, vo);
            vo.setContent(content);
            vo.setStaticUrl(path);

            kafkaTemplate.send(ArticleConstants.ARTICLE_ES_SYNC_TOPIC, JSON.toJSONString(vo));
        }
    }

  3. 在 ArticleConstants 类中添加新的常量

    1
    2
    3
    4
    5
    6
    7
    8
    public class ArticleConstants {
        public static final Short LOADTYPE_LOAD_MORE = 1;
        public static final Short LOADTYPE_LOAD_NEW = 2;
        public static final String DEFAULT_TAG = "__all__";

        public static final String ARTICLE_ES_SYNC_TOPIC = "article.es.sync.topic";

    }

  4. 文章微服务集成 kafka 发送消息，在文章微服务的 nacos 的配置中心添加配置

    1
    2
    3
    4
    5
    6
    kafka:
        bootstrap-servers: 192.168.88.129:9092
        producer:
          retries: 10
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer

  5. 搜索微服务中添加 kafka 监听，在文章微服务的 nacos 的配置中心添加配置

    1
    2
    3
    4
    5
    6
    7
    spring:
      kafka:
        bootstrap-servers: 192.168.88.129:9092
        consumer:
          group-id: ${spring.application.name}
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

  6. 定义监听接收消息，保存索引数据

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    @Component
    @Slf4j
    public class SyncArticleListener {

        @Autowired
        private RestHighLevelClient restHighLevelClient;

        @KafkaListener(topics = ArticleConstants.ARTICLE_ES_SYNC_TOPIC)
        public void onMessage(String message){
            if (StringUtils.isNotBlank(message)) {

                log.info("SyncArticleListener，message = {}", message);

                SearchArticleVo searchArticleVo = JSON.parseObject(message, SearchArticleVo.class);

                IndexRequest indexRequest = new IndexRequest("app_info_article");

                indexRequest.id(searchArticleVo.getId().toString());

                indexRequest.source(message, XContentType.JSON);

                try {
                    restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
                } catch (IOException e) {
                    log.error("sync es error = {}", e);
                    throw new RuntimeException(e);
                }
            }
        }
    }

app端搜索 记录功能实现

需求分析：

  • 展示用户的搜索记录10条，按照搜索关键词的时间倒序
  • 可以删除搜索记录
  • 保存历史记录，保存10条，多余的则删除最久的历史记录

数据存储说明：

用户的搜索记录，需要给 每一个用户都保存一份，数据量较大，要求加载速度快，通常这样的数据存储到 mongodb 更合适，不建议直接存储到关系型数据库中

数据存储说明

MongoDB 安装与集成

流程：

  1. 直接在docker中安装MongoDB，先拉去镜像

    1
    docker pull mongo

  2. 创建容器

    1
    docker run -di --name mongo-service --restart=always -p 27017:27017 -v ~/data/mongodata:/data mongo

  3. 导入 mongo 依赖

    1
    2
    3
    4
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-mongodb</artifactId>
    </dependency>

  4. 设置 mongo 配置

    1
    2
    3
    4
    5
    6
    7
    8
    server:
      port: 9998
    spring:
      data:
        mongodb:
          host: 192.168.88.129
          port: 27017
          database: leadnews-history

  5. 创建集合的映射

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    /**
     * <p>
     * 联想词表
     * </p>
     *
     * @author itheima
     */
    @Data
    @Document("ap_associate_words")
    public class ApAssociateWords implements Serializable {

        private static final long serialVersionUID = 1L;

        private String id;

        /**
         * 联想词
         */
        private String associateWords;

        /**
         * 创建时间
         */
        private Date createdTime;

    }

  6. 测试 增删改查

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    @SpringBootTest(classes = MongoApplication.class)
    @RunWith(SpringRunner.class)
    public class MongoTest {


        @Autowired
        private MongoTemplate mongoTemplate;

        // 保存
        @Test
        public void saveTest() {
            ApAssociateWords apAssociateWords = new ApAssociateWords();
            apAssociateWords.setAssociateWords("今日头条");
            apAssociateWords.setCreatedTime(new Date());
            mongoTemplate.save(apAssociateWords);
        }

        // 查询一条信息
        @Test
        public void saveFindOne() {
            ApAssociateWords apAssociateWords = mongoTemplate
                    .findById("66a0d49c87e1db149c1afa76", ApAssociateWords.class);
            System.out.println(apAssociateWords);
        }

        // 条件查询
        @Test
        public void testQuery() {
            Query query = Query.query(Criteria.where("associateWords").is("今日头条"))
                    .with(Sort.by(Sort.Direction.DESC, "createdTime"));
            List<ApAssociateWords> apAssociateWordsList = mongoTemplate.find(query, ApAssociateWords.class);
            System.out.println(apAssociateWordsList);
        }

        // 删除
        @Test
        public void testDel() {
            mongoTemplate.remove(Query.query(Criteria.where("associateWords").is("今日头条")), ApAssociateWords.class);
        }
    }

保存搜索记录 功能实现

实现思路：

实现思路

  • 用户输入关键字进行搜索的异步记录关键字

整体保存记录实现思路

用户搜索记录对应的集合，生成对应实体类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@Data
@Document("ap_user_search")
public class ApUserSearch implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * 主键
     */
    private String id;

    /**
     * 用户ID
     */
    private Integer userId;

    /**
     * 搜索词
     */
    private String keyword;

    /**
     * 创建时间
     */
    private Date createdTime;

}

流程：

  1. 导入依赖

    1
    2
    3
    4
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-mongodb</artifactId>
    </dependency>

  2. 设置nacos配置

    1
    2
    3
    4
    5
    6
    spring:
      data:
       mongodb:
        host: 192.168.88.129
        port: 27017
        database: leadnews-history

  3. 创建 ApUserSearchService 新增 insert 方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    @Service
    @Slf4j
    public class ApUserSearchServiceImpl implements ApUserSearchService {
        /**
         * 保存用户搜索历史记录
         *
         * @param keyword 搜索关键字
         * @param userId  当前线程传过来的userid
         */

        @Autowired
        private MongoTemplate mongoTemplate;

        @Override
        @Async
        public void insert(String keyword, Integer userId) {
            // 1. 查询当前用户的搜索关键词
            Query query = Query.query(Criteria
                    .where("userId")
                    .is(userId)
                    .and("keyword")
                    .is(keyword));
            ApUserSearch apUserSearch = mongoTemplate.findOne(query, ApUserSearch.class);

            // 2. 存在，更新创建时间
            if (apUserSearch != null) {
                apUserSearch.setCreatedTime(new Date());
                mongoTemplate.save(apUserSearch);
                return;
            }

            // 3. 不存在，判断当前历史记录总数量是否超过10
            apUserSearch = new ApUserSearch();
            apUserSearch.setUserId(userId);
            apUserSearch.setKeyword(keyword);
            apUserSearch.setCreatedTime(new Date());

            Query query1 = Query.query(Criteria.where("userId").is(userId));
            query1.with(Sort.by(Sort.Direction.DESC, "createdTime"));

            List<ApUserSearch> apUserSearchList = mongoTemplate.find(query1, ApUserSearch.class);

            if (apUserSearchList == null || apUserSearchList.size() < 10) {
                mongoTemplate.save(apUserSearch);
            } else {
                ApUserSearch lastUserSearch = apUserSearchList.get(apUserSearchList.size() - 1);
                mongoTemplate.findAndReplace(Query.query(Criteria.where("id").is(lastUserSearch.getId())), apUserSearch);
            }
        }
    }

  4. 参考自媒体相关微服务，在搜索微服务中获取当前登录的用户（ThreadLocalUtils）

    • 流程：在app端网关中获取用户id存到header中，在utils下创建AppThreadLocalUtil，在search下创建一个拦截器，拦截用户id，同时在config包下创建WebMvcConfig，配置拦截器

  5. 在 ArticleSearchService 的search方法中调用保存历史记录

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    @Service
    @Slf4j
    public class ArticleSearchServiceImpl implements ArticleSearchService {
        /**
         * es文章分页检索
         *
         * @param dto
         * @return
         */

        @Autowired
        private RestHighLevelClient restHighLevelClient;

        @Autowired
        private ApUserSearchService apUserSearchService;

        @Override
        public ResponseResult search(UserSearchDto dto) throws IOException {

            // 1. 检查参数
            if (dto == null || StringUtils.isBlank(dto.getSearchWords())) {
                return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
            }

            // TODO：这里新增
            // 异步调用 保存搜索记录

            // 1.1 获取当前的用户
            ApUser apUser = AppThreadLocalUtil.getUser();

            if (apUser != null && dto.getFromIndex() == 0) {
                apUserSearchService.insert(dto.getSearchWords(), apUser.getId());
            }

            // 2. 设置查询条件
            SearchRequest request = new SearchRequest("app_info_article");

            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            // boolean查询
            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

            // 2.1 关键字的分词之后查询
            QueryStringQueryBuilder queryStringQueryBuilder = QueryBuilders.queryStringQuery(dto.getSearchWords())
                    .field("title")
                    .field("content")
                    .defaultOperator(Operator.OR);

            boolQueryBuilder.must(queryStringQueryBuilder);

            // 2.2 查询小于mindate的数据
            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("publishTime")
                    .lt(dto.getMinBehotTime().getTime());

            boolQueryBuilder.filter(rangeQueryBuilder);

            // 2.3 分页查询
            searchSourceBuilder.from(0);
            searchSourceBuilder.size(dto.getPageSize());

            // 2.4 按照发布时间倒序查询
            searchSourceBuilder.sort("publishTime", SortOrder.DESC);

            // 2.5 设置高亮 title
            HighlightBuilder highlightBuilder = new HighlightBuilder();
            highlightBuilder.field("title");
            // 设置高亮的前缀
            highlightBuilder.preTags("<font style='color: red; font-size: inherit;'>");
            // 设置高亮的后缀
            highlightBuilder.postTags("</font>");
            searchSourceBuilder.highlighter(highlightBuilder);


            searchSourceBuilder.query(boolQueryBuilder);
            request.source(searchSourceBuilder);
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);


            // 3. 结果封装返回

            List<Map> list = new ArrayList<>();

            SearchHit[] hits = response.getHits().getHits();

            for (SearchHit hit : hits) {
                String json = hit.getSourceAsString();
                Map map = JSON.parseObject(json, Map.class);

                // 处理高亮
                if (hit.getHighlightFields() != null && hit.getHighlightFields().size() > 0) {
                    Text[] titles = hit.getHighlightFields().get("title").getFragments();
                    String title = StringUtils.join(titles);
                    // 高亮标题
                    map.put("h_title", title);
                } else {
                    // 原始标题
                    map.put("h_title", map.get("title"));
                }

                list.add(map);
            }

            return ResponseResult.okResult(list);
        }
    }

  6. 保存历史记录中开启异步调用，添加注解 @Async，以及在启动类添加注解 @EnableAsync

加载搜索历史 功能实现

接口说明：

接口说明

  • 按照当前用户，按照时间倒序查询

流程：

  1. 在 controller 包下定义接口

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    @RestController
    @RequestMapping("/api/v1/history")
    public class ApUserSearchController {

        @Autowired
        private ApUserSearchService apUserSearchService;
        @PostMapping("/load")
        public ResponseResult findUserSearch(){
            return apUserSearchService.findUserSearch();
        }
    }

  2. 在 ApUserSearchService 中新增方法

    1
    2
    3
    4
    5
    /**
     * 查询搜索历史
     * @return
     */
    public ResponseResult findUserSearch();

  3. 实现方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    /**
     * 查询搜索历史
     *
     * @return
     */
    @Override
    public ResponseResult findUserSearch() {
        
        // 1. 获取当前用户
        ApUser user = AppThreadLocalUtil.getUser();
        if (user == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.NEED_LOGIN);
        }
        
        // 2. 根据用户查询数据，按照时间倒序
        Query query = Query.query(Criteria.where("userId")
                        .is(user.getId()))
                .with(Sort.by(Sort.Direction.DESC, "createdTime"));
        List<ApUserSearch> apUserSearches = mongoTemplate.find(query, ApUserSearch.class);
        return ResponseResult.okResult(apUserSearches);
    }

删除搜索历史 功能实现

接口说明：

接口说明

流程：

  1. 在 ApUserSearchController 接口新增方法

    1
    2
    3
    4
    @PostMapping("/del")
    public ResponseResult delUserSearch(@RequestBody HistorySearchDto dto) {
        return apUserSearchService.delUserSearch(dto);
    }

  2. 创建 HistorySearchDto

    1
    2
    3
    4
    5
    6
    7
    @Data
    public class HistorySearchDto {
        /**
        * 接收搜索历史记录id
        */
        String id;
    }

  3. 在 ApUserSearchService 中新增方法

    1
    2
    3
    4
    5
    6
    /**
     * 删除历史记录
     * @param dto
     * @return
     */
    public ResponseResult delUserSearch(HistorySearchDto dto);

  4. 实现方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    /**
     * 删除历史记录
     *
     * @param dto
     * @return
     */
    @Override
    public ResponseResult delUserSearch(HistorySearchDto dto) {
        // 1. 检查参数
        if (dto.getId() == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        
        // 2. 判断是否登录
        ApUser user = AppThreadLocalUtil.getUser();
        if (user == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.NEED_LOGIN);
        }
        
        // 3. 删除
        Query query = Query.query(Criteria
                .where("userId")
                .is(user.getId())
                .and("id")
                .is(dto.getId()));
        mongoTemplate.remove(query, ApUserSearch.class);
        return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
    }

app端搜索 关键字联系词实现

需求分析：

根据用户输入的关键字展示联想词

对应的实体类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
/**
 * <p>
 * 联想词表
 * </p>
 *
 * @author itheima
 */
@Data
@Document("ap_associate_words")
public class ApAssociateWords implements Serializable {

    private static final long serialVersionUID = 1L;

    private String id;

    /**
     * 联想词
     */
    private String associateWords;

    /**
     * 创建时间
     */
    private Date createdTime;

}

搜索词-数据来源:

通常是网上搜索频率比较高的一些词，通常在企业中有两部分来源：

  1. 自己维护搜索词
    • 通过分析用户搜索频率较高的词，按照排名作为搜索词
  2. 第三方获取
    • 关键词规划师（百度）、5118、爱站网

接口说明：

接口说明

正则表达式说明：

正则表达式说明

流程：

  1. 在 ApAssociateWordsController 下新建接口

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    @RestController
    @RequestMapping("/api/v1/associate")
    public class ApAssociateWordsController {

        @Autowired
        private ApAssociateWordsService apAssociateWordsService;

        @PostMapping("/search")
        public ResponseResult search(@RequestBody UserSearchDto dto) {
            return apAssociateWordsService.search(dto);
        }
    }

  2. 在 ApAssociateWordsService 下新建 联想词业务层 接口

    1
    2
    3
    4
    5
    6
    7
    8
    9
    public interface ApAssociateWordsService {

        /**
         * 搜索联想词
         * @param dto
         * @return
         */
        public ResponseResult search(UserSearchDto dto);
    }

  3. 实现类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    @Service
    @Slf4j
    public class ApAssociateWordsServiceImpl implements ApAssociateWordsService {
        /**
         * 搜索联想词
         *
         * @param dto
         * @return
         */

        @Autowired
        private MongoTemplate mongoTemplate;

        @Override
        public ResponseResult search(UserSearchDto dto) {

            // 1. 检查参数
            if (StringUtils.isBlank(dto.getSearchWords())) {
                return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
            }

            // 2. 分页检查
            if (dto.getPageSize() > 20) {
                dto.setPageSize(20);
            }

            // 3. 执行查询，模糊查询
            // regex 部分见上面的正则表达式说明
            Query query = Query.query(Criteria
                    .where("associateWords")
                    .regex(".*?\\" + dto.getSearchWords() + ".*"));
            query.limit(dto.getPageSize());
            List<ApAssociateWords> apAssociateWords = mongoTemplate.find(query, ApAssociateWords.class);

            return ResponseResult.okResult(apAssociateWords);
        }
    }

热点文章 - 定时计算

目前存在的弊端

目前实现的思路：从数据库直接按照发布 时间倒序 查询

  • 问题1：

    如何访问量较大，直接查询数据库，压力较大

  • 问题2：

    新发布的文章会展示在前面，并不是热点文章

解决方案：

把 热点数据 存入 redis 进行展示

  • 判断文章是否是热点，有几项标准： 点赞数量，评论数量，阅读数量，收藏数量 …

计算文章热度，有两种方案：

  • 定时计算 文章热度

  • 实时计算 文章热度

定时计算

定时计算

  • 根据文章的行为（点赞、评论、阅读、收藏）计算文章的分值，利用定时任务每天完成一次计算

  • 把分值较大的文章数据存入到 redis 中

  • App端用户查询文章列表的时候，优先从 redis 中查询热度较高的文章数据

定时任务框架 - xxljob

spring 传统的定时任务 @Scheduled，但是这样存在这一些问题 ：

  • 做集群任务的重复执行问题

  • cron表达式定义在代码之中，修改不方便

  • 定时任务失败了，无法重试也没有统计

  • 如果任务量过大，不能有效的分片执行

解决这些问题的方案为：

  • xxl-job 分布式任务调度框架

分布式任务调度

在分布式架构下，一个服务往往会部署多个实例来运行我们的业务，如果在这种分布式系统环境下运行任务调度，称之为 分布式任务调度

分布式任务调度

将任务调度程序分布式构建，这样就可以具有分布式系统的特点，并且提高任务的调度处理能力：

  1. 并行任务调度
    • 并行任务调度实现靠多线程，如果有大量任务需要调度，此时光靠多线程就会有瓶颈了，因为一台计算机CPU的处理能力是有限的
    • 如果将任务调度程序分布式部署，每个结点还可以部署为集群，这样就可以让多台计算机共同去完成任务调度，可以将任务分割为若干个分片，由不同的实例并行执行，来提高任务调度的处理效率。
  2. 高可用
    • 若某一个实例宕机，不影响其他实例来执行任务。
  3. 弹性扩容
    • 当集群中增加实例就可以提高并执行任务的处理效率。
  4. 任务管理与监测
    • 对系统中存在的所有定时任务进行统一的管理及监测。让开发人员及运维人员能够时刻了解任务执行情况，从而做出快速的应急处理响应。

分布式任务调度面临的问题：

  • 当任务调度以集群方式部署，同一个任务调度可能会执行多次，例如：电商系统定期发放优惠券，就可能重复发放优惠券，对公司造成损失，信用卡还款提醒就会重复执行多次，给用户造成烦恼，所以需要控制相同的任务在多个运行实例上只执行一次。常见解决方案：
    • 分布式锁，多个实例在任务执行前首先需要获取锁，如果获取失败那么就证明有其他服务已经在运行，如果获取成功那么证明没有服务在运行定时任务，那么就可以执行
    • ZooKeeper选举，利用 ZooKeeper 对 Leader 实例执行定时任务，执行定时任务的时候判断自己是否是Leader，如果不是则不执行，如果是则执行业务逻辑，这样也能达到目的

xxl-Job 介绍

XXL-JOB 是一个分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用

其他同类型产品

xxl-job 源码说明

配置部署 “调度中心”

作用：统一管理任务调度平台上调度任务，负责触发调度执行，并且提供任务管理平台

流程：

  1. 初始化 “调度数据库”

    • 位置：/xxl-job/doc/db/tables_xxl_job.sql 共8张表

      表结构

  2. 调度中心配置

    • 配置文件地址：/xxl-job/xxl-job-admin/src/main/resources/application.properties
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    ### web
    server.port=8080
    server.servlet.context-path=/xxl-job-admin

    ### actuator
    management.server.servlet.context-path=/actuator
    management.health.mail.enabled=false

    ### resources
    spring.mvc.servlet.load-on-startup=0
    spring.mvc.static-path-pattern=/static/**
    spring.resources.static-locations=classpath:/static/

    ### freemarker
    spring.freemarker.templateLoaderPath=classpath:/templates/
    spring.freemarker.suffix=.ftl
    spring.freemarker.charset=UTF-8
    spring.freemarker.request-context-attribute=request
    spring.freemarker.settings.number_format=0.##########
    spring.freemarker.settings.new_builtin_class_resolver=safer

    ### mybatis
    mybatis.mapper-locations=classpath:/mybatis-mapper/*Mapper.xml
    #mybatis.type-aliases-package=com.xxl.job.admin.core.model

    ### xxl-job, datasource
    ### 这块要修改，连接本地自己的数据库
    spring.datasource.url=jdbc:mysql://localhost:3306/xxl_job?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=Asia/Shanghai
    spring.datasource.username=root
    spring.datasource.password=123456
    spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

    ### datasource-pool
    spring.datasource.type=com.zaxxer.hikari.HikariDataSource
    spring.datasource.hikari.minimum-idle=10
    spring.datasource.hikari.maximum-pool-size=30
    spring.datasource.hikari.auto-commit=true
    spring.datasource.hikari.idle-timeout=30000
    spring.datasource.hikari.pool-name=HikariCP
    spring.datasource.hikari.max-lifetime=900000
    spring.datasource.hikari.connection-timeout=10000
    spring.datasource.hikari.connection-test-query=SELECT 1
    spring.datasource.hikari.validation-timeout=1000

    ### xxl-job, email
    spring.mail.host=smtp.qq.com
    spring.mail.port=25
    spring.mail.username=xxx@qq.com
    spring.mail.from=xxx@qq.com
    spring.mail.password=xxx
    spring.mail.properties.mail.smtp.auth=true
    spring.mail.properties.mail.smtp.starttls.enable=true
    spring.mail.properties.mail.smtp.starttls.required=true
    spring.mail.properties.mail.smtp.socketFactory.class=javax.net.ssl.SSLSocketFactory

    ### xxl-job, access token
    xxl.job.accessToken=default_token

    ### xxl-job, i18n (default is zh_CN, and you can choose "zh_CN", "zh_TC" and "en")
    xxl.job.i18n=zh_CN

    ## xxl-job, triggerpool max size
    xxl.job.triggerpool.fast.max=200
    xxl.job.triggerpool.slow.max=100

    ### xxl-job, log retention days
    xxl.job.logretentiondays=30

  3. 启动调度中心，调度中心网址是：http://localhost:8080/xxl-job-admin/，默认登录账号：admin，密码：123456

    调度中心界面

配置部署调度中心（docker安装）

流程：

  1. 创建MySQL容器，初始化xxl-job的SQL脚本

    1
    docker run --name mysql --restart=always -v /home/ljaer/mysql:/var/lib/mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root -e TZ=Asia/Shanghai -d mysql:8.0.26

  2. 拉取镜像

    1
    docker pull xuxueli/xxl-job-admin:2.3.0

  3. 创建容器

    1
    2
    3
    4
    5
    docker run -e PARAMS="--spring.datasource.url=jdbc:mysql://192.168.88.129:3306/xxl_job?Unicode=true&characterEncoding=UTF-8 \
    --spring.datasource.username=root \
    --spring.datasource.password=root" \
    -p 8888:8080 -v /tmp:/data/applogs \
    --name xxl-job-admin --restart=always  -d xuxueli/xxl-job-admin:2.3.0

  4. 在按照配置部署 “调度中心” 的步骤完成

xxl-Job 入门案例

流程：

  1. 登录调度中心，点击下图所示 “新建任务” 按钮，新建示例任务

    image-20240725134025684

  2. 创建 xxljob-demo 项目，导入依赖

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!--xxl-job-->
        <dependency>
            <groupId>com.xuxueli</groupId>
            <artifactId>xxl-job-core</artifactId>
            <version>2.3.0</version>
        </dependency>
    </dependencies>

  3. 设置 application.yml 配置

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    server:
      port: 8881


    xxl:
      job:
        admin:
          addresses: http://192.168.88.129:8888/xxl-job-admin
        executor:
          appname: xxl-job-executor-sample
          port: 9999

    executor.appname

  4. 在 config 目录下，新建配置类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    @Configuration
    public class XxlJobConfig {
        private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

        @Value("${xxl.job.admin.addresses}")
        private String adminAddresses;

        @Value("${xxl.job.executor.appname}")
        private String appname;

        @Value("${xxl.job.executor.port}")
        private int port;


        @Bean
        public XxlJobSpringExecutor xxlJobExecutor() {
            logger.info(">>>>>>>>>>> xxl-job config init.");
            XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
            xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
            xxlJobSpringExecutor.setAppname(appname);
            xxlJobSpringExecutor.setPort(port);
            return xxlJobSpringExecutor;
        }
    }

  5. 任务代码，重要注解：@XxlJob("JobHandler")

    @XxlJob注解内容

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    /**
     * 简单任务示例（Bean模式下）
     */
    @Component
    public class HelloJob {

        @XxlJob("JobHandler")
        public void HelloJob() {
            System.out.println("简单任务执行了 ... ");
        }
    }

    /**
     * 修改任务调度中心新任务的corn表达式
     * CRON：0/1 * * * * ? * 表示每秒钟执行一次
     */

任务详解

执行器

  • 执行器：任务的绑定的执行器，任务触发调度时将会自动发现注册成功的执行器，实现任务自动发现功能
  • 另一方面也可以方便的进行任务分组，每个任务必须绑定一个执行器

执行器

属性名称 说明
AppName 是每个执行器集群的唯一标识 AppName，执行器会周期性以 AppName 为对象进行自动注册。可通过该配置自动发现注册成功的执行器, 供任务调度时使用
名称 执行器的名称，因为 AppName 限制字母数字等组成，可读性不强，名称为了提高执行器的可读性
排序 执行器的排序，系统中需要执行器的地方，如任务新增，将会按照该排序读取可用的执行器列表
注册方式 调度中心获取执行器地址的方式
机器地址 注册方式为 “手动录入” 时有效，支持人工维护执行器的地址信息

基础配置

  • 执行器：每个任务必须绑定一个执行器, 方便给任务进行分组

  • 任务描述：任务的描述信息，便于任务管理

  • 负责人：任务的负责人

  • 报警邮件：任务调度失败时邮件通知的邮箱地址，支持配置多邮箱地址，配置多个邮箱地址时用逗号分隔

基础配置

调度类型

调度类型：

  • 无：该类型不会主动触发调度
  • CRON：该类型将会通过 CRON，触发任务调度
  • 固定速度：该类型将会以固定速度，触发任务调度，按照固定的间隔时间，周期性触发

调度类型

任务配置

  • 运行模式：
    • BEAN模式：任务以 JobHandler 方式维护在执行器端；需要结合 “JobHandler” 属性匹配执行器中任务
  • JobHandler：运行模式为 “BEAN模式” 时生效，对应执行器中新开发的 JobHandler 类 @JobHandler 注解自定义的value值
  • 执行参数：任务执行所需的参数

任务配置

阻塞处理策略

阻塞处理策略：调度过于密集执行器来不及处理时的处理策略

  • 单机串行（默认）：调度请求进入单机执行器后，调度请求进入 FIFO(First Input First Output) 队列并以串行方式运行

  • 丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败

  • 覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务

阻塞处理策略

路由策略

当执行器集群部署时，提供丰富的路由策略，包括；

  • FIRST（第一个）：固定选择第一个机器

  • LAST（最后一个）：固定选择最后一个机器

  • ROUND（轮询）

  • RANDOM（随机）：随机选择在线的机器

  • CONSISTENT_HASH（一致性HASH）：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上

  • LEAST_FREQUENTLY_USED（最不经常使用）：使用频率最低的机器优先被选举

  • LEAST_RECENTLY_USED（最近最久未使用）：最久未使用的机器优先被选举

  • FAILOVER（故障转移）：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度

  • BUSYOVER（忙碌转移）：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度

  • SHARDING_BROADCAST(分片广播)：广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务

路由策略

路由策略 （轮询）案例

流程：

  1. 修改任务为 轮询

    轮询模式

  2. 启动多个微服务，先编辑 application.yml 配置文件

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    server:
      port: ${port:8881}  # 默认端口号是8881


    xxl:
      job:
        admin:
          addresses: http://192.168.88.129:8888/xxl-job-admin
        executor:
          appname: xxl-job-executor-sample
          port: ${executor.port:9999} # 默认执行器的端口号是9999

    配置两个引导类

  3. 启动多个微服务，每个微服务轮询的去执行任务

路由策略（分片广播）

执行器集群部署时，任务路由策略选择 ”分片广播” 情况下，一次任务调度将会广播触发对应集群中所有执行器执行一次任务

分片广播策略

案例需求：让两个节点同时执行10000个任务，每个节点分别执行5000个任务

流程：

  1. 创建分片执行器

    分片执行器

  2. 创建任务，路由策略为分片广播

    路由策略：分片广播

  3. 修改 yml 配置

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    server:
      port: ${port:8881}  # 默认端口号是8881


    xxl:
      job:
        admin:
          addresses: http://192.168.88.129:8888/xxl-job-admin
        executor:
          # 主要修改 executor.appname
          appname: xxl-job-sharding-executor
          port: ${executor.port:9999} # 默认执行器的端口号是9999

  4. 分片广播代码

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    @Component
    public class HelloJob {
        

        @XxlJob("shardingJobHandler")
        public void shardingJobHandler() {

            // 获取分片的参数
            int shardIndex = XxlJobHelper.getShardIndex();

            // 获取总分片
            int shardTotal = XxlJobHelper.getShardTotal();

            // 业务逻辑
            List<Integer> list = getList();

            for (Integer integer : list) {
                if (integer % shardTotal == shardIndex) {
                    System.out.println("当前第 " + shardIndex
                            + " 分片执行了，任务项为：" + integer);
                }
            }
        }

        public List getList() {

            List<Integer> list = new ArrayList<>();

            for (int i = 0; i < 10000; i++) {
                list.add(i);
            }
            return list;
        }
    }
    • 分片参数
      • Shardindex：当前分片序号（从0开始），执行器集群列表中当前执行器的序号
      • Shardtotal：总分片数，执行器集群的总机器数量

热点文章 任务流程

需求分析：为每个频道缓存热度较高的30条文章优先展示

判断 文章热度较高 的标准是：阅读量，点赞，评论，收藏

对应ap_article表

实现思路：

实现思路

实现步骤

由于分值计算不涉及到前端工程，也无需提供 api 接口，是一个纯后台的功能的开发

  1. 由于计算完成新热数据后，需要给每个频道缓存一份数据，所以需要查询所有频道信息

    • 在 heima-leadnews-feign-api 定义远程接口

      1
      2
      3
      4
      5
      6
      @FeignClient("leadnews-wemedia")
      public interface IWemediaClient {

          @GetMapping("/api/v1/channel/list")
          public ResponseResult getChannels();
      }

    • 在 heima-leadnews-wemedia 端提供接口

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      @RestController
      public class WemediaClient implements IWemediaClient {

          @Autowired
          private WmChannelService wmChannelService;

          @GetMapping("/api/v1/channel/list")
          @Override
          public ResponseResult getChannels() {
              return wmChannelService.findAll();
          }
      }

  2. 在 ApArticleMapper.xml 新增方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    <select id="findArticleListByLast5days" resultMap="resultMap">
        SELECT
        aa.*
        FROM
        `ap_article` aa
        LEFT JOIN ap_article_config aac ON aa.id = aac.article_id
        <where>
            and aac.is_delete != 1
            and aac.is_down != 1
            <if test="dayParam != null">
                and aa.publish_time <![CDATA[>=]]> #{dayParam}
            </if>
        </where>
    </select>

  3. 修改 ApArticleMapper 类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    @Mapper
    public interface ApArticleMapper extends BaseMapper<ApArticle> {

        /**
         * 加载文章列表
         * @param dto
         * @param type  1  加载更多   2记载最新
         * @return
         */
        public List<ApArticle> loadArticleList(ArticleHomeDto dto,Short type);

        public List<ApArticle> findArticleListByLast5days(@Param("dayParam") Date dayParam);
    }

  4. 定义业务层接口

    1
    2
    3
    4
    5
    6
    7
    public interface HotArticleService {

        /**
         * 计算热点文章
         */
        public void computeHotArticle();
    }

  5. 修改 ArticleConstants 类

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    public class ArticleConstants {
        public static final Short LOADTYPE_LOAD_MORE = 1;
        public static final Short LOADTYPE_LOAD_NEW = 2;
        public static final String DEFAULT_TAG = "__all__";

        public static final String ARTICLE_ES_SYNC_TOPIC = "article.es.sync.topic";

        public static final Integer HOT_ARTICLE_LIKE_WEIGHT = 3;
        public static final Integer HOT_ARTICLE_COMMENT_WEIGHT = 5;
        public static final Integer HOT_ARTICLE_COLLECTION_WEIGHT = 8;

        public static final String HOT_ARTICLE_FIRST_PAGE = "hot_article_first_page_";
    }

  6. 创建一个 vo 接收计算分值后的对象

    1
    2
    3
    4
    5
    6
    7
    @Data
    public class HotArticleVo extends ApArticle {
        /**
         * 文章分值
         */
        private Integer score;
    }

  7. 实现业务层

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    @Service
    @Slf4j
    @Transactional
    public class HotArticleServiceImpl implements HotArticleService {
        /**
         * 计算热点文章
         */

        @Autowired
        private ApArticleMapper apArticleMapper;

        @Override
        public void computeHotArticle() {

            // 1. 查询前5天的文章数据
            Date dateParam = DateTime.now().minusDays(5).toDate();
            List<ApArticle> articleList = apArticleMapper.findArticleListByLast5days(dateParam);

            // 2. 计算文章的分值
            List<HotArticleVo> hotArticleVoList = computeHotArticle(articleList);

            // 3. 为每个频道缓存30条分值较高的文章
            catchTagToRedis(hotArticleVoList);

        }

        /**
         * 为每个频道缓存30条分值较高的文章
         *
         * @param hotArticleVoList
         */

        @Autowired
        private IWemediaClient wemediaClient;

        @Autowired
        private CacheService cacheService;

        private void catchTagToRedis(List<HotArticleVo> hotArticleVoList) {
            // 1. 为每个频道缓存30条分值较高的文章
            ResponseResult responseResult = wemediaClient.getChannels();

            if (responseResult.getCode().equals(200)) {
                String channelJson = JSON.toJSONString(responseResult.getData());
                List<WmChannel> wmChannels = JSON.parseArray(channelJson, WmChannel.class);

                // 检索出每个频道的文章
                if (wmChannels != null && wmChannels.size() > 0) {
                    for (WmChannel wmChannel : wmChannels) {
                        List<HotArticleVo> hotArticleVos = hotArticleVoList.stream()
                                .filter(x -> x.getChannelId().equals(wmChannel.getId()))
                                .collect(Collectors.toList());

                        // 给文章进行排序，取30条分值较高的文章存入redis
                        // key：为频道的id
                        // value：30条分值较高文章的数据

                        hotArticleVos = hotArticleVos.stream()
                                .sorted(Comparator
                                        .comparing(HotArticleVo::getScore)
                                        .reversed())
                                .collect(Collectors.toList());

                        if (hotArticleVos.size() > 30) {
                            hotArticleVos = hotArticleVos.subList(0, 30);
                        }

                        cacheService.set(ArticleConstants.HOT_ARTICLE_FIRST_PAGE + wmChannel.getId(), JSON.toJSONString(hotArticleVos));
                    }
                }
            }

            // 2. 设置推荐数据
            // 给文章进行排序，取30条分值较高的文章存入redis
            // key：为频道的id
            // value：30条分值较高文章的数据

            hotArticleVoList = hotArticleVoList.stream()
                    .sorted(Comparator
                            .comparing(HotArticleVo::getScore)
                            .reversed())
                    .collect(Collectors.toList());

            if (hotArticleVoList.size() > 30) {
                hotArticleVoList = hotArticleVoList.subList(0, 30);
            }

            cacheService.set(ArticleConstants.DEFAULT_TAG, JSON.toJSONString(hotArticleVoList));

        }

        /**
         * 计算文章的分值
         *
         * @param articleList
         * @return
         */
        private List<HotArticleVo> computeHotArticle(List<ApArticle> articleList) {

            List<HotArticleVo> hotArticleVoList = new ArrayList<>();

            if (articleList != null && articleList.size() > 0) {
                for (ApArticle apArticle : articleList) {
                    HotArticleVo hot = new HotArticleVo();
                    BeanUtils.copyProperties(apArticle, hot);
                    Integer score = computeScore(apArticle);
                    hot.setScore(score);
                    hotArticleVoList.add(hot);
                }
            }
            return hotArticleVoList;
        }

        /**
         * 计算文章的具体分值
         *
         * @param apArticle
         * @return
         */
        private Integer computeScore(ApArticle apArticle) {

            Integer score = 0;

            if (apArticle.getLikes() != null) {
                score += apArticle.getLikes() * ArticleConstants.HOT_ARTICLE_LIKE_WEIGHT;
            }

            if (apArticle.getViews() != null) {
                score += apArticle.getViews();
            }

            if (apArticle.getComment() != null) {
                score += apArticle.getComment() * ArticleConstants.HOT_ARTICLE_COMMENT_WEIGHT;
            }

            if (apArticle.getCollection() != null) {
                score += apArticle.getCollection() * ArticleConstants.HOT_ARTICLE_COLLECTION_WEIGHT;
            }

            return score;
        }
    }

  8. 在 ArticleApplication 的引导类中添加注解 @EnableFeignClients

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    @SpringBootApplication
    @EnableDiscoveryClient
    @MapperScan("com.heima.article.mapper")
    @EnableAsync
    @EnableFeignClients(basePackages = "com.heima.apis")
    public class ArticleApplication {

        public static void main(String[] args) {
            SpringApplication.run(ArticleApplication.class,args);
        }

        @Bean
        // Mybatis-Plus的分页插件
        public MybatisPlusInterceptor mybatisPlusInterceptor() {
            MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
            interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.MYSQL));
            return interceptor;
        }
    }

定时计算 任务流程

实现步骤

  1. 在 heima-leadnews-article 中的 pom 文件中新增依赖

    1
    2
    3
    4
    5
    6
    <!--xxl-job-->
    <dependency>
        <groupId>com.xuxueli</groupId>
        <artifactId>xxl-job-core</artifactId>
        <version>2.3.0</version>
    </dependency>

  2. 在 xxl-job-admin 中新建执行器和任务

    • 新建执行器：leadnews-hot-article-executor

      新建执行器

    • 新建任务：路由策略为 轮询，Cron表达式为 0 0 2 * * ?

      新建任务

  3. leadnews-article 中的 config 目录下集成 xxl-job

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    @Configuration
    public class XxlJobConfig {
        private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

        @Value("${xxl.job.admin.addresses}")
        private String adminAddresses;

        @Value("${xxl.job.executor.appname}")
        private String appname;

        @Value("${xxl.job.executor.port}")
        private int port;


        @Bean
        public XxlJobSpringExecutor xxlJobExecutor() {
            logger.info(">>>>>>>>>>> xxl-job config init.");
            XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
            xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
            xxlJobSpringExecutor.setAppname(appname);
            xxlJobSpringExecutor.setPort(port);
            return xxlJobSpringExecutor;
        }
    }

  4. 在 nacos 中的 leadnews-article 下配置新增配置

    1
    2
    3
    4
    5
    6
    7
    xxl:
      job:
        admin:
          addresses: http://192.168.200.130:8888/xxl-job-admin
        executor:
          appname: leadnews-hot-article-executor
          port: 9999

  5. 在 article 微服务中新建任务类，路径为 com.heima.article.job

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    @Component
    @Slf4j
    public class ComputeHotArticleJob {

        @Autowired
        private HotArticleService hotArticleService;

        @XxlJob("computerHotArticleJob")
        public void handle() {
            log.info("热文章分值计算调度任务开始执行...");
            hotArticleService.computeHotArticle();
            log.info("热文章分值计算调度任务结束执行...");
        }
    }

查询文章接口改造

思路分析：

思路分析

实现步骤

  1. 在 ApArticleService 中新增方法

    1
    2
    3
    4
    5
    6
    7
    8
    /**
     * 加载文章列表，并判断是不是首页
     * @param dto
     * @param type 1  加载更多  2  加载最新
     * @param firstPage true 是首页  false 非首页
     * @return
     */
    public ResponseResult load2(ArticleHomeDto dto, Short type, boolean firstPage);

  2. 实现方法 ApArticleServiceImpl

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    /**
     * 加载文章列表，并判断是不是首页
     *
     * @param dto
     * @param type      1  加载更多  2  加载最新
     * @param firstPage true 是首页  false 非首页
     * @return
     */
    @Autowired
    private CacheService cacheService;
    @Override
    public ResponseResult load2(ArticleHomeDto dto, Short type, boolean firstPage) {
        
        if (firstPage) {
            String jsonStr = cacheService.get(ArticleConstants.HOT_ARTICLE_FIRST_PAGE + dto.getTag());
            if (StringUtils.isNotBlank(jsonStr)) {
                List<HotArticleVo> hotArticleVoList = JSON.parseArray(jsonStr, HotArticleVo.class);
                return ResponseResult.okResult(hotArticleVoList);
            }
        }
        
        return load(dto, type);
    }

  3. 修改控制器

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    /**
     * 加载首页
     *
     * @param dto
     * @return
     */
    @PostMapping("/load")
    public ResponseResult load(@RequestBody ArticleHomeDto dto) {
        // return apArticleService.load(dto, ArticleConstants.LOADTYPE_LOAD_MORE);
        return apArticleService.load2(dto, ArticleConstants.LOADTYPE_LOAD_MORE, true);
    }

Kafka Stream

实时流式计算

批量数据 vs 流式数据

  • 流式计算就相当于上图的右侧扶梯，是 可以源源不断的产生数据，源源不断的接收数据，时间上是没有边界的

应用场景：

  • 日志分析

    • 网站的用户访问日志进行实时的分析，计算访问量，用户画像，留存率等等，实时的进行数据分析，帮助企业进行决策

  • 大屏看板统计

    • 可以实时的查看网站注册数量，订单数量，购买数量，金额等

  • 公交实时数据

    • 可以随时更新公交车方位，计算多久到达站牌等

  • 实时文章分值计算

    • 头条类文章的分值计算，通过用户的行为实时文章的分值，分值越高就越被推荐

技术方案选型：

  • Hadoop
  • Apche Storm
  • Flink
  • Kafka Stream
    • 可以轻松地将其嵌入任何 Java 应用程序中，并与用户为其流应用程序所拥有的任何现有打包，部署和操作工具集成

Kafka Stream 使用

概述

Kafka Stream：是 Apache Kafka 从 0.10 版本引入的一个新Feature。它是提供了对存储于 Kafka 内的数据进行流式处理和分析的功能

Kafka Stream 的特点：

  • Kafka Stream提供了一个非常简单而轻量的Library，它可以非常方便地嵌入任意Java应用中，也可以任意方式打包和部署
  • 除了 Kafka 外，无任何外部依赖
  • 充分利用 Kafka 分区机制实现水平扩展和顺序性保证
  • 通过可容错的 state store 实现高效的状态操作（如windowed join和aggregation）
  • 支持正好一次处理语义
  • 提供记录级的处理能力，从而实现毫秒级的低延迟
  • 支持基于事件时间的窗口操作，并且可处理晚到的数据（late arrival of records）
  • 同时提供底层的处理原语 Processor（类似于Storm的spout和bolt），以及高层抽象的 DSL（类似于Spark的map/group/reduce）

Kafka Stream

关键概念

Kafka Stream消息流转

  • 源处理器（Source Processor）：源处理器是一个没有任何上游处理器的特殊类型的流处理器。它从一个或多个kafka主题生成输入流。通过消费这些主题的消息并将它们转发到下游处理器
  • Sink处理器：Sink处理器是一个没有下游流处理器的特殊类型的流处理器。它接收上游流处理器的消息发送到一个指定的Kafka主题

KStream

  1. 数据结构类似于 map（key-value 键值对）

    KStream

  2. KStream 数据流（data stream），即是一段顺序的，可以无限长，不断更新的数据集

    KStream数据流

    • KStream 负责抽象的，就是数据流。与 Kafka 自身 topic 中的数据一样，类似日志，每一次操作都是 向其中插入（insert）新数据

Kafka Stream 入门案例

需求：求单词个数（word count）

需求分析

流程：

  1. 导入依赖

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
        <exclusions>
            <exclusion>
                <artifactId>connect-json</artifactId>
                <groupId>org.apache.kafka</groupId>
            </exclusion>
            <exclusion>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
            </exclusion>
        </exclusions>
    </dependency>

  2. 创建 kafka stream入门案例

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    /**
     * 流式处理
     */
    public class KafkaStreamQuickStart {

        public static void main(String[] args) {

            // 1. kafka的配置信息
            Properties properties = new Properties();
            properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.88.129:9092");
            // 获得kafkaStream的key的序列化器Serdes
            properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            // 获得kafkaStream的value的序列化器Serdes
            properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-quickstart");


            // 2. stream的构建器
            StreamsBuilder streamsBuilder = new StreamsBuilder();

            // 流式计算
            streamProcessor(streamsBuilder);

            // 3. 创建kafkaStream的对象
            KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);

            // 4. 开启流式计算
            kafkaStreams.start();
        }

        /**
         * 流式计算
         *
         * 消息的内容：hello kafka hello world
         *
         * @param streamsBuilder
         */
        private static void streamProcessor(StreamsBuilder streamsBuilder) {

            // 1. 创建kStream对象，同时指定在哪个topic中接收消息
            KStream<String, String> stream = streamsBuilder.stream("coo1heisenberg-topic-input");

            // 2. 处理消息的value值
            stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
                        @Override
                        public Iterable<String> apply(String value) {
                            String[] valAry = value.split(" ");
                            List<String> valList = Arrays.asList(valAry);
                            return valList;
                        }
                    // 按照value进行聚合处理
                    }).groupBy((key, value) -> value)
                    // 时间窗口
                    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                    // 统计单词的个数
                    .count()
                    // 转换为kstream对象
                    .toStream()
                    .map((key, value) -> {
                        System.out.println("key: " + key + ", value: " + value);
                        return new KeyValue<>(key.key().toString(), value.toString());
                    })
                    // 发送消息
                    .to("coo1heisenberg-topic-out");
        }
    }

  3. 测试准备

    使用生产者在 topic 为：coo1heisenberg_topic_input中发送多条消息

    使用消费者接收 topic 为：coo1heisenberg_topic_out

  4. 结果

    • 通过流式计算，会把生产者的多条消息汇总成一条发送到消费者中输出

SpringBoot 集成 Kafka Stream

流程：

  1. 自定配置参数

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    /**
     * 通过重新注册KafkaStreamsConfiguration对象，设置自定配置参数
     */

    @Setter
    @Getter
    @Configuration
    @EnableKafkaStreams
    @ConfigurationProperties(prefix="kafka")
    public class KafkaStreamConfig {
        private static final int MAX_MESSAGE_SIZE = 16* 1024 * 1024;
        private String hosts;
        private String group;
        @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
        public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
            Map<String, Object> props = new HashMap<>();
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, this.getGroup()+"_stream_aid");
            props.put(StreamsConfig.CLIENT_ID_CONFIG, this.getGroup()+"_stream_cid");
            props.put(StreamsConfig.RETRIES_CONFIG, 10);
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            return new KafkaStreamsConfiguration(props);
        }
    }

  2. 修改 application.yml 文件，在最下方添加自定义配置

    1
    2
    3
    kafka:
      hosts: 192.168.88.129:9092
      group: ${spring.application.name}

  3. 新增配置类，创建 KStream 对象，进行聚合

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    @Configuration
    @Slf4j
    public class KafkaStreamHelloList {

        @Bean
        public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
            // 1. 创建kStream对象，同时指定在哪个topic中接收消息
            KStream<String, String> stream = streamsBuilder.stream("coo1heisenberg-topic-input");

            // 2. 处理消息的value值
            stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
                        @Override
                        public Iterable<String> apply(String value) {
                            String[] valAry = value.split(" ");
                            List<String> valList = Arrays.asList(valAry);
                            return valList;
                        }
                        // 按照value进行聚合处理
                    }).groupBy((key, value) -> value)
                    // 时间窗口
                    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                    // 统计单词的个数
                    .count()
                    // 转换为kstream对象
                    .toStream()
                    .map((key, value) -> {
                        System.out.println("key: " + key + ", value: " + value);
                        return new KeyValue<>(key.key().toString(), value.toString());
                    })
                    // 发送消息
                    .to("coo1heisenberg-topic-out");

            return stream;
        }
    }
    • 可注入 StreamsBuilder
    • 返回值必须是 KStream 且放入spring容器中

  4. 测试：

    • 启动微服务，正常发送消息，可以正常接收到消息

热点文章 - 实时计算

定时计算 vs 实时计算

思路说明：

思路说明

流程：

用户行为（阅读量，评论，点赞，收藏）发送消息，以 阅读 和 点赞 为例

  1. 在 heima-leadnews-behavior 微服务中集成 kafka 生产者配置，修改nacos，新增内容

    1
    2
    3
    4
    5
    6
    7
    8
    9
    spring:
      application:
        name: leadnews-behavior
      kafka:
        bootstrap-servers: 192.168.88.129:9092
        producer:
          retries: 10
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer

  2. 修改ApLikesBehaviorServiceImpl新增发送消息

  3. 定义消息发送封装类：UpdateArticleMess

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    @Data
    public class UpdateArticleMess {

        /**
         * 修改文章的字段类型
          */
        private UpdateArticleType type;
        /**
         * 文章ID
         */
        private Long articleId;
        /**
         * 修改数据的增量，可为正负
         */
        private Integer add;

        public enum UpdateArticleType{
            COLLECTION,COMMENT,LIKES,VIEWS;
        }
    }

  4. 定义topic常量类

    1
    2
    3
    4
    5
    public class HotArticleConstants {

        public static final String HOT_ARTICLE_SCORE_TOPIC="hot.article.score.topic";
       
    }

  5. ApLikesBehaviorServiceImpl 修改后的代码

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    @Service
    @Transactional
    @Slf4j
    public class ApLikesBehaviorServiceImpl implements ApLikesBehaviorService {

        @Autowired
        private CacheService cacheService;

        @Autowired
        private KafkaTemplate<String,String> kafkaTemplate;

        @Override
        public ResponseResult like(LikesBehaviorDto dto) {

            //1.检查参数
            if (dto == null || dto.getArticleId() == null || checkParam(dto)) {
                return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
            }

            //2.是否登录
            ApUser user = AppThreadLocalUtil.getUser();
            if (user == null) {
                return ResponseResult.errorResult(AppHttpCodeEnum.NEED_LOGIN);
            }

            UpdateArticleMess mess = new UpdateArticleMess();
            mess.setArticleId(dto.getArticleId());
            mess.setType(UpdateArticleMess.UpdateArticleType.LIKES);

            //3.点赞  保存数据
            if (dto.getOperation() == 0) {
                Object obj = cacheService.hGet(BehaviorConstants.LIKE_BEHAVIOR + dto.getArticleId().toString(), user.getId().toString());
                if (obj != null) {
                    return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID, "已点赞");
                }
                // 保存当前key
                log.info("保存当前key:{} ,{}, {}", dto.getArticleId(), user.getId(), dto);
                cacheService.hPut(BehaviorConstants.LIKE_BEHAVIOR + dto.getArticleId().toString(), user.getId().toString(), JSON.toJSONString(dto));
                mess.setAdd(1);
            } else {
                // 删除当前key
                log.info("删除当前key:{}, {}", dto.getArticleId(), user.getId());
                cacheService.hDelete(BehaviorConstants.LIKE_BEHAVIOR + dto.getArticleId().toString(), user.getId().toString());
                mess.setAdd(-1);
            }

            //发送消息，数据聚合
            kafkaTemplate.send(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC,JSON.toJSONString(mess));


            return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);

        }

        /**
         * 检查参数
         *
         * @return
         */
        private boolean checkParam(LikesBehaviorDto dto) {
            if (dto.getType() > 2 || dto.getType() < 0 || dto.getOperation() > 1 || dto.getOperation() < 0) {
                return true;
            }
            return false;
        }
    }

  6. 修改阅读行为的类 ApReadBehaviorServiceImpl 发送消息

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    @Service
    @Transactional
    @Slf4j
    public class ApReadBehaviorServiceImpl implements ApReadBehaviorService {

        @Autowired
        private CacheService cacheService;

        @Autowired
        private KafkaTemplate<String,String> kafkaTemplate;

        @Override
        public ResponseResult readBehavior(ReadBehaviorDto dto) {
            //1.检查参数
            if (dto == null || dto.getArticleId() == null) {
                return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
            }

            //2.是否登录
            ApUser user = AppThreadLocalUtil.getUser();
            if (user == null) {
                return ResponseResult.errorResult(AppHttpCodeEnum.NEED_LOGIN);
            }
            //更新阅读次数
            String readBehaviorJson = (String) cacheService.hGet(BehaviorConstants.READ_BEHAVIOR + dto.getArticleId().toString(), user.getId().toString());
            if (StringUtils.isNotBlank(readBehaviorJson)) {
                ReadBehaviorDto readBehaviorDto = JSON.parseObject(readBehaviorJson, ReadBehaviorDto.class);
                dto.setCount((short) (readBehaviorDto.getCount() + dto.getCount()));
            }
            // 保存当前key
            log.info("保存当前key:{} {} {}", dto.getArticleId(), user.getId(), dto);
            cacheService.hPut(BehaviorConstants.READ_BEHAVIOR + dto.getArticleId().toString(), user.getId().toString(), JSON.toJSONString(dto));

            //发送消息，数据聚合
            UpdateArticleMess mess = new UpdateArticleMess();
            mess.setArticleId(dto.getArticleId());
            mess.setType(UpdateArticleMess.UpdateArticleType.VIEWS);
            mess.setAdd(1);
            kafkaTemplate.send(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC,JSON.toJSONString(mess));
            
            
            return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);

        }
    }

stream聚合操作

  1. 在 leadnews-article 微服务中集成 kafkaStream

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    /**
     * 通过重新注册KafkaStreamsConfiguration对象，设置自定配置参数
     */

    @Setter
    @Getter
    @Configuration
    @EnableKafkaStreams
    @ConfigurationProperties(prefix="kafka")
    public class KafkaStreamConfig {
        private static final int MAX_MESSAGE_SIZE = 16* 1024 * 1024;
        private String hosts;
        private String group;
        @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
        public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
            Map<String, Object> props = new HashMap<>();
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, this.getGroup()+"_stream_aid");
            props.put(StreamsConfig.CLIENT_ID_CONFIG, this.getGroup()+"_stream_cid");
            props.put(StreamsConfig.RETRIES_CONFIG, 10);
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            return new KafkaStreamsConfiguration(props);
        }
    }
    1
    2
    3
    kafka:
      hosts: 192.168.88.129:9092
      group: ${spring.application.name}

  2. 定义实体类，用于聚合之后的分值封装

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    @Data
    public class ArticleVisitStreamMess {
        /**
         * 文章id
         */
        private Long articleId;
        /**
         * 阅读
         */
        private int view;
        /**
         * 收藏
         */
        private int collect;
        /**
         * 评论
         */
        private int comment;
        /**
         * 点赞
         */
        private int like;
    }

  3. 修改常量类：增加常量

    1
    2
    3
    4
    5
    public class HotArticleConstants {

        public static final String HOT_ARTICLE_SCORE_TOPIC="hot.article.score.topic";
        public static final String HOT_ARTICLE_INCR_HANDLE_TOPIC="hot.article.incr.handle.topic";
    }

  4. 定义stream，接收消息并聚合

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    @Configuration
    @Slf4j
    public class HotArticleStreamHandler {

        @Bean
        public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
            // 接收消息
            KStream<String,String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
            
            // 聚合流式处理
            stream.map((key,value)->{
                // 将UpdateArticleMess反序列化
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                
                // 重置消息的 key 和 value
                return new KeyValue<>(mess.getArticleId().toString(), mess.getType().name()+":"+mess.getAdd());
                //  key:1234343434, value: likes:1
            })
                    // 按照文章id进行聚合
                    .groupBy((key,value)->key)
                    // 时间窗口
                    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                    /**
                     * 自行的完成聚合的计算
                     */
                    .aggregate(new Initializer<String>() {
                        /**
                         * 初始方法，返回值是消息的value
                         * @return
                         */
                        @Override
                        public String apply() {
                            return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                        }
                        /**
                         * 真正的聚合操作，返回值是消息的value
                         */
                    }, new Aggregator<String, String, String>() {
                        
                        @Override
                        // aggValue：为初始化方法返回的值
                        public String apply(String key, String value, String aggValue) {
                            if(StringUtils.isBlank(value)){
                                return aggValue;
                            }
                            
                            // 循环操作
                            String[] aggAry = aggValue.split(",");
                            int col = 0, com = 0, lik = 0, vie = 0;
                            for (String agg : aggAry) {
                                String[] split = agg.split(":");
                                
                                /**
                                 * 获得初始值，也是时间窗口内计算之后的值
                                 */
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                    case COLLECTION:
                                        col = Integer.parseInt(split[1]);
                                        break;
                                    case COMMENT:
                                        com = Integer.parseInt(split[1]);
                                        break;
                                    case LIKES:
                                        lik = Integer.parseInt(split[1]);
                                        break;
                                    case VIEWS:
                                        vie = Integer.parseInt(split[1]);
                                        break;
                                }
                            }
                            
                            /**
                             * 累加操作
                             */
                            String[] valAry = value.split(":");
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])){
                                case COLLECTION:
                                    col += Integer.parseInt(valAry[1]);
                                    break;
                                case COMMENT:
                                    com += Integer.parseInt(valAry[1]);
                                    break;
                                case LIKES:
                                    lik += Integer.parseInt(valAry[1]);
                                    break;
                                case VIEWS:
                                    vie += Integer.parseInt(valAry[1]);
                                    break;
                            }

                            String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                            System.out.println("文章的id:" + key);
                            System.out.println("当前时间窗口内的消息处理结果：" + formatStr);
                            return formatStr;
                        }
                    }, Materialized.as("hot-atricle-stream-count-001"))
                    .toStream()
                    .map((key,value)->{
                        return new KeyValue<>(key.key().toString(), formatObj(key.key().toString(), value));
                    })
                    // 发送消息
                    .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

            return stream;


        }

        /**
         * 格式化消息的value数据
         * @param articleId
         * @param value
         * @return
         */
        public String formatObj(String articleId,String value){
            ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
            mess.setArticleId(Long.valueOf(articleId));
            // COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
            String[] valAry = value.split(",");
            for (String val : valAry) {
                String[] split = val.split(":");
                switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                    case COLLECTION:
                        mess.setCollect(Integer.parseInt(split[1]));
                        break;
                    case COMMENT:
                        mess.setComment(Integer.parseInt(split[1]));
                        break;
                    case LIKES:
                        mess.setLike(Integer.parseInt(split[1]));
                        break;
                    case VIEWS:
                        mess.setView(Integer.parseInt(split[1]));
                        break;
                }
            }
            log.info("聚合消息处理之后的结果为:{}",JSON.toJSONString(mess));
            return JSON.toJSONString(mess);
            
        }
    }

重新计算文章的分值，更新到数据库和缓存中

  1. 在 ApArticleService 添加方法，用于更新数据库中的文章分值

    1
    2
    3
    4
    5
    /**
         * 更新文章的分值  同时更新缓存中的热点文章数据
         * @param mess
         */
    public void updateScore(ArticleVisitStreamMess mess);

  2. 实现方法

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    /**
         * 更新文章的分值  同时更新缓存中的热点文章数据
         * @param mess
    */
    @Override
    public void updateScore(ArticleVisitStreamMess mess) {
        // 1. 更新文章的阅读、点赞、收藏、评论的数量
        ApArticle apArticle = updateArticle(mess);
        
        // 2. 计算文章的分值
        Integer score = computeScore(apArticle);
        score = score * 3;

        // 3. 替换当前文章对应频道的热点数据
        replaceDataToRedis(apArticle, score, ArticleConstants.HOT_ARTICLE_FIRST_PAGE + apArticle.getChannelId());

        // 4. 替换推荐对应的热点数据
        replaceDataToRedis(apArticle, score, ArticleConstants.HOT_ARTICLE_FIRST_PAGE + ArticleConstants.DEFAULT_TAG);

    }

    /**
         * 替换数据并且存入到redis
         * @param apArticle
         * @param score
         * @param s
    */
    private void replaceDataToRedis(ApArticle apArticle, Integer score, String s) {
        String articleListStr = cacheService.get(s);
        
        if (StringUtils.isNotBlank(articleListStr)) {
            List<HotArticleVo> hotArticleVoList = JSON.parseArray(articleListStr, HotArticleVo.class);

            boolean flag = true;

            // 如果缓存中存在该文章，只更新分值
            for (HotArticleVo hotArticleVo : hotArticleVoList) {
                if (hotArticleVo.getId().equals(apArticle.getId())) {
                    hotArticleVo.setScore(score);
                    flag = false;
                    break;
                }
            }

            // 如果缓存中不存在，查询缓存中分值最小的一条数据，进行分值的比较，如果当前文章的分值大于缓存中的数据，就替换
            if (flag) {
                if (hotArticleVoList.size() >= 30) {
                    hotArticleVoList = hotArticleVoList.stream().sorted(Comparator.comparing(HotArticleVo::getScore).reversed()).collect(Collectors.toList());
                    HotArticleVo lastHot = hotArticleVoList.get(hotArticleVoList.size() - 1);
                    if (lastHot.getScore() < score) {
                        hotArticleVoList.remove(lastHot);
                        HotArticleVo hot = new HotArticleVo();
                        BeanUtils.copyProperties(apArticle, hot);
                        hot.setScore(score);
                        hotArticleVoList.add(hot);
                    }

                } else {
                    HotArticleVo hot = new HotArticleVo();
                    BeanUtils.copyProperties(apArticle, hot);
                    hot.setScore(score);
                    hotArticleVoList.add(hot);
                }
            }
            
            // 缓存到redis
            hotArticleVoList = hotArticleVoList.stream().sorted(Comparator.comparing(HotArticleVo::getScore).reversed()).collect(Collectors.toList());
            cacheService.set(s, JSON.toJSONString(hotArticleVoList));

        }
    }

    /**
         * 更新文章行为数量
         * @param mess
    */
    private ApArticle updateArticle(ArticleVisitStreamMess mess) {
        
        ApArticle apArticle = getById(mess.getArticleId());
        
        apArticle.setCollection(apArticle.getCollection()==null?0:apArticle.getCollection() + mess.getCollect());
        
        apArticle.setComment(apArticle.getComment()==null?0:apArticle.getComment() + mess.getComment());
        
        apArticle.setLikes(apArticle.getLikes()==null?0:apArticle.getLikes() + mess.getLike());
        
        apArticle.setViews(apArticle.getViews()==null?0:apArticle.getViews() + mess.getView());
        
        updateById(apArticle);
        
        return apArticle;
    }

    /**
         * 计算文章的具体分值
         * @param apArticle
         * @return
    */
    private Integer computeScore(ApArticle apArticle) {
        Integer score = 0;
        
        if(apArticle.getLikes() != null){
            score += apArticle.getLikes() * ArticleConstants.HOT_ARTICLE_LIKE_WEIGHT;
        }
        
        if(apArticle.getViews() != null){
            score += apArticle.getViews();
        }
        
        if(apArticle.getComment() != null){
            score += apArticle.getComment() * ArticleConstants.HOT_ARTICLE_COMMENT_WEIGHT;
        }
        
        if(apArticle.getCollection() != null){
            score += apArticle.getCollection() * ArticleConstants.HOT_ARTICLE_COLLECTION_WEIGHT;
        }

        return score;
    }

  3. 定义监听，接收聚合之后的数据，文章的分值重新进行计算

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    @Component
    @Slf4j
    public class ArticleIncrHandleListener {

        @Autowired
        private ApArticleService apArticleService;

        @KafkaListener(topics = HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC)
        public void onMessage(String mess){
            if(StringUtils.isNotBlank(mess)){
                ArticleVisitStreamMess articleVisitStreamMess = JSON.parseObject(mess, ArticleVisitStreamMess.class);
                apArticleService.updateScore(articleVisitStreamMess);

            }
        }
    }

持续集成

持续集成（ Continuous integration ， 简称 CI ）指的是：频繁地（一天多次）将代码集成到主干

持续集成

持续集成的组成要素

  1. 一个自动构建过程， 从检出代码、 编译构建、 运行测试、 结果记录、 测试统计等都是自动完成的， 无需人工干预
  2. 一个代码存储库，即需要版本控制软件来保障代码的可维护性，同时作为构建过程的素材库，一般使用SVN或Git
  3. 一个持续集成服务器， Jenkins 就是一个配置简单和使用方便的持续集成服务器

持续集成的好处：

  1. 降低风险，由于持续集成不断去构建，编译和测试，可以很早期发现问题，所以修复的代价就少
  2. 对系统健康持续检查，减少发布风险带来的问题
  3. 减少重复性工作
  4. 持续部署，提供可部署单元包
  5. 持续交付可供使用的版本
  6. 增强团队信心

软件开发模式

软件开发生命周期

软件开发生命周期又叫做SDLC（Software Development Life Cycle），它是集合了 计划、开发、测试 和 部署 过程的集合

SDLC

  • 需求分析

    这是生命周期的第一阶段，根据项目需求，团队执行一个可行性计划的分析。项目需求可能是公司内部或者客户提出的。这阶段主要是对信息的收集，也有可能是对现有项目的改善和重新做一个新的项目。还要分析项目的预算多长，可以从哪方面受益及布局，这也是项目创建的目标。

  • 设计

    第二阶段就是设计阶段，系统架构和满意状态（就是要做成什么样子，有什么功能），和创建一个项目计划。计划可以使用图表，布局设计或者文字的方式呈现。

  • 实现

    第三阶段就是实现阶段，项目经理创建和分配工作给开者，开发者根据任务和在设计阶段定义的目标进行开发代码。依据项目的大小和复杂程度，可以需要数月或更长时间才能完成。

  • 测试

    测试人员进行代码测试 ，包括功能测试、代码测试、压力测试等。

  • 进化

    最后进阶段就是对产品不断的进化改进和维护阶段，根据用户的使用情况，可能需要对某功能进行修改，bug修复，功能增加等

软件开发瀑布模型

软件开发瀑布模型

优势 劣势
简单易用和理解 各个阶段的划分完全固定，阶段之间产生大量的文档，极大地增加了工作量。
当前一阶段完成后，您只需要去关注后续阶段。 由于开发模型是线性的，用户只有等到整个过程的末期才能见到开发成果，从而增加了开发风险。
为项目提供了按阶段划分的检查节点 瀑布模型的突出缺点是不适应用户需求的变化。

软件的敏捷开发

软件的敏捷开发

  • 敏捷开发：
    • 敏捷开发（Agile Development） 的 核心是迭代开发（Iterative Development） 与 增量开发（Incremental Development）
  • 迭代开发：
    • 对于大型软件项目，传统的开发方式是采用一个大周期（比如一年）进行开发，整个过程就是一次 “大开发”
    • 迭代开发的方式则不一样，它将开发过程拆分成多个小周期，即一次 “大开发” 变成多次 “小开发”，每次小开发都是同样的流程，所以看上去就好像重复在做同样的步骤
  • 增量开发：
    • 软件的每个版本，都会新增一个用户可以感知的完整功能。也就是说，按照新增功能来划分迭代
  • 软件敏捷开发的好处：
    • 早期交付
    • 降低风险

Jenkins 安装配置

Jenkins 是一款流行的开源持续集成（Continuous Integration）工具，广泛用于项目开发，具有自动化构建、测试和部署等功能

官网： http://jenkins-ci.org/

Jenkins流程

流程：

  1. 采用YUM方式安装

    加入jenkins安装源：

    1
    2
    3
    sudo wget -O /etc/yum.repos.d/jenkins.repo https://pkg.jenkins.io/redhat-stable/jenkins.repo --no-check-certificate

    sudo rpm --import https://pkg.jenkins.io/redhat-stable/jenkins.io.key

    执行yum命令安装：

    1
    yum -y install jenkins

  2. 采用RPM安装包方式

    Jenkins安装包下载地址

    1
    wget https://pkg.jenkins.io/redhat-stable/jenkins-2.190.1-1.1.noarch.rpm

    执行安装：

    1
    rpm -ivh jenkins-2.190.1-1.1.noarch.rpm

  3. 配置：

    修改配置文件：

    1
    vi /etc/sysconfig/jenkins

    修改内容：

    1
    2
    3
    4
    # 修改为对应的目标用户， 这里使用的是root
    $JENKINS_USER="root"
    # 服务监听端口
    JENKINS_PORT="16060"

    目录权限：

    1
    2
    3
    chown -R root:root /var/lib/jenkins
    chown -R root:root /var/cache/jenkins
    chown -R root:root /var/log/jenkins

    重启：

    1
    systemctl restart jenkins

    如果启动失败， 出现错误信息：

    1
    Starting Jenkins bash: /usr/bin/java: No such file or directory

    创建JAVA环境的软链接：

    1
    ln -s /usr/local/jdk/bin/java /usr/bin/java

  4. 管理后台初始化设置

    http://192.168.200.100:16060/

    需要输入管理密码， 在以下位置查看：

    1
    cat /var/lib/jenkins/secrets/initialAdminPassword

    按默认设置，把建议的插件都安装上

    这一步等待时间较长， 安装完成之后， 创建管理员用户：

配置访问地址：

配置完成之后， 会进行重启， 之后可以看到管理后台：

Jenkins 安装 - 插件安装

在实现持续集成之前， 需要确保以下插件安装成功。

  • Maven Integration plugin：Maven 集成管理插件
  • Docker plugin：Docker集成插件
  • GitLab Plugin：GitLab集成插件
  • Publish Over SSH：远程文件发布插件
  • SSH：远程脚本执行插件

JDK 安装（略）

Git 安装配置

  1. yum 安装方式

    1
    yum -y install git

  2. 采用源码包方式安装

    • 安装依赖包

      1
      2
      yum -y install curl-devel expat-devel gettext-devel openssl-devel zlib-devel
      yum -y install gcc perl-ExtUtils-MakeMaker

    • 如果之前有安装旧版本， 先做卸载， 没有安装则忽略

      1
      yum remove git

    • 下载源码包

      1
      2
      3
      cd /usr/local
      wget https://mirrors.edge.kernel.org/pub/software/scm/git/git-1.8.3.1.tar.gz
      tar -xvf git-1.8.3.1.tar.gz

      也可以安装其他版本， 地址：https://mirrors.edge.kernel.org/pub/software/scm/git/

    • 编译安装

      1
      2
      3
      4
      5
      cd git-1.8.3.1
      make prefix=/usr/local/git all
      make prefix=/usr/local/git install
      echo "export PATH=$PATH:/usr/local/git/bin" >> /etc/bashrc
      source /etc/bashrc

  3. 检查git版本

    1
    2
    [root@localhost jenkins]# git version
    git version 1.8.3.1

Maven 安装配置

  1. 下载安装包

    下载地址： https://maven.apache.org/download.cgi

  2. 解压安装包

    1
    2
    cd /usr/local
    unzip -o apache-maven-3.6.1.zip

    上传本地仓库并解压

  3. 配置

    环境变量配置

    1
    vi /etc/profile

    增加：

    1
    2
    export MAVEN_HOME=/usr/local/maven/apache-maven-3.6.1
    export PATH=$PATH:$MAVEN_HOME/bin

    如果权限不够，则需要增加当前目录的权限

    1
    chmod 777 /usr/local/maven/apache-maven-3.6.1/bin/mvn

    修改镜像仓库配置：

    1
    vi /usr/local/maven/apache-maven-3.6.1/conf/settings.xml

    需要把本机的仓库打包上传到服务器上（不上传会自动下载）

    然后指定上传后的仓库配置

Docker 安装配置

  1. 更新软件包版本

    1
    yum -y update

  2. 卸载旧版本

    1
    yum -y remove docker  docker-common docker-selinux docker-engine

  3. 安装软件依赖包

    1
    yum install -y yum-utils device-mapper-persistent-data lvm2

  4. 设置yum源为阿里云

    1
    sudo yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

  5. 安装后查看docker版本

    1
    docker -v

  6. 启动

    设置开机启动：

    1
    systemctl enable docker

    启动docker

    1
    systemctl start docker

Jenkins 工具配置

  1. 进入【系统管理】–> 【全局工具配置】

  2. MAVEN配置全局设置

  3. 指定JDK配置

  4. 指定 MAVEN 目录

  5. 指定 DOCKER 目录

    如果不清楚docker的安装的目录，可以使用whereis docker 命令查看docker的安装的目录

后端项目部署

多环境切换

在项目开发部署的过程中，一般都会有三套项目环境

  • Development：开发环境
  • Production：生产环境
  • Test：测试环境

微服务中多环境配置

  1. 在微服务中的bootstrap.yml中新增配置（以user微服务为例）

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    server:
      port: 51801
    spring:
      application:
        name: leadnews-user
      cloud:
        nacos:
          discovery:
            server-addr: 192.168.88.129:8848
          config:
            server-addr: 192.168.88.129:8848
            file-extension: yml
      profiles:
        active: dev
        # prod
        # test

  2. 在nacos的配置中心中新增各个环境的配置文件（以user微服务为例）

    创建开发环境的相关配置文件

    • 也可以配置 leadnews-user-test 的配置

整体思路

整体思路

  • 注意：192.168.200.100 与 192.168.200.130 必须使用 NAT 这个网卡，必须在同一个网段，是可以互相通信的，可以使用 ping 命令来检查

服务集成Docker配置

目标：部署的每一个微服务都是先创建 docker镜像 后创建对应容器启动

方法：

  1. 本地微服务打包以后上传到服务器，编写 Dockerfile 文件完成
  2. 使用 dockerfile-maven-plugin 插件，可以直接把微服务创建为镜像使用（更省事）

步骤：

  1. 每个微服务都引入该依赖

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <executions>
                <execution>
                    <goals>
                        <goal>repackage</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.7.0</version>
            <configuration>
                <source>${java.version}</source>
                <target>${java.version}</target>
            </configuration>
        </plugin>
        <plugin>
            <groupId>com.spotify</groupId>
            <artifactId>dockerfile-maven-plugin</artifactId>
            <version>1.3.6</version>
            <configuration>
                <repository>${docker.image}/${project.artifactId}</repository>
                <buildArgs>
                    <JAR_FILE>target/${project.build.finalName}.jar</JAR_FILE>
                </buildArgs>
            </configuration>
        </plugin>
    </plugins>

    说明

  2. 创建Dockerfile文件

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    # 设置JAVA版本
    FROM java:8
    # 指定存储卷, 任何向/tmp写入的信息都不会记录到容器存储层
    VOLUME /tmp
    # 拷贝运行JAR包
    ARG JAR_FILE
    COPY ${JAR_FILE} app.jar
    # 设置JVM运行参数， 这里限定下内存大小，减少开销
    ENV JAVA_OPTS="\
    -server \
    -Xms256m \
    -Xmx512m \
    -XX:MetaspaceSize=256m \
    -XX:MaxMetaspaceSize=512m"
    # 空参数，方便创建容器时传参
    ENV PARAMS=""
    # 入口点， 执行JAVA运行命令
    ENTRYPOINT ["sh","-c","java -jar $JAVA_OPTS /app.jar $PARAMS"]

  3. 通过的maven的命令可以把微服务打成镜像

    1
    dockerfile:build

  4. 后面会结合着jenkins一起使用