## （1）如何定位慢查询

出现原因：

聚合查询

多表查询

表数据过大

深度分页查询

### 1.开源工具

Arthas Prometheus Skywalking

![image.png](/static/img/b24b69523d480b09365adbb8c00516dc.image.webp)

### 2.Mysql自带的慢日志

![image.png](/static/img/f60e1d8a3767f85b168b89b83fe9b091.image.webp)

面试话术

![image.png](/static/img/0c57893423ba4b40645827e01174c171.image.webp)

![image.png](/static/img/02de93ed8f20c83497e2b90e5866dc93.image.webp)

## （2）如何优化SQL

聚合查询 新增临时表

多表查询 优化SQL语句结构                     SQL执行计划(找到慢的原因)

表数据过大查询 添加索引

深度分页查询

![image.png](/static/img/82eacd3eee7c58a02e0f16cd5a2dfa32.image.webp)

![image.png](/static/img/e99c1acf2a72f9638a17bfe49774807b.image.webp)

![image.png](/static/img/1f843779fc4d03efa34c884ea186225a.image.webp)

面试话术

![image.png](/static/img/f931a0187af5af92f01792f734ffa2c4.image.webp)

## （3）索引

话术

![image.png](/static/img/c62ef96d8bea7130c1a024816a4e6d2c.image.webp)

![image.png](/static/img/9544d7e5a043821808b94d5e53681a7c.image.webp)

![image.png](/static/img/96837575a24345465d5a5a0ae070b58e.image.webp)

## （4）聚集索引，二级索引(聚簇索引，非聚簇索引)回表查询

### 1.聚集索引

![image.png](/static/img/fbf6cedd683ba1d59523c651623e22a0.image.webp)

### 2.回表查询

![image.png](/static/img/93e5dd9abf5cc86de2bb6f8ca0beedfa.image.webp)

面试话术

![image.png](/static/img/e5920646b57574c2e29385cb60db3811.image.webp)

![image.png](/static/img/12c9965addf8c6139707df91942722e7.image.webp)

## （5）覆盖索引，超大分页优化

### 1.覆盖索引

![image.png](/static/img/8d24298c83d075b5558c06aab4b9136f.image.webp)

![image.png](/static/img/f5bbdce603cbe8b6c8f56935500ddd1f.image.webp)

覆盖索引没有回表查询

面试话术

![image.png](/static/img/2a619ea56af53c04012aae0d7af5a53f.image.webp)

![image.png](/static/img/775231c87abc119f3a2f51a4822d384f.image.webp)
### 2.Mysql超大分页

![image.png](/static/img/5e42e7854c609559bd1ee1d6a0387e7a.image.webp)

![image.png](/static/img/73fe114ac9d9470ef4a71ef68b9f88bd.image.webp)

![image.png](/static/img/2db9a0197f211a267e44fba9dd85732a.image.webp)

## （6）创建索引的原则

![image.png](/static/img/142f0937182ae29d5a2d95323fc684af.image.webp)

面试话术

![image.png](/static/img/681256e06e9c0d48d6002bdc6dd35ecb.image.webp)

![image.png](/static/img/0728897b251b6a2dc4e0ec165a5ea4e6.image.webp)

## （7）索引失效

**失效情况**
 
 ### 1.违反最短前缀法则
 
![image.png](/static/img/f842ffea93eda8c758af52c2eaed95e7.image.webp)

![image.png](/static/img/bc858fee39f39b42151f324518d83886.image.webp)

### 2.范围查询右边的列不能使用索引

![image.png](/static/img/d4c99c3526a9f2dab5888bc9bba9cdaf.image.webp)

### 3.不要在索引列上进行运算操作

![image.png](/static/img/350539394f13945c3e3d9b0ad95129c6.image.webp)

### 4.字符串不加单引号，造成索引失效

![image.png](/static/img/68b9e2bf37973c65feb2fc6630909615.image.webp)

### 5.以%开头的like模糊查询，索引失效。如果仅仅是尾部匹配模糊查询索引不会失效。

![image.png](/static/img/aa179db68df1e011e375633d095e1858.image.webp)

面试话术

![image.png](/static/img/edc38d72e2db808739d3d7bbee1bb0b1.image.webp)

![image.png](/static/img/10707343269e24b5c5541bab2716c4e6.image.webp)

 




Mysql篇

# 1.redis常见面试题

## （1）redis缓存穿透





![image.png](/static/img/78044e91fa212a3424e1e34e5931d2ae.image.webp)

使用不存在的id发送请求 不直接访问redis而是直接访问mysql

![image.png](/static/img/ff40159c8d19aaf6c4482bbd9c51ad26.image.webp)

**解决方案:**

### 1.缓存空数据,查询结果为空仍把这个空结果进行缓存
优点：简单
缺点：消耗内存，可能发生不一致的问题

### 2.布隆过滤器

![image.png](/static/img/268aa004c978781cd4b70adce9b5b9ad.image.webp)

优点：内存占用较小，没有多余的key
缺点：实现较为复杂，存在误判

创建一个bitmap 初始全为空 预热布隆过滤器对传入id进行多次hash计算 由计算结果bitmap的对应位置该为1 如果传入idhash计算后在bitmap中不全为1就拦截

![image.png](/static/img/25ce76d07a6f9c26ffb03b08dd9cf4c0.image.webp)

## （2）缓存击穿

![image.png](/static/img/65a1f4c6d3b04624a34354089d82dce7.image.webp)

![image.png](/static/img/192be4d88f55dc01131c3a73ffce8302.image.webp)

**解决方案:**

### 1.互斥锁

![image.png](/static/img/8b1bedda11d6f01f5ef351bbe59fa803.image.webp)

强一致 性能差

![image.png](/static/img/0329965fbfd1bbe68708bb6922e7d476.image.webp)

### 2.逻辑过期

![image.png](/static/img/b75e594e7956b08ca8c1ab065dd5faa8.image.webp)

高可用 性能优

![image.png](/static/img/edb4362b6446db8cd4e2e935b0cffed6.image.webp)

##  （3）缓存雪崩

![image.png](/static/img/961bbed3743bac4287d75c0517e39561.image.webp)

![image.png](/static/img/c449e6a26eb5256a67679e88b2477311.image.webp)

**解决方案**

![image.png](/static/img/e269057dce9cc531109f29834d4fafd3.image.webp)

![image.png](/static/img/59528bd685662234e6257219e0ff7b90.image.webp)

## （4）双写一致性

![image.png](/static/img/ff09fd0e0281c1ca2bc9003a672e008f.image.webp)

强一致

![image.png](/static/img/3ead054c08a3bc4474993c8a517a390a.image.webp)

读写锁

![image.png](/static/img/4440007991641b651962b5398f30a26d.image.webp)

延时双删

![image.png](/static/img/6714028a1c52350ee149320dac20e834.image.webp)

总结

![image.png](/static/img/614da7fad2ea3f813ab24f942bc82da7.image.webp)

## （5）redis数据持久化

### 1.RDB

![image.png](/static/img/d06bc50fc2a9fee697d59a0970d0d9af.image.webp)

执行原理

![image.png](/static/img/7c062d617595b705eb9eef88b8e102c5.image.webp)

### 2.AOF

![image.png](/static/img/73202da1de69fd5bd16c064e28bdcc20.image.webp)

RDB和AOF对比

![image.png](/static/img/5308f6af0e6f1a6d6f64006ca67eda46.image.webp)

![image.png](/static/img/0f57477042d1356642a57a622cf87efd.image.webp)

## （6）数据过期策略

### 1.惰性删除

设置该key过期后，我们不去管他，当需要该key时我们再检查其是否过期，如果过期，我们就删除它，反之则返回key。

![image.png](/static/img/15ee1ce590afe4ee1ab31d35ee371401.image.webp)

### 2.定期删除

每隔一段时间我们就对一些key进行检查，删除里面过期的key

![image.png](/static/img/8b6b499e85c753e0dc8c7e6dd8872d8f.image.webp)

![image.png](/static/img/1d1f3136191b7d58cf7145ac10250b59.image.webp)

![image.png](/static/img/8a5161b893c6cd9ca64f7b44bafeb738.image.webp)

## （7）redis数据淘汰策略

![image.png](/static/img/d7a8716844b7b9d959942bd712cd2538.image.webp)

数据淘汰策略推荐

![image.png](/static/img/7cc91663a4da79a5b6ebcd58d41b7e7a.image.webp)

其他问题

![image.png](/static/img/ccac6e8f31f0eab23112e607327b4bc2.image.webp)

![image.png](/static/img/b6cce873ae3abaf2a255852e0715d62a.image.webp)

## （8）redis分布式锁

### 1.setnx

![image.png](/static/img/bfdf6287071ca17a6d78f91719bdc11c.image.webp)

给锁续期

### 2.redisson实现的分布式锁

执行流程

![image.png](/static/img/6fa4461e67be4995a903064fe193c6ee.image.webp)

![image.png](/static/img/374ad521853ab1c66d9ab95945fe7751.image.webp)

同一线程中可重入

**主从一致性**
 
 当主节点redis在获取锁时宕机会导致其他进程也获取锁成功
 
![image.png](/static/img/20ce392bd795e6d8bd746a51739ba82b.image.webp)

红锁

![image.png](/static/img/a97c7fa5ef6e22f9165040e474a5d4e7.image.webp)

面试话术

![image.png](/static/img/db11e8cc2e84904555f5081d331217c7.image.webp)

![image.png](/static/img/a0d94e1791c504bce780a90a689c1cb4.image.webp)

![image.png](/static/img/b0a96400c5e2ae2b0071e87f781bfe7d.image.webp)

## （9）redis集群方案

**主从复制**

**哨兵模式**

**分片集群**

### 1.主从复制

全量同步：

![image.png](/static/img/925311aebda91cc53789ca82c7254f81.image.webp)

增量同步：

![image.png](/static/img/5d3336c9c29c2eff16afce1d23e6e11a.image.webp)

面试话术

![image.png](/static/img/3f87d72b71418e779e082ebd883af830.image.webp)

![image.png](/static/img/b85d3710296d75890a5010759bc7929e.image.webp)

### 2.哨兵模式

**哨兵作用: 监控 自动故障恢复 通知**

![image.png](/static/img/2afa91b8d7fe9ed87a3482a1717afac1.image.webp)

服务状态监控

![image.png](/static/img/42a6f69761580a884aa2d510afa29b2d.image.webp)

脑裂问题

![image.png](/static/img/5bb7c880b21f21e9da2e31f72f517b03.image.webp)

面试话术

![image.png](/static/img/efebfddc9c6d5cdfa8301d2e0fb20e91.image.webp)

![image.png](/static/img/b5e011af8dd5a2a0c315fe6890c9e053.image.webp)

### 3.分片集群

![image.png](/static/img/f364eda1dbbe2dcc25b77ff16314348b.image.webp)

数据读写

![image.png](/static/img/1ba0db13cbeb2744038a12ce65cfae3b.image.webp)

面试话术

![image.png](/static/img/80ae5c8de917c97d16492f16db881e19.image.webp)

![image.png](/static/img/51b7e26f37410cf672338a57f03b9ca4.image.webp)

## （9）redis其他面试问题

### 1.redis是单线程的为什么还这么快

reids是纯内存操作，执行速度非常快

采用单线程，避免不必要的上下文切换可竞争条件，多线程还要考虑线程安全问题

使用i/o多路复用模型，非阻塞io

![image.png](/static/img/f724a415f8f3b69a469802103243dfc8.image.webp)

### 2.解释一下i/o多路复用模型

![image.png](/static/img/15adf0a435f96228b4b2aff8a8620d04.image.webp)

![image.png](/static/img/08958a8b380aaab617772aeeeafc17f0.image.webp)

![image.png](/static/img/123e1cc3070b8a1e42e866d1c9ed7352.image.webp)

Redis篇

# 1.mongodb实现搜索记录




![image.png](/static/img/0d3706623b1b1bf8a77f3bad25477869.image.webp)

docker启动mongodb

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


java集成mongodb

![image.png](/static/img/2c63d6fb65201c8b7979f65c507884a3.image.webp)

实现用户搜索记录保存功能

```java
package com.heima.search.service.impl;

import com.heima.search.service.ApUserSearchService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import pojos.ApUserSearch;

import java.util.Date;
import java.util.List;


@Service
@Slf4j
public class ApUserSearchServiceImpl implements ApUserSearchService {


    @Autowired
    private MongoTemplate mongoTemplate;


    /**
     * 保存用户搜索历史记录
     *
     * @param keyward
     * @param userId
     */
    @Override
    @Async
    public void insert(String keyward, Integer userId) {
        // 1.查询当前用户的搜索关键词记录
        Query query = Query.query(Criteria.where("userId").is(userId).and("keyword").is(keyward));
        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.setKeyword(keyward);
        apUserSearch.setUserId(userId);
        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 lastApUserSearch = apUserSearchList.get(apUserSearchList.size() - 1);
            mongoTemplate.findAndReplace(Query.query(Criteria.where("id").is(lastApUserSearch.getId())),apUserSearch);
        }
    }
}

```

在调用es搜索时异步调用

```java
        if (apUser !=null && dto.getFromIndex() == 0){
            // 异步调用保存搜索记录
            apUserSearchService.insert(dto.getSearchWords(), apUser.getId());
        }
```

# 2.实现搜索记录查询

![image.png](/static/img/50c36b6ec1d52c5c55117aeee24fc25e.image.webp)

编写查询历史记录的实现类

```java
 /**
     * 查询用户搜索记录
     *
     * @return
     */
    @Override
    public ResponseResult findUserSearch() {
        // 获取当前用户
        ApUser user = AppThreadLocalUtil.getApUser();
        if (user==null){
            return ResponseResult.errorResult(AppHttpCodeEnum.NEED_LOGIN);
        }
        // 根据用户查询记录 降序排序
        List<ApUserSearch> apUserSearchList = mongoTemplate.find(Query.query(Criteria.where("userId").is(user.getId())).with(Sort.by(Sort.Direction.DESC, "createdTime")), ApUserSearch.class);
        return ResponseResult.okResult(apUserSearchList);
    }
```
编写接口层

```java
    @PostMapping("/load")
    public ResponseResult findUserSearch(){
        return apUserSearchService.findUserSearch();
    }
```

# 3.实现搜索记录删除功能

![image.png](/static/img/b68153e9e96cecafb8161b7f3df92378.image.webp)

代码实现

```java
    /**
     * 删除用户搜索记录
     *
     * @param dto
     */
    @Override
    public ResponseResult delUserSearch(HistorySearchDto dto) {
        // 1.检查参数
        if (dto == null || dto.getId() == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        // 2.判断是否登录
        ApUser user = AppThreadLocalUtil.getApUser();
        if (user == null) {
            return ResponseResult.errorResult(AppHttpCodeEnum.NEED_LOGIN);
        }
        // 3.删除搜索记录
        mongoTemplate.remove(Query.query(Criteria.where("userId").is(user.getId()).and("id").is(dto.getId())), ApUserSearch.class);

        return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
    }
```
# 4.关键词联想

代码实现

```java
    /**
     * 实现关键词联想功能
     *
     * @param dto
     * @return
     */
    @Override
    public ResponseResult search(UserSearchDto dto) {
        // 1.检查参数
        if (dto == null || StringUtils.isBlank(dto.getSearchWords())) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        // 2.分页检查
        if (dto.getPageSize()>20){
            dto.setPageSize(20);
        }
        // 3.执行查询,模糊查询
        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);
    }
```

# 5.xxl入门

定时任务

![image.png](/static/img/1899176fafcbb0672fcdeea9b310217a.image.webp)

定时任务框架

![image.png](/static/img/67a1558d07a7356c4c59e0a81cffb716.image.webp)

分布式任务调度

![image.png](/static/img/c2d0a06da6257b58a2175a2162f5c77b.image.webp)

 创建mysql容器
 
```shell
docker run -p 3306:3306 --name mysql57 \
-v /opt/mysql/conf:/etc/mysql \
-v /opt/mysql/logs:/var/log/mysql \
-v /opt/mysql/data:/var/lib/mysql \
-e MYSQL_ROOT_PASSWORD=root \
-d mysql:5.7。25
```

创建xxl-job-admin


```shell
docker run -e PARAMS="--spring.datasource.url=jdbc:mysql://192.168.42.10: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
```

快速入门

![image.png](/static/img/eaa079d0b3a72779ecd29a5cac2ed6a4.image.webp)

![image.png](/static/img/328c58c12e7ba01fc18bd5410f634b15.image.webp)

config

```java
package com.heima.config;

import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * xxl-job config
 *
 * @author xuxueli 2017-04-28
 */
@Configuration
public class XxlJobConfig {
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

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

//    @Value("${xxl.job.accessToken}")
//    private String accessToken;

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

    @Value("${xxl.job.executor.port}")
    private int port;
//
//    @Value("${xxl.job.executor.logpath}")
//    private String logPath;
//
//    @Value("${xxl.job.executor.logretentiondays}")
//    private int logRetentionDays;


    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
        logger.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppname(appname);
//        xxlJobSpringExecutor.setAddress(address);
//        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
//        xxlJobSpringExecutor.setAccessToken(accessToken);
//        xxlJobSpringExecutor.setLogPath(logPath);
//        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);

        return xxlJobSpringExecutor;
    }

    /**
     * 针对多网卡、容器内部署等情况，可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP；
     *
     *      1、引入依赖：
     *          <dependency>
     *             <groupId>org.springframework.cloud</groupId>
     *             <artifactId>spring-cloud-commons</artifactId>
     *             <version>${version}</version>
     *         </dependency>
     *
     *      2、配置文件，或者容器启动变量
     *          spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'
     *
     *      3、获取IP
     *          String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
     */


}
```

job


```java
package com.heima.job;


import com.xxl.job.core.handler.annotation.XxlJob;
import org.springframework.stereotype.Component;

@Component
public class HelloJob {

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

```
# 6.xxl-job分片广播

![image.png](/static/img/7fc8462f4365bdef7ad47d2585cc3db2.image.webp)

![image.png](/static/img/e158e5b87941500505c2b03d2cb7fa67.image.webp)

# 7.热点文章计算

![image.png](/static/img/f90ac3c8cb1e3ef223da74853e20e4d0.image.webp)

查询前5天的文章

![image.png](/static/img/f9bd3cf3231c1375e6969cffd7ffe5e6.image.webp)

计算分值

![image.png](/static/img/ae8c206a768e22e789439f71b4453167.image.webp)

为每一个频道缓存30条文章

![image.png](/static/img/57791ffc98719f53f50c7d46c77f0cb8.image.webp)

代码实现

```java
package com.heima.article.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.heima.apis.wemedia.IWemediaClient;
import com.heima.article.mapper.ApArticleMapper;
import com.heima.article.service.HotArticleService;
import com.heima.common.redis.CacheService;
import com.heima.model.article.pojos.ApArticle;
import com.heima.model.article.vos.HotArticleVo;
import com.heima.model.common.dtos.ResponseResult;
import com.heima.model.wemedia.pojos.WmChannel;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;


@Service
@Slf4j
@Transactional
public class HotArticleServiceImpl implements HotArticleService {


    @Autowired
    private IWemediaClient iWemediaClient;


    @Autowired
    private ApArticleMapper apArticleMapper;


    @Autowired
    private CacheService cacheService;

    /**
     * 计算热点文章
     */
    @Override
    public void computeHotArticle() {
        // 查询前5天的文章数据
        Date dateParam = DateTime.now().minusDays(5).toDate();
        List<ApArticle> articleListByLast5Days = apArticleMapper.findArticleListByLast5Days(dateParam);
        // 计算文章的分值
        List<HotArticleVo> hotArticleVoList = computeHotArticleScore(articleListByLast5Days);
        // 为每个频道缓存30条分值高的文章
        cacheTagToRedis(hotArticleVoList);
    }




    /**
     * 为每个频道缓存30条文章
     * @param hotArticleVoList
     */
    private void cacheTagToRedis(List<HotArticleVo> hotArticleVoList) {
       // 每个频道缓存前30条数据
        ResponseResult channels = iWemediaClient.getChannels();
        if (channels.getCode().equals(200)){
            String channelStr = JSON.toJSONString(channels.getData());
            List<WmChannel> wmChannels = JSON.parseArray(channelStr, 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
                    sortAndCache(hotArticleVos, "hot_article_first_page" + wmChannel.getId());
                }
            }
        }

        // 设置推荐数据
        sortAndCache(hotArticleVoList, "__all__");
    }

    /**
     * 排序并且缓存数据
     * @param hotArticleVos
     * @param wmChannel
     */
    private void sortAndCache(List<HotArticleVo> hotArticleVos, String wmChannel) {
        hotArticleVos = hotArticleVos.stream().sorted(Comparator.comparing(HotArticleVo::getScore).reversed()).collect(Collectors.toList());
        if (hotArticleVos.size() > 30) {
            hotArticleVos = hotArticleVos.subList(0, 30);
        }
        cacheService.set(wmChannel, JSON.toJSONString(hotArticleVos));
    }


    /**
     * 计算文章分值返回热点文章列表
     * @param articleListByLast5Days
     * @return
     */
    private List<HotArticleVo> computeHotArticleScore(List<ApArticle> articleListByLast5Days) {
        List<HotArticleVo> hotArticleVoList = new ArrayList<>();

        for (ApArticle apArticle : articleListByLast5Days) {
            HotArticleVo hotArticleVo = new HotArticleVo();
            BeanUtils.copyProperties(apArticle,hotArticleVo);
            Integer score = computeScore(apArticle);
            hotArticleVo.setScore(score);
            hotArticleVoList.add(hotArticleVo);
        }
        return hotArticleVoList;
    }


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

```
# 8.流式计算

kafkaStream

![image.png](/static/img/6d972062090213e44800698f4aa67559.image.webp)

KStream

![image.png](/static/img/ad71da346202b9b02c3ddc93a44d6a79.image.webp)

入门案例

![image.png](/static/img/5d5fbfc65cda689b90774de8a50477be.image.webp)


![image.png](/static/img/46814542e7d78a8342ca48f2d9eebc63.image.webp)

黑马头条(4)

# 1.发布文章自动添加/消费任务

发布任务




```java
    /**
     * 添加文章审核任务到延迟队列中
     *
     * @param id
     * @param publicTime
     */
    @Override
    @Async // 开启异步调用
    public void addNewToTask(Integer id, Date publicTime) {
        log.info("添加任务到延迟队列中-----begin");
        Task task = new Task();
        task.setExecuteTime(publicTime.getTime());
        task.setTaskType(TaskTypeEnum.NEWS_SCAN_TIME.getTaskType());
        task.setPriority(TaskTypeEnum.NEWS_SCAN_TIME.getPriority());
        WmNews wmNews = new WmNews();
        wmNews.setId(id);
        task.setParameters(ProtostuffUtil.serialize(wmNews));
        scheduleClient.addTask(task);
        log.info("添加任务到延迟队列中-----end");
    }
```

消费任务

```java
    /**
     * 消费任务审核文章
     */
    @Scheduled(fixedRate = 1000)  // 每秒一次
    @Override
    public void scanNewsByTask() {
        log.info("消费任务-----审核文章");
        ResponseResult responseResult = scheduleClient.pull(TaskTypeEnum.NEWS_SCAN_TIME.getTaskType(), TaskTypeEnum.NEWS_SCAN_TIME.getPriority());
        if (responseResult.getCode().equals(200) && responseResult.getData() != null){
            String jsonString = JSON.toJSONString(responseResult.getData());
            Task task = JSON.parseObject(jsonString, Task.class);
            WmNews wmNews = ProtostuffUtil.deserialize(task.getParameters(), WmNews.class);
            wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
        }
    }
```

# 2.Kafka

![image.png](/static/img/0833dcf69d133913353828dc6c05909f.image.webp)

安装

![image.png](/static/img/687d68cccd571902a2d1dd0c1f631142.image.webp)

![image.png](/static/img/3c2d67849430cee4a094d0fcd5c3b2c0.image.webp)

```shell
docker run -d --name kafka \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.42.10 \
--env KAFKA_ZOOKEEPER_CONNECT=127.0.0.1:2181 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.42.10:9092 \
--env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
--env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
--net=host wurstmeister/kafka:2.12-2.3.1
```

Kafka快速入门

producer

```java
package com.heima.kafka.sample;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

/**
 * 生产者
 */
public class ProducerQuickStart {


    public static void main(String[] args) {
        // kafka连接配置
        Properties prop = new Properties();
        // kafka连接地址
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.42.10:9092");
        // kafka 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");
        // 创建kafka生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String,String>(prop);
        // 发送消息
        /**
         * 第一个参数 topic
         * 第二 key
         * 第三 value
         */
        ProducerRecord<String,String> message = new ProducerRecord<String,String>("topic-first","key-001","hello kafka");

        producer.send(message);
        // 关闭消息通道

        // 必须关闭否则发送不成功
        producer.close();
    }
}

```

consumer

```java
package com.heima.kafka.sample;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class ConsumerQuickStart {

    public static void main(String[] args) {

        // kafka连接配置
        Properties prop = new Properties();
        // kafka连接地址
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.42.10:9092");
        // 指定消费者组
        prop.put(ConsumerConfig.GROUP_ID_CONFIG,"group2");
        // 指定反序列化
        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");

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

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

        // 拉取消息
        while (true) {
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> stringStringConsumerRecord : poll) {
                System.out.println(stringStringConsumerRecord.key());
                System.out.println(stringStringConsumerRecord.value());
            }
        }

    }
}

```

kafka分区机制

![image.png](/static/img/b54ecec2d842be7a1d57d3bfc80dd877.image.webp)

![image.png](/static/img/d2b0e90945d2098a6faeeecd298a418d.image.webp)

![image.png](/static/img/71072c9ae60bd35b5c68ca2c7f294da1.image.webp)

kafka高可用机制

![image.png](/static/img/d4fa8f0c343e96d3fe554cc3fae09601.image.webp)

备份机制

![image.png](/static/img/d128a75311e3f0ddcb3add67fd387cc5.image.webp)

生产者发送类型(同步发送

![image.png](/static/img/12438e59486c6065304fe6fd9f20ffd7.image.webp)

![image.png](/static/img/b89e688b7dbe41a3ef1042c6598d2c5e.image.webp)

ack确认机制

![image.png](/static/img/29007e35649d870a5f224ca2d11d4fca.image.webp)

消费者详情

消息有序性

![image.png](/static/img/fe30d7e1d34092e89ee1ad280bbeeb69.image.webp)

# 3.springboot集成kafka

![image.png](/static/img/37b73321b8d7f38792951f67babf4220.image.webp)

![image.png](/static/img/143c3bdb9b4e2897fb61ae4d75dac307.image.webp)

# 3.自媒体文章上下架

![image.png](/static/img/577ec80701abb829cd8ca5b43fdf387c.image.webp)

编写生产者向kafka发送消息

```java
    /**
     * 文章的上下架
     *
     * @param dto
     * @return
     */
    @Override
    public ResponseResult downOrUp(WmNewsDto dto) {
        // 1.检查参数
        if (dto.getId() == null){
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        // 2.查询文章
        WmNews wmNews = getById(dto.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.修改文章
        if (dto.getEnable()!=null&&dto.getEnable() > -1&&dto.getEnable()<2 ){
            update(Wrappers.<WmNews>lambdaUpdate().set(WmNews::getEnable,dto.getEnable())
                    .eq(WmNews::getId,dto.getId()));
        }
        // 通知article下架 kafka生产者发送消息
        if (wmNews.getArticleId()!=null){
            Map<String,Object> map = new HashMap<>();
            map.put("articleId",wmNews.getArticleId());
            map.put("enable",dto.getEnable());
            kafkaTemplate.send(WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC,JSON.toJSONString(map));
        }
        return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
    }
```

app端消费kafka的消息上下架文章

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

# 4.elasticsearch实现App端搜索

![image.png](/static/img/c4c4bb5d91b319ad4439a8a97744def3.image.webp)

搭建elasticsearch环境

![image.png](/static/img/c866740559314ca7665dea925ed96118.image.webp)

`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
`
创建索引库

![image.png](/static/img/68487e6ced167ec601b1437ee562cb35.image.webp)

```
{
    "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"
            }
        }
    }
}
```

导入leadnews-search到项目中 nacos中添加配置
```yaml
spring:
  autoconfigure:
    exclude: org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration
elasticsearch:
  host: 192.168.42.10
  port: 9200
```

es查询接口

```java
 /**
     * es文章分页检索
     *
     * @param dto
     * @return
     */
    @Override
    public ResponseResult search(UserSearchDto dto) throws IOException {
        // 1.检查参数
        if (dto == null|| StringUtils.isBlank(dto.getSearchWords())) {
            return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
        }
        // 2.设置查询条件
           //  关键字分词之后查询
        SearchRequest searchRequest = new SearchRequest("app_info_article");
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

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

        QueryStringQueryBuilder queryStringQueryBuilder = QueryBuilders.queryStringQuery(dto.getSearchWords()).field("title").field("content")
                .defaultOperator(Operator.OR);

        boolQueryBuilder.must(queryStringQueryBuilder);

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

        boolQueryBuilder.filter(rangeQueryBuilder);
        //     按照发布时间
        searchSourceBuilder.sort("publishTime", SortOrder.DESC);
        //     关键字高亮
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.field("title");
        highlightBuilder.preTags("<font style='color: red; font-size: inherit;'>");
        highlightBuilder.postTags("</font>");
        searchSourceBuilder.highlighter(highlightBuilder);
        // 设置分页
        searchSourceBuilder.from(0)
                .size(dto.getPageSize());


        searchSourceBuilder.query(boolQueryBuilder);

        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        // 3.结果封装返回
        SearchHit[] hits = searchResponse.getHits().getHits();
        List<Map> resultList = new ArrayList<>();
        for (SearchHit hit : hits) {
            String JSONString = hit.getSourceAsString();
            Map map = JSON.parseObject(JSONString, Map.class);
            // 处理高亮
            if (hit.getHighlightFields()!=null && hit.getHighlightFields().size()>0) {
                Map<String, HighlightField> highlightFields = hit.getHighlightFields();
                Text[] titles = highlightFields.get("title").getFragments();
                String title = StringUtils.join(titles);
                // 高亮标题
                map.put("h_title", title);
            }else{
                // 原始标题
                map.put("h_title", map.get("title"));
            }
            resultList.add(map);
        }
        return ResponseResult.okResult(resultList);
    }
```

创建索引接口