devbox
秋刀鱼在做梦
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

《黑马头条》 内容安全 自动审核 feign 延迟任务精准发布 kafka

作者：秋刀鱼在做梦 | 2024-08-16 12:03:34

赞

踩

黑马头条

目录

《黑马头条》SpringBoot+SpringCloud+ Nacos等企业级微服务架构项目_黑马头条项目_软工菜鸡的博客-CSDN博客

04自媒体文章-自动审核

1)自媒体文章自动审核流程

2)内容安全第三方接口

2.1)概述

2.2)准备工作

2.3)文本内容审核接口

2.4)图片审核接口

2.5)项目集成

3)app端文章保存接口

3.1)表结构说明

3.2)分布式id

分布式id-技术选型

3.3)思路分析

3.4)feign接口

4)自媒体文章自动审核功能实现

4.1)表结构说明

4.2)实现

4.3)单元测试

4.4)feign远程接口调用方式

4.5)服务降级处理

5)发布文章提交审核集成

5.1)同步调用与异步调用

5.2)Springboot集成异步线程调用

6)文章审核功能-综合测试

6.1)服务启动列表

6.2)测试情况列表

7)新需求-自管理敏感词

7.1)需求分析

7.2)敏感词-过滤

7.3)DFA实现原理

7.4)自管理敏感词集成到文章审核中

测试了一下 源码不能检测 标题的敏感词汇；加了个这： wmNews.getTitle()+

8)新需求-图片识别文字审核敏感词

8.1)需求分析

8.2)图片文字识别

8 .3)Tess4j案例

8.4)管理敏感词和图片文字识别集成到文章审核

9)文章详情-静态文件生成

9.1)思路分析

9.2)实现步骤

05延迟任务精准发布文章

1)文章定时发布

2)延迟任务概述

2.1)什么是延迟任务

2.2)技术对比

2.2.1)DelayQueue

2.2.2)RabbitMQ实现延迟任务

2.2.3)redis实现

3)redis实现延迟任务

锐评：完全为了学list zset而编出来的场景，实际工作中延迟队列要设计成这样只能说太蠢了

4)延迟任务服务实现

4.1)搭建heima-leadnews-schedule模块

4.2)数据库准备

乐观锁/悲观锁

4.3)安装redis

4.4)项目集成redis

4.5)添加任务

4.6)取消任务

4.7)消费任务

4.8)未来数据定时刷新

4.8.1)reids key值匹配

4.8.2)reids管道

4.8.3)未来数据定时刷新-功能完成

4.9)分布式锁解决集群下的方法抢占执行

4.9.1)问题描述

4.9.2)分布式锁

4.9.3)redis分布式锁

4.9.4)在工具类CacheService中添加方法

4.10)数据库同步到redis

5)延迟队列解决精准时间发布文章

5.1)延迟队列服务提供对外接口

5.2)发布文章集成添加延迟队列接口

序列化工具对比

5.3)消费任务进行审核文章

06kafka及异步通知文章上下架

1)自媒体文章上下架

2)kafka概述

消息中间件对比-选择建议

kafka介绍-名词解释

3)kafka安装配置

4)kafka入门

分区机制—topic剖析

5)kafka高可用设计

5.1)集群

5.2)备份机制(Replication）

6)kafka生产者详解

6.1)发送类型

6.2)参数详解

ack确认机制

retries 重试次数

消息压缩

7)kafka消费者详解

7.1)消费者组

7.2)消息有序性

7.3)提交和偏移量

1.提交当前偏移量（同步提交）

2.异步提交

3.同步和异步组合提交

8)springboot集成kafka

8.1)入门

8.2)传递消息为对象

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

9.1)需求分析

9.2)流程说明

9.3)接口定义

9.4)自媒体文章上下架-功能实现

9.5)消息通知article端文章上下架


04自媒体文章-自动审核

1)自媒体文章自动审核流程

1 自媒体端发布文章后，开始审核文章
2 审核的主要是审核文章的 内容（文本内容和图片）
3 借助 第三方提供的接口审核文本
4 借助第三方提供的接口审核图片，由于图片存储到minIO中，需要先下载才能审核
5 如果审核失败，则需要修改自媒体文章的状态，status:2 审核失败 status:3 转到人工审核
6 如果审核成功，则需要在文章微服务中创建app端需要的文章

2)内容安全第三方接口

2.1)概述

内容安全是识别服务，支持对图片、视频、文本、语音等对象多样化场景检测，有效降低内容违规风险

目前很多平台都支持内容检测，如阿里云、腾讯云、百度AI、网易云等国内大型互联网公司都对外提供了API。

按照性能和收费来看，黑马头条项目使用的就是阿里云的内容安全接口，使用到了图片和文本的审核。

阿里云收费标准：https://www.aliyun.com/price/product/?spm=a2c4g.11186623.2.10.4146401eg5oeu8#/lvwang/detail

2.2)准备工作

您在使用内容检测API之前，需要先注册阿里云账号，添加Access Key并签约云盾内容安全。

操作步骤

  1. 前往阿里云官网注册账号。如果已有注册账号，请跳过此步骤。

进入阿里云首页后，如果没有阿里云的账户需要先进行注册，才可以进行登录。由于注册较为简单，课程和讲义不在进行体现（注册可以使用多种方式，如淘宝账号、支付宝账号、微博账号等...）。

需要实名认证和活体认证。

  1. 打开云盾内容安全产品试用页面，单击立即开通，正式开通服务。

内容安全控制台

  1. 在AccessKey管理页面管理您的AccessKeyID和AccessKeySecret。

管理自己的AccessKey,可以新建和删除AccessKey

查看自己的AccessKey，

AccessKey默认是隐藏的，第一次申请的时候可以保存AccessKey，点击显示，通过验证手机号后也可以查看

2.3)文本内容审核接口

文本垃圾内容检测：如何调用文本检测接口进行文本内容审核_内容安全-阿里云帮助中心

文本垃圾内容Java SDK: 如何使用JavaSDK文本反垃圾接口_内容安全-阿里云帮助中心

2.4)图片审核接口

图片垃圾内容检测：调用图片同步检测接口/green/image/scan审核图片内容_内容安全-阿里云帮助中心

图片垃圾内容Java SDK: 如何使用JavaSDK接口检测图片是否包含风险内容_内容安全-阿里云帮助中心

2.5)项目集成

①：拷贝资料文件夹中的类到common模块下面，并添加到自动配置

包括了GreenImageScan和GreenTextScan及对应的工具类

添加到自动配置中

②： accessKeyId和secret（需自己申请）

在heima-leadnews-wemedia中的nacos配置中心添加以下配置：

  1. aliyun:
  2.  accessKeyId: ...
  3.  secret: ...
  4. #aliyun.scenes=porn,terrorism,ad,qrcode,live,logo
  5.  scenes: terrorism

③：在自媒体微服务中测试类中注入审核文本和图片的bean进行测试

  1.  
  2. package com.heima.wemedia;
  3.  
  4. import java.util.Arrays;
  5. import java.util.Map;
  6.  
  7. @SpringBootTest(classes = WemediaApplication.class)
  8. @RunWith(SpringRunner.class)
  9. public class AliyunTest {
  10.  
  11.     @Autowired
  12.     private GreenTextScan greenTextScan;
  13.  
  14.     @Autowired
  15.     private GreenImageScan greenImageScan;
  16.  
  17.     @Autowired
  18.     private FileStorageService fileStorageService;
  19.  
  20.     @Test
  21.     public void testScanText() throws Exception {
  22.         Map map = greenTextScan.greeTextScan("我是一个好人,冰毒");
  23.         System.out.println(map);
  24.     }
  25.  
  26.     @Test
  27.     public void testScanImage() throws Exception {
  28.         byte[] bytes = fileStorageService.downLoadFile("http://192.168.200.130:9000/leadnews/2021/04/26/ef3cbe458db249f7bd6fb4339e593e55.jpg");
  29.         Map map = greenImageScan.imageScan(Arrays.asList(bytes));
  30.         System.out.println(map);
  31.     }
  32. }

我用的是 阿里云 云安全 增强版1小时，没审核出效果为null；估计是阿里 改接口了；

图片审核页报错

  1. java.lang.RuntimeException: upload file fail.
  2.  
  3.     at com.heima.common.aliyun.util.ClientUploader.uploadBytes(ClientUploader.java:129)
  4.     at com.heima.common.aliyun.GreenImageScan.imageScan(GreenImageScan.java:71)
  5.     at com.heima.wemedia.test.AliyunTest.testScanImage(AliyunTest.java:51)
  6.     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  7.     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  8.     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  9.     at java.lang.reflect.Method.invoke(Method.java:498)
  10.     at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
  11.     at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
  12.     at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
  13.     at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
  14.     at org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:74)
  15.     at org.springframework.test.context.junit4.statements.RunAfterTestExecutionC

3)app端文章保存接口

3.1)表结构说明
3.2)分布式id

随着业务的增长，文章表可能要占用很大的物理存储空间，为了解决该问题，后期使用数据库分片技术。将一个数据库进行拆分，通过数据库中间件连接。如果数据库中该表选用ID自增策略，则可能产生重复的ID，此时应该使用分布式ID生成策略来生成ID。

分布式id-技术选型

snowflake是Twitter开源的分布式ID生成算法，结果是一个long型的ID。

其核心思想是：使用41bit作为毫秒数，10bit作为机器的ID（5个bit是数据中心，5个bit的机器ID)（最多32个机房*32台机器(也可以自己设)），12bit作为毫秒内的流水号（意味着每个节点在每毫秒可以产生 4096 个 ID），最后还有一个符号位，永远是0（1为负数）

文章端相关的表都使用雪花算法生成id,包括ap_article、 ap_article_config、 ap_article_content

mybatis-plus已经集成了雪花算法，完成以下两步即可在项目中集成雪花算法

第一：在实体类中的id上加入如下配置，指定类型为id_worker

  1.  
  2. @TableId(value = "id",type = IdType.ID_WORKER)
  3. private Long id;

第二：在application.yml文件中配置数据中心id和机器id

  1. mybatis-plus:
  2.   mapper-locations: classpath*:mapper/*.xml
  3.   # 设置别名包扫描路径，通过该属性可以给包中的类注册别名
  4.   type-aliases-package: com.heima.model.article.pojos
  5.   global-config:
  6.     datacenter-id: 1
  7.     workerId: 1

datacenter-id:数据中心id(取值范围：0-31) ；workerId:机器id(取值范围：0-31)

3.3)思路分析

在文章审核成功以后需要在app的article库中新增文章数据

1.保存文章信息 ap_article

2.保存文章配置信息 ap_article_config

3.保存文章内容 ap_article_content

实现思路：

3.4)feign接口

ArticleDto

  1.  
  2. package com.heima.model.article.dtos;
  3.  
  4. import com.heima.model.article.pojos.ApArticle;
  5. import lombok.Data;
  6.  
  7. @Data
  8. public class ArticleDto  extends ApArticle {
  9.     /**
  10.      * 文章内容
  11.      */
  12.     private String content;
  13. }

功能实现：

①：在heima-leadnews-feign-api中新增接口

第一：线导入feign的依赖

  1. <dependency>
  2.     <groupId>org.springframework.cloud</groupId>
  3.     <artifactId>spring-cloud-starter-openfeign</artifactId>
  4. </dependency>

第二：定义文章端的接口

  1.  
  2. package com.heima.apis.article;
  3. import org.springframework.web.bind.annotation.RequestBody;
  4.  
  5. @FeignClient(value = "leadnews-article")
  6. public interface IArticleClient {
  7.  
  8.     @PostMapping("/api/v1/article/save")
  9.     public ResponseResult saveArticle(@RequestBody ArticleDto dto) ;
  10. }

②：在heima-leadnews-article中实现该方法

  1.  
  2. package com.heima.article.feign;
  3. import java.io.IOException;
  4.  
  5. @RestController
  6. public class ArticleClient implements IArticleClient {
  7.  
  8.     @Autowired
  9.     private ApArticleService apArticleService;
  10.  
  11.     @Override
  12.     @PostMapping("/api/v1/article/save")
  13.     public ResponseResult saveArticle(@RequestBody ArticleDto dto) {
  14.         return apArticleService.saveArticle(dto);
  15.     }
  16. }

③：拷贝mapper

在资料文件夹中拷贝ApArticleConfigMapper类到mapper文件夹中

同时，修改ApArticleConfig类，添加如下构造函数

  1.  
  2. package com.heima.model.article.pojos;
  3.  
  4. import java.io.Serializable;
  5.  
  6. /**
  7.  * <p>
  8.  * APP已发布文章配置表
  9.  * </p>
  10.  *
  11.  * @author itheima
  12.  */
  13.  
  14. @Data
  15. @NoArgsConstructor
  16. @TableName("ap_article_config")
  17. public class ApArticleConfig implements Serializable {
  18.  
  19.  
  20.     public ApArticleConfig(Long articleId){
  21.         this.articleId = articleId;
  22.         this.isComment = true;
  23.         this.isForward = true;
  24.         this.isDelete = false;
  25.         this.isDown = false;
  26.     }
  27.  
  28.     @TableId(value = "id",type = IdType.ID_WORKER)
  29.     private Long id;
  30.  
  31.     /**
  32.      * 文章id
  33.      */
  34.     @TableField("article_id")
  35.     private Long articleId;
  36.  
  37.     /**
  38.      * 是否可评论
  39.      * true: 可以评论   1
  40.      * false: 不可评论  0
  41.      */
  42.     @TableField("is_comment")
  43.     private Boolean isComment;
  44.  
  45.     /**
  46.      * 是否转发
  47.      * true: 可以转发   1
  48.      * false: 不可转发  0
  49.      */
  50.     @TableField("is_forward")
  51.     private Boolean isForward;
  52.  
  53.     /**
  54.      * 是否下架
  55.      * true: 下架   1
  56.      * false: 没有下架  0
  57.      */
  58.     @TableField("is_down")
  59.     private Boolean isDown;
  60.  
  61.     /**
  62.      * 是否已删除
  63.      * true: 删除   1
  64.      * false: 没有删除  0
  65.      */
  66.     @TableField("is_delete")
  67.     private Boolean isDelete;
  68. }

④：在ApArticleService中新增方法

  1.  
  2. /**
  3.      * 保存app端相关文章
  4.      * @param dto
  5.      * @return
  6.      */
  7. ResponseResult saveArticle(ArticleDto dto) ;

实现类：

  1.  
  2. @Autowired
  3. private ApArticleConfigMapper apArticleConfigMapper;
  4.  
  5. @Autowired
  6. private ApArticleContentMapper apArticleContentMapper;
  7.  
  8. /**
  9.      * 保存app端相关文章
  10.      * @param dto
  11.      * @return
  12.      */
  13. @Override
  14. public ResponseResult saveArticle(ArticleDto dto) {
  15.     //1.检查参数
  16.     if(dto == null){
  17.         return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
  18.     }
  19.  
  20.     ApArticle apArticle = new ApArticle();
  21.     BeanUtils.copyProperties(dto,apArticle);
  22.  
  23.     //2.判断是否存在id
  24.     if(dto.getId() == null){
  25.         //2.1 不存在id  保存  文章  文章配置  文章内容
  26.  
  27.         //保存文章
  28.         save(apArticle);
  29.  
  30.         //保存配置
  31.         ApArticleConfig apArticleConfig = new ApArticleConfig(apArticle.getId());
  32.         apArticleConfigMapper.insert(apArticleConfig);
  33.  
  34.         //保存 文章内容
  35.         ApArticleContent apArticleContent = new ApArticleContent();
  36.         apArticleContent.setArticleId(apArticle.getId());
  37.         apArticleContent.setContent(dto.getContent());
  38.         apArticleContentMapper.insert(apArticleContent);
  39.  
  40.     }else {
  41.         //2.2 存在id   修改  文章  文章内容
  42.  
  43.         //修改  文章
  44.         updateById(apArticle);
  45.  
  46.         //修改文章内容
  47.         ApArticleContent apArticleContent = apArticleContentMapper.selectOne(Wrappers.<ApArticleContent>lambdaQuery().eq(ApArticleContent::getArticleId, dto.getId()));
  48.         apArticleContent.setContent(dto.getContent());
  49.         apArticleContentMapper.updateById(apArticleContent);
  50.     }
  51.  
  52.  
  53.     //3.结果返回  文章的id
  54.     return ResponseResult.okResult(apArticle.getId());
  55. }

⑤：测试

编写junit单元测试，或使用postman进行测试

http://localhost:51802/api/v1/article/save

  1. {
  2. 	"id":这个id要去数据库自己找 ,
  3.     "title":"黑马头条项目背景22222222222222",
  4.     "authoId":1102,
  5.     "layout":1,
  6.     "labels":"黑马头条",
  7.     "publishTime":"2028-03-14T11:35:49.000Z",
  8.     "images": "http://192.168.200.130:9000/leadnews/2021/04/26/5ddbdb5c68094ce393b08a47860da275.jpg",
  9.     "content":"22222222222222222黑马头条项目背景,黑马头条项目背景,黑马头条项目背景,黑马头条项目背景，黑马头条项目背景"
  10. }

4)自媒体文章自动审核功能实现

4.1)表结构说明

wm_news 自媒体文章表

status字段：0 草稿 1 待审核 2 审核失败 3 人工审核 4 人工审核通过 8 审核通过（待发布） 9 已发布

4.2)实现

在heima-leadnews-wemedia中的service新增接口

  1.  
  2. package com.heima.wemedia.service;
  3. public interface WmNewsAutoScanService {
  4.  
  5.     /**
  6.      * 自媒体文章审核
  7.      * @param id  自媒体文章id
  8.      */
  9.     public void autoScanWmNews(Integer id);
  10. }

实现类：

  1.  
  2. package com.heima.wemedia.service.impl;
  3. import java.util.*;
  4. import java.util.stream.Collectors;
  5.  
  6. @Service
  7. @Slf4j
  8. @Transactional
  9. public class WmNewsAutoScanServiceImpl implements WmNewsAutoScanService {
  10.  
  11.     @Autowired
  12.     private WmNewsMapper wmNewsMapper;
  13.  
  14.     /**
  15.      * 自媒体文章审核
  16.      *
  17.      * @param id 自媒体文章id
  18.      */
  19.     @Override
  20.     public void autoScanWmNews(Integer id) {
  21.         //1.查询自媒体文章
  22.         WmNews wmNews = wmNewsMapper.selectById(id);
  23.         if(wmNews == null){
  24.             throw new RuntimeException("WmNewsAutoScanServiceImpl-文章不存在");
  25.         }
  26.  
  27.         if(wmNews.getStatus().equals(WmNews.Status.SUBMIT.getCode())){
  28.             //从内容中提取纯文本内容和图片
  29.             Map<String,Object> textAndImages = handleTextAndImages(wmNews);
  30.  
  31.             //2.审核文本内容  阿里云接口
  32.             boolean isTextScan = handleTextScan((String) textAndImages.get("content"),wmNews);
  33.             if(!isTextScan)return;
  34.  
  35.             //3.审核图片  阿里云接口
  36.             boolean isImageScan =  handleImageScan((List<String>) textAndImages.get("images"),wmNews);
  37.             if(!isImageScan)return;
  38.  
  39.             //4.审核成功，保存app端的相关的文章数据
  40.             ResponseResult responseResult = saveAppArticle(wmNews);
  41.             if(!responseResult.getCode().equals(200)){
  42.                 throw new RuntimeException("WmNewsAutoScanServiceImpl-文章审核，保存app端相关文章数据失败");
  43.             }
  44.             //回填article_id
  45.             wmNews.setArticleId((Long) responseResult.getData());
  46.             updateWmNews(wmNews,(short) 9,"审核成功");
  47.  
  48.         }
  49.     }
  50.  
  51.     @Autowired
  52.     private IArticleClient articleClient;
  53.  
  54.     @Autowired
  55.     private WmChannelMapper wmChannelMapper;
  56.  
  57.     @Autowired
  58.     private WmUserMapper wmUserMapper;
  59.  
  60.     /**
  61.      * 保存app端相关的文章数据
  62.      * @param wmNews
  63.      */
  64.     private ResponseResult saveAppArticle(WmNews wmNews) {
  65.  
  66.         ArticleDto dto = new ArticleDto();
  67.         //属性的拷贝
  68.         BeanUtils.copyProperties(wmNews,dto);
  69.         //文章的布局
  70.         dto.setLayout(wmNews.getType());
  71.         //频道
  72.         WmChannel wmChannel = wmChannelMapper.selectById(wmNews.getChannelId());
  73.         if(wmChannel != null){
  74.             dto.setChannelName(wmChannel.getName());
  75.         }
  76.  
  77.         //作者
  78.         dto.setAuthorId(wmNews.getUserId().longValue());
  79.         WmUser wmUser = wmUserMapper.selectById(wmNews.getUserId());
  80.         if(wmUser != null){
  81.             dto.setAuthorName(wmUser.getName());
  82.         }
  83.  
  84.         //设置文章id
  85.         if(wmNews.getArticleId() != null){
  86.             dto.setId(wmNews.getArticleId());
  87.         }
  88.         dto.setCreatedTime(new Date());
  89.  
  90.         ResponseResult responseResult = articleClient.saveArticle(dto);
  91.         return responseResult;
  92.     }
  93.  
  94.  
  95.     @Autowired
  96.     private FileStorageService fileStorageService;
  97.     @Autowired
  98.     private GreenImageScan greenImageScan;
  99.  
  100.     /**
  101.      * 审核图片
  102.      * @param images
  103.      * @param wmNews
  104.      * @return
  105.      */
  106.     private boolean handleImageScan(List<String> images, WmNews wmNews) {
  107.  
  108.         boolean flag = true;
  109.  
  110.         if(images == null || images.size() == 0){
  111.             return flag;
  112.         }
  113.  
  114.         //下载图片 minIO
  115.         //图片去重
  116.         images = images.stream().distinct().collect(Collectors.toList());
  117.  
  118.         List<byte[]> imageList = new ArrayList<>();
  119.  
  120.         for (String image : images) {
  121.             byte[] bytes = fileStorageService.downLoadFile(image);
  122.             imageList.add(bytes);
  123.         }
  124.  
  125.  
  126.         //审核图片
  127.         try {
  128.             Map map = greenImageScan.imageScan(imageList);
  129.             if(map != null){
  130.                 //审核失败
  131.                 if(map.get("suggestion").equals("block")){
  132.                     flag = false;
  133.                     updateWmNews(wmNews, (short) 2, "当前文章中存在违规内容");
  134.                 }
  135.  
  136.                 //不确定信息  需要人工审核
  137.                 if(map.get("suggestion").equals("review")){
  138.                     flag = false;
  139.                     updateWmNews(wmNews, (short) 3, "当前文章中存在不确定内容");
  140.                 }
  141.             }
  142.  
  143.         } catch (Exception e) {
  144.             flag = false;
  145.             e.printStackTrace();
  146.         }
  147.         return flag;
  148.     }
  149.  
  150.     @Autowired
  151.     private GreenTextScan greenTextScan;
  152.  
  153.     /**
  154.      * 审核纯文本内容
  155.      * @param content
  156.      * @param wmNews
  157.      * @return
  158.      */
  159.     private boolean handleTextScan(String content, WmNews wmNews) {
  160.  
  161.         boolean flag = true;
  162.  
  163.         if((wmNews.getTitle()+"-"+content).length() == 0){
  164.             return flag;
  165.         }
  166.  
  167.         try {
  168.             Map map = greenTextScan.greeTextScan((wmNews.getTitle()+"-"+content));
  169.             if(map != null){
  170.                 //审核失败
  171.                 if(map.get("suggestion").equals("block")){
  172.                     flag = false;
  173.                     updateWmNews(wmNews, (short) 2, "当前文章中存在违规内容");
  174.                 }
  175.  
  176.                 //不确定信息  需要人工审核
  177.                 if(map.get("suggestion").equals("review")){
  178.                     flag = false;
  179.                     updateWmNews(wmNews, (short) 3, "当前文章中存在不确定内容");
  180.                 }
  181.             }
  182.         } catch (Exception e) {
  183.             flag = false;
  184.             e.printStackTrace();
  185.         }
  186.  
  187.         return flag;
  188.  
  189.     }
  190.  
  191.     /**
  192.      * 修改文章内容
  193.      * @param wmNews
  194.      * @param status
  195.      * @param reason
  196.      */
  197.     private void updateWmNews(WmNews wmNews, short status, String reason) {
  198.         wmNews.setStatus(status);
  199.         wmNews.setReason(reason);
  200.         wmNewsMapper.updateById(wmNews);
  201.     }
  202.  
  203.     /**
  204.      * 1。从自媒体文章的内容中提取文本和图片
  205.      * 2.提取文章的封面图片
  206.      * @param wmNews
  207.      * @return
  208.      */
  209.     private Map<String, Object> handleTextAndImages(WmNews wmNews) {
  210.  
  211.         //存储纯文本内容
  212.         StringBuilder stringBuilder = new StringBuilder();
  213.  
  214.         List<String> images = new ArrayList<>();
  215.  
  216.         //1。从自媒体文章的内容中提取文本和图片
  217.         if(StringUtils.isNotBlank(wmNews.getContent())){
  218.             List<Map> maps = JSONArray.parseArray(wmNews.getContent(), Map.class);
  219.             for (Map map : maps) {
  220.                 if (map.get("type").equals("text")){
  221.                     stringBuilder.append(map.get("value"));
  222.                 }
  223.  
  224.                 if (map.get("type").equals("image")){
  225.                     images.add((String) map.get("value"));
  226.                 }
  227.             }
  228.         }
  229.         //2.提取文章的封面图片
  230.         if(StringUtils.isNotBlank(wmNews.getImages())){
  231.             String[] split = wmNews.getImages().split(",");
  232.             images.addAll(Arrays.asList(split));
  233.         }
  234.  
  235.         Map<String, Object> resultMap = new HashMap<>();
  236.         resultMap.put("content",stringBuilder.toString());
  237.         resultMap.put("images",images);
  238.         return resultMap;
  239.  
  240.     }
  241. }
4.3)单元测试
  1.  
  2. package com.heima.wemedia.service;
  3.  
  4. import com.heima.wemedia.WemediaApplication;
  5. import org.junit.Test;
  6. import org.junit.runner.RunWith;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.boot.test.context.SpringBootTest;
  9. import org.springframework.test.context.junit4.SpringRunner;
  10.  
  11. import static org.junit.Assert.*;
  12.  
  13.  
  14. @SpringBootTest(classes = WemediaApplication.class)
  15. @RunWith(SpringRunner.class)
  16. public class WmNewsAutoScanServiceTest {
  17.  
  18.     @Autowired
  19.     private WmNewsAutoScanService wmNewsAutoScanService;
  20.  
  21.     @Test
  22.     public void autoScanWmNews() {
  23.  
  24.         wmNewsAutoScanService.autoScanWmNews(6238);
  25.     }
  26. }
4.4)feign远程接口调用方式

在heima-leadnews-wemedia服务中已经依赖了heima-leadnews-feign-apis工程，只需要在自媒体的引导类中开启feign的远程调用即可

注解为：@EnableFeignClients(basePackages = "com.heima.apis") 需要指向apis这个包

4.5)服务降级处理

  • 服务降级是服务自我保护的一种方式，或者保护下游服务的一种方式，用于确保服务不会受请求突增影响变得不可用，确保服务不会崩溃

  • 服务降级虽然会导致请求失败，但是不会导致阻塞。

实现步骤：

①：在heima-leadnews-feign-api编写降级逻辑

  1.  
  2. package com.heima.apis.article.fallback;
  3. import org.springframework.stereotype.Component;
  4.  
  5. /**
  6.  * feign失败配置
  7.  * @author itheima
  8.  */
  9. @Component
  10. public class IArticleClientFallback implements IArticleClient {
  11.     @Override
  12.     public ResponseResult saveArticle(ArticleDto dto)  {
  13.         return ResponseResult.errorResult(AppHttpCodeEnum.SERVER_ERROR,"获取数据失败");
  14.     }
  15. }

在自媒体微服务中添加类，扫描降级代码类的包

  1.  
  2. package com.heima.wemedia.config;
  3. import org.springframework.context.annotation.ComponentScan;
  4. import org.springframework.context.annotation.Configuration;
  5.  
  6. @Configuration
  7. @ComponentScan("com.heima.apis.article.fallback")
  8. public class InitConfig {
  9. }

②：远程接口中指向降级代码

  1.  
  2. package com.heima.apis.article;
  3. import org.springframework.web.bind.annotation.RequestBody;
  4. @FeignClient(value = "leadnews-article",fallback = IArticleClientFallback.class)
  5. public interface IArticleClient {
  6.  
  7.     @PostMapping("/api/v1/article/save")
  8.     public ResponseResult saveArticle(@RequestBody ArticleDto dto);
  9. }

③：客户端开启降级heima-leadnews-wemedia

在wemedia的nacos配置中心里添加如下内容，开启服务降级，也可以指定服务响应的超时的时间

  1. feign:
  2.   # 开启feign对hystrix熔断降级的支持
  3.   hystrix:
  4.     enabled: true
  5.   # 修改调用超时时间
  6.   client:
  7.     config:
  8.       default:
  9.         connectTimeout: 2000
  10.         readTimeout: 2000

④：测试

在ApArticleServiceImpl类中saveArticle方法添加代码

  1.  
  2. try {
  3.     Thread.sleep(3000);
  4. } catch (InterruptedException e) {
  5.     e.printStackTrace();
  6. }

在自媒体端进行审核测试，会出现服务降级的现象

5)发布文章提交审核集成

5.1)同步调用与异步调用

同步：就是在发出一个调用时，在没有得到结果之前， 该调用就不返回（实时处理）

异步：调用在发出之后，这个调用就直接返回了，没有返回结果（分时处理）

异步线程的方式审核文章

5.2)Springboot集成异步线程调用

①：在自动审核的方法上加上@Async注解（标明要异步调用）

  1.  
  2. @Override
  3. @Async  //标明当前方法是一个异步方法
  4. public void autoScanWmNews(Integer id) {
  5. 	//代码略
  6. }

②：在文章发布成功后调用审核的方法

  1.  
  2. @Autowired
  3. private WmNewsAutoScanService wmNewsAutoScanService;
  4.  
  5. /**
  6.  * 发布修改文章或保存为草稿
  7.  * @param dto
  8.  * @return
  9.  */
  10. @Override
  11. public ResponseResult submitNews(WmNewsDto dto) {
  12.  
  13.     //代码略
  14.  
  15.     //审核文章
  16.     wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
  17.  
  18.     return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
  19.  
  20. }

③：在自媒体引导类中使用@EnableAsync注解开启异步调用

  1.  
  2. @SpringBootApplication
  3. @EnableDiscoveryClient
  4. @MapperScan("com.heima.wemedia.mapper")
  5. @EnableFeignClients(basePackages = "com.heima.apis")
  6. @EnableAsync  //开启异步调用
  7. public class WemediaApplication {
  8.  
  9.     public static void main(String[] args) {
  10.         SpringApplication.run(WemediaApplication.class,args);
  11.     }
  12.  
  13.     @Bean
  14.     public MybatisPlusInterceptor mybatisPlusInterceptor() {
  15.         MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
  16.         interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.MYSQL));
  17.         return interceptor;
  18.     }
  19. }

6)文章审核功能-综合测试

6.1)服务启动列表

1，nacos服务端

2，article微服务

3，wemedia微服务

4，启动wemedia网关微服务

5，启动前端系统wemedia

6.2)测试情况列表

1，自媒体前端发布一篇正常的文章

审核成功后，app端的article相关数据是否可以正常保存，自媒体文章状态和app端文章id是否回显

2，自媒体前端发布一篇包含敏感词的文章

正常是审核失败， wm_news表中的状态是否改变，成功和失败原因正常保存

3，自媒体前端发布一篇包含敏感图片的文章

正常是审核失败， wm_news表中的状态是否改变，成功和失败原因正常保存

7)新需求-自管理敏感词

7.1)需求分析

文章审核功能已经交付了，文章也能正常发布审核。突然，产品经理过来说要开会。

会议的内容核心有以下内容：

  • 文章审核不能过滤一些敏感词：

私人侦探、针孔摄象、信用卡提现、广告代理、代开发票、刻章办、出售答案、小额贷款…

需要完成的功能：

需要自己维护一套敏感词，在文章审核的时候，需要验证文章是否包含这些敏感词

7.2)敏感词-过滤

技术选型

方案

说明

数据库模糊查询

效率太低

String.indexOf("")查找

数据库量大的话也是比较慢

全文检索

分词再匹配

DFA算法

确定有穷自动机(一种数据结构)

7.3)DFA实现原理

DFA全称为：Deterministic Finite Automaton,即确定有穷自动机。

存储：一次性的把所有的敏感词存储到了多个map中，就是下图表示这种结构

敏感词：冰毒、大麻、大坏蛋

检索的过程

7.4)自管理敏感词集成到文章审核中

①：创建敏感词表，导入资料中wm_sensitive到leadnews_wemedia库中

  1.  
  2. package com.heima.model.wemedia.pojos;
  3. import java.io.Serializable;
  4. import java.util.Date;
  5.  
  6. /**
  7.  * <p>
  8.  * 敏感词信息表
  9.  * </p>
  10.  *
  11.  * @author itheima
  12.  */
  13. @Data
  14. @TableName("wm_sensitive")
  15. public class WmSensitive implements Serializable {
  16.  
  17.     private static final long serialVersionUID = 1L;
  18.  
  19.     /**
  20.      * 主键
  21.      */
  22.     @TableId(value = "id", type = IdType.AUTO)
  23.     private Integer id;
  24.  
  25.     /**
  26.      * 敏感词
  27.      */
  28.     @TableField("sensitives")
  29.     private String sensitives;
  30.  
  31.     /**
  32.      * 创建时间
  33.      */
  34.     @TableField("created_time")
  35.     private Date createdTime;
  36.  
  37. }

②：拷贝对应的wm_sensitive的mapper到项目中

  1.  
  2. package com.heima.wemedia.mapper;
  3. import com.baomidou.mybatisplus.core.mapper.BaseMapper;
  4. import com.heima.model.wemedia.pojos.WmSensitive;
  5. import org.apache.ibatis.annotations.Mapper;
  6.  
  7. @Mapper
  8. public interface WmSensitiveMapper extends BaseMapper<WmSensitive> {
  9. }

③：在文章审核的代码中添加自管理敏感词审核

第一：在WmNewsAutoScanServiceImpl中的autoScanWmNews方法上添加如下代码

  1.  
  2. //从内容中提取纯文本内容和图片
  3. //.....省略
  4.  
  5. //自管理的敏感词过滤
  6. boolean isSensitive = handleSensitiveScan((String) textAndImages.get("content"), wmNews);
  7. if(!isSensitive) return;
  8.  
  9. //2.审核文本内容  阿里云接口
  10. //.....省略
测试了一下 源码不能检测 标题的敏感词汇；加了个这： wmNews.getTitle()+

//自管理的敏感词过滤

boolean isSensitive = handleSensitiveScan(

wmNews.getTitle()+textAndImages.get("content"), wmNews);

新增自管理敏感词审核代码

  1.  
  2. @Autowired
  3. private WmSensitiveMapper wmSensitiveMapper;
  4.  
  5. /**
  6.      * 自管理的敏感词审核
  7.      * @param content
  8.      * @param wmNews
  9.      * @return
  10.      */
  11. private boolean handleSensitiveScan(String content, WmNews wmNews) {
  12.  
  13.     boolean flag = true;
  14.  
  15.     //获取所有的敏感词
  16.     List<WmSensitive> wmSensitives = wmSensitiveMapper.selectList(Wrappers.<WmSensitive>lambdaQuery().select(WmSensitive::getSensitives));
  17.     List<String> sensitiveList = wmSensitives.stream().map(WmSensitive::getSensitives).collect(Collectors.toList());
  18.  
  19.     //初始化敏感词库
  20.     SensitiveWordUtil.initMap(sensitiveList);
  21.  
  22.     //查看文章中是否包含敏感词
  23.     Map<String, Integer> map = SensitiveWordUtil.matchWords(content);
  24.     if(map.size() >0){
  25.         updateWmNews(wmNews,(short) 2,"当前文章中存在违规内容"+map);
  26.         flag = false;
  27.     }
  28.  
  29.     return flag;
  30. }

8)新需求-图片识别文字审核敏感词

8.1)需求分析

产品经理召集开会，文章审核功能已经交付了，文章也能正常发布审核。对于上次提出的自管理敏感词也很满意，这次会议核心的内容如下：

  • 文章中包含的图片要识别文字，过滤掉图片文字的敏感词

8.2)图片文字识别

什么是OCR?

OCR (Optical Character Recognition，光学字符识别）是指电子设备（例如扫描仪或数码相机）检查纸上打印的字符，通过检测暗、亮的模式确定其形状，然后用字符识别方法将形状翻译成计算机文字的过程

方案

说明

百度OCR

收费

Tesseract-OCR

Google维护的开源OCR引擎，支持Java，Python等语言调用

Tess4J

封装了Tesseract-OCR ，支持Java调用

8 .3)Tess4j案例

①：创建项目导入tess4j对应的依赖

  1. <dependency>
  2.     <groupId>net.sourceforge.tess4j</groupId>
  3.     <artifactId>tess4j</artifactId>
  4.     <version>4.1.1</version>
  5. </dependency>

②：导入中文字体库， 把资料中的tessdata文件夹拷贝到自己的工作空间下

③：编写测试类进行测试

  1.  
  2. package com.heima.tess4j;
  3.  
  4. import net.sourceforge.tess4j.ITesseract;
  5. import net.sourceforge.tess4j.Tesseract;
  6.  
  7. import java.io.File;
  8.  
  9. public class Application {
  10.  
  11.     public static void main(String[] args) {
  12.         try {
  13.             //获取本地图片
  14.             File file = new File("D:\\26.png");
  15.             //创建Tesseract对象
  16.             ITesseract tesseract = new Tesseract();
  17.             //设置字体库路径
  18.             tesseract.setDatapath("D:\\workspace\\tessdata");
  19.             //中文识别
  20.             tesseract.setLanguage("chi_sim");
  21.             //执行ocr识别
  22.             String result = tesseract.doOCR(file);
  23.             //替换回车和tal键  使结果为一行
  24.             result = result.replaceAll("\\r|\\n","-").replaceAll(" ","");
  25.             System.out.println("识别的结果为："+result);
  26.         } catch (Exception e) {
  27.             e.printStackTrace();
  28.         }
  29.     }
  30. }
8.4)管理敏感词和图片文字识别集成到文章审核

①：在heima-leadnews-common中创建工具类，简单封装一下tess4j

需要先导入pom

  1. <dependency>
  2.     <groupId>net.sourceforge.tess4j</groupId>
  3.     <artifactId>tess4j</artifactId>
  4.     <version>4.1.1</version>
  5. </dependency>

工具类

  1.  
  2. package com.heima.common.tess4j;
  3.  
  4. import lombok.Getter;
  5. import lombok.Setter;
  6. import net.sourceforge.tess4j.ITesseract;
  7. import net.sourceforge.tess4j.Tesseract;
  8. import net.sourceforge.tess4j.TesseractException;
  9. import org.springframework.boot.context.properties.ConfigurationProperties;
  10. import org.springframework.stereotype.Component;
  11.  
  12. import java.awt.image.BufferedImage;
  13.  
  14. @Getter
  15. @Setter
  16. @Component
  17. @ConfigurationProperties(prefix = "tess4j")
  18. public class Tess4jClient {
  19.  
  20.     private String dataPath;
  21.     private String language;
  22.  
  23.     public String doOCR(BufferedImage image) throws TesseractException {
  24.         //创建Tesseract对象
  25.         ITesseract tesseract = new Tesseract();
  26.         //设置字体库路径
  27.         tesseract.setDatapath(dataPath);
  28.         //中文识别
  29.         tesseract.setLanguage(language);
  30.         //执行ocr识别
  31.         String result = tesseract.doOCR(image);
  32.         //替换回车和tal键  使结果为一行
  33.         result = result.replaceAll("\\r|\\n", "-").replaceAll(" ", "");
  34.         return result;
  35.     }
  36.  
  37. }

在spring.factories配置中添加该类,完整如下：

  1.  
  2. org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  3.   com.heima.common.exception.ExceptionCatch,\
  4.   com.heima.common.swagger.SwaggerConfiguration,\
  5.   com.heima.common.swagger.Swagger2Configuration,\
  6.   com.heima.common.aliyun.GreenTextScan,\
  7.   com.heima.common.aliyun.GreenImageScan,\
  8.   com.heima.common.tess4j.Tess4jClient

②：在heima-leadnews-wemedia中的配置中添加两个属性

  1. tess4j:
  2.   data-path: D:\workspace\tessdata
  3.   language: chi_sim

③：在WmNewsAutoScanServiceImpl中的handleImageScan方法上添加如下代码

  1.  
  2. try {
  3.     for (String image : images) {
  4.         byte[] bytes = fileStorageService.downLoadFile(image);
  5.  
  6.         //图片识别文字审核---begin-----
  7.  
  8.         //从byte[]转换为butteredImage
  9.         ByteArrayInputStream in = new ByteArrayInputStream(bytes);
  10.         BufferedImage imageFile = ImageIO.read(in);
  11.         //识别图片的文字
  12.         String result = tess4jClient.doOCR(imageFile);
  13.  
  14.         //审核是否包含自管理的敏感词
  15.         boolean isSensitive = handleSensitiveScan(result, wmNews);
  16.         if(!isSensitive){
  17.             return isSensitive;
  18.         }
  19.  
  20.         //图片识别文字审核---end-----
  21.  
  22.  
  23.         imageList.add(bytes);
  24.  
  25.     } 
  26. }catch (Exception e){
  27.     e.printStackTrace();
  28. }

最后附上文章审核的完整代码如下：

  1.  
  2. package com.heima.wemedia.service.impl;
  3.  
  4. import com.alibaba.fastjson.JSONArray;
  5. import com.baomidou.mybatisplus.core.toolkit.Wrappers;
  6. import com.heima.apis.article.IArticleClient;
  7. import com.heima.common.aliyun.GreenImageScan;
  8. import com.heima.common.aliyun.GreenTextScan;
  9. import com.heima.common.tess4j.Tess4jClient;
  10. import com.heima.file.service.FileStorageService;
  11. import com.heima.model.article.dtos.ArticleDto;
  12. import com.heima.model.common.dtos.ResponseResult;
  13. import com.heima.model.wemedia.pojos.WmChannel;
  14. import com.heima.model.wemedia.pojos.WmNews;
  15. import com.heima.model.wemedia.pojos.WmSensitive;
  16. import com.heima.model.wemedia.pojos.WmUser;
  17. import com.heima.utils.common.SensitiveWordUtil;
  18. import com.heima.wemedia.mapper.WmChannelMapper;
  19. import com.heima.wemedia.mapper.WmNewsMapper;
  20. import com.heima.wemedia.mapper.WmSensitiveMapper;
  21. import com.heima.wemedia.mapper.WmUserMapper;
  22. import com.heima.wemedia.service.WmNewsAutoScanService;
  23. import lombok.extern.slf4j.Slf4j;
  24. import org.apache.commons.lang3.StringUtils;
  25. import org.springframework.beans.BeanUtils;
  26. import org.springframework.beans.factory.annotation.Autowired;
  27. import org.springframework.scheduling.annotation.Async;
  28. import org.springframework.stereotype.Service;
  29. import org.springframework.transaction.annotation.Transactional;
  30.  
  31. import javax.imageio.ImageIO;
  32. import java.awt.image.BufferedImage;
  33. import java.io.ByteArrayInputStream;
  34. import java.util.*;
  35. import java.util.stream.Collectors;
  36.  
  37.  
  38. @Service
  39. @Slf4j
  40. @Transactional
  41. public class WmNewsAutoScanServiceImpl implements WmNewsAutoScanService {
  42.  
  43.     @Autowired
  44.     private WmNewsMapper wmNewsMapper;
  45.  
  46.     /**
  47.      * 自媒体文章审核
  48.      *
  49.      * @param id 自媒体文章id
  50.      */
  51.     @Override
  52.     @Async  //标明当前方法是一个异步方法
  53.     public void autoScanWmNews(Integer id) {
  54.  
  55. //        int a = 1/0;
  56.  
  57.         //1.查询自媒体文章
  58.         WmNews wmNews = wmNewsMapper.selectById(id);
  59.         if (wmNews == null) {
  60.             throw new RuntimeException("WmNewsAutoScanServiceImpl-文章不存在");
  61.         }
  62.  
  63.         if (wmNews.getStatus().equals(WmNews.Status.SUBMIT.getCode())) {
  64.             //从内容中提取纯文本内容和图片
  65.             Map<String, Object> textAndImages = handleTextAndImages(wmNews);
  66.  
  67.             //自管理的敏感词过滤
  68.             boolean isSensitive = handleSensitiveScan((String) textAndImages.get("content"), wmNews);
  69.             if(!isSensitive) return;
  70.  
  71.             //2.审核文本内容  阿里云接口
  72.             boolean isTextScan = handleTextScan((String) textAndImages.get("content"), wmNews);
  73.             if (!isTextScan) return;
  74.  
  75.             //3.审核图片  阿里云接口
  76.             boolean isImageScan = handleImageScan((List<String>) textAndImages.get("images"), wmNews);
  77.             if (!isImageScan) return;
  78.  
  79.             //4.审核成功，保存app端的相关的文章数据
  80.             ResponseResult responseResult = saveAppArticle(wmNews);
  81.             if (!responseResult.getCode().equals(200)) {
  82.                 throw new RuntimeException("WmNewsAutoScanServiceImpl-文章审核，保存app端相关文章数据失败");
  83.             }
  84.             //回填article_id
  85.             wmNews.setArticleId((Long) responseResult.getData());
  86.             updateWmNews(wmNews, (short) 9, "审核成功");
  87.  
  88.         }
  89.     }
  90.  
  91.     @Autowired
  92.     private WmSensitiveMapper wmSensitiveMapper;
  93.  
  94.     /**
  95.      * 自管理的敏感词审核
  96.      * @param content
  97.      * @param wmNews
  98.      * @return
  99.      */
  100.     private boolean handleSensitiveScan(String content, WmNews wmNews) {
  101.  
  102.         boolean flag = true;
  103.  
  104.         //获取所有的敏感词
  105.         List<WmSensitive> wmSensitives = wmSensitiveMapper.selectList(Wrappers.<WmSensitive>lambdaQuery().select(WmSensitive::getSensitives));
  106.         List<String> sensitiveList = wmSensitives.stream().map(WmSensitive::getSensitives).collect(Collectors.toList());
  107.  
  108.         //初始化敏感词库
  109.         SensitiveWordUtil.initMap(sensitiveList);
  110.  
  111.         //查看文章中是否包含敏感词
  112.         Map<String, Integer> map = SensitiveWordUtil.matchWords(content);
  113.         if(map.size() >0){
  114.             updateWmNews(wmNews,(short) 2,"当前文章中存在违规内容"+map);
  115.             flag = false;
  116.         }
  117.  
  118.         return flag;
  119.     }
  120.  
  121.     @Autowired
  122.     private IArticleClient articleClient;
  123.  
  124.     @Autowired
  125.     private WmChannelMapper wmChannelMapper;
  126.  
  127.     @Autowired
  128.     private WmUserMapper wmUserMapper;
  129.  
  130.     /**
  131.      * 保存app端相关的文章数据
  132.      *
  133.      * @param wmNews
  134.      */
  135.     private ResponseResult saveAppArticle(WmNews wmNews) {
  136.  
  137.         ArticleDto dto = new ArticleDto();
  138.         //属性的拷贝
  139.         BeanUtils.copyProperties(wmNews, dto);
  140.         //文章的布局
  141.         dto.setLayout(wmNews.getType());
  142.         //频道
  143.         WmChannel wmChannel = wmChannelMapper.selectById(wmNews.getChannelId());
  144.         if (wmChannel != null) {
  145.             dto.setChannelName(wmChannel.getName());
  146.         }
  147.  
  148.         //作者
  149.         dto.setAuthorId(wmNews.getUserId().longValue());
  150.         WmUser wmUser = wmUserMapper.selectById(wmNews.getUserId());
  151.         if (wmUser != null) {
  152.             dto.setAuthorName(wmUser.getName());
  153.         }
  154.  
  155.         //设置文章id
  156.         if (wmNews.getArticleId() != null) {
  157.             dto.setId(wmNews.getArticleId());
  158.         }
  159.         dto.setCreatedTime(new Date());
  160.  
  161.         ResponseResult responseResult = articleClient.saveArticle(dto);
  162.         return responseResult;
  163.  
  164.     }
  165.  
  166.  
  167.     @Autowired
  168.     private FileStorageService fileStorageService;
  169.  
  170.     @Autowired
  171.     private GreenImageScan greenImageScan;
  172.  
  173.     @Autowired
  174.     private Tess4jClient tess4jClient;
  175.  
  176.     /**
  177.      * 审核图片
  178.      *
  179.      * @param images
  180.      * @param wmNews
  181.      * @return
  182.      */
  183.     private boolean handleImageScan(List<String> images, WmNews wmNews) {
  184.  
  185.         boolean flag = true;
  186.  
  187.         if (images == null || images.size() == 0) {
  188.             return flag;
  189.         }
  190.  
  191.         //下载图片 minIO
  192.         //图片去重
  193.         images = images.stream().distinct().collect(Collectors.toList());
  194.  
  195.         List<byte[]> imageList = new ArrayList<>();
  196.  
  197.         try {
  198.             for (String image : images) {
  199.                 byte[] bytes = fileStorageService.downLoadFile(image);
  200.  
  201.                 //图片识别文字审核---begin-----
  202.  
  203.                 //从byte[]转换为butteredImage
  204.                 ByteArrayInputStream in = new ByteArrayInputStream(bytes);
  205.                 BufferedImage imageFile = ImageIO.read(in);
  206.                 //识别图片的文字
  207.                 String result = tess4jClient.doOCR(imageFile);
  208.  
  209.                 //审核是否包含自管理的敏感词
  210.                 boolean isSensitive = handleSensitiveScan(result, wmNews);
  211.                 if(!isSensitive){
  212.                     return isSensitive;
  213.                 }
  214.  
  215.                 //图片识别文字审核---end-----
  216.  
  217.  
  218.                 imageList.add(bytes);
  219.  
  220.             }
  221.         }catch (Exception e){
  222.             e.printStackTrace();
  223.         }
  224.  
  225.  
  226.         //审核图片
  227.         try {
  228.             Map map = greenImageScan.imageScan(imageList);
  229.             if (map != null) {
  230.                 //审核失败
  231.                 if (map.get("suggestion").equals("block")) {
  232.                     flag = false;
  233.                     updateWmNews(wmNews, (short) 2, "当前文章中存在违规内容");
  234.                 }
  235.  
  236.                 //不确定信息  需要人工审核
  237.                 if (map.get("suggestion").equals("review")) {
  238.                     flag = false;
  239.                     updateWmNews(wmNews, (short) 3, "当前文章中存在不确定内容");
  240.                 }
  241.             }
  242.  
  243.         } catch (Exception e) {
  244.             flag = false;
  245.             e.printStackTrace();
  246.         }
  247.         return flag;
  248.     }
  249.  
  250.     @Autowired
  251.     private GreenTextScan greenTextScan;
  252.  
  253.     /**
  254.      * 审核纯文本内容
  255.      *
  256.      * @param content
  257.      * @param wmNews
  258.      * @return
  259.      */
  260.     private boolean handleTextScan(String content, WmNews wmNews) {
  261.  
  262.         boolean flag = true;
  263.  
  264.         if ((wmNews.getTitle() + "-" + content).length() == 0) {
  265.             return flag;
  266.         }
  267.         try {
  268.             Map map = greenTextScan.greeTextScan((wmNews.getTitle() + "-" + content));
  269.             if (map != null) {
  270.                 //审核失败
  271.                 if (map.get("suggestion").equals("block")) {
  272.                     flag = false;
  273.                     updateWmNews(wmNews, (short) 2, "当前文章中存在违规内容");
  274.                 }
  275.  
  276.                 //不确定信息  需要人工审核
  277.                 if (map.get("suggestion").equals("review")) {
  278.                     flag = false;
  279.                     updateWmNews(wmNews, (short) 3, "当前文章中存在不确定内容");
  280.                 }
  281.             }
  282.         } catch (Exception e) {
  283.             flag = false;
  284.             e.printStackTrace();
  285.         }
  286.  
  287.         return flag;
  288.  
  289.     }
  290.  
  291.     /**
  292.      * 修改文章内容
  293.      *
  294.      * @param wmNews
  295.      * @param status
  296.      * @param reason
  297.      */
  298.     private void updateWmNews(WmNews wmNews, short status, String reason) {
  299.         wmNews.setStatus(status);
  300.         wmNews.setReason(reason);
  301.         wmNewsMapper.updateById(wmNews);
  302.     }
  303.  
  304.     /**
  305.      * 1。从自媒体文章的内容中提取文本和图片
  306.      * 2.提取文章的封面图片
  307.      *
  308.      * @param wmNews
  309.      * @return
  310.      */
  311.     private Map<String, Object> handleTextAndImages(WmNews wmNews) {
  312.  
  313.         //存储纯文本内容
  314.         StringBuilder stringBuilder = new StringBuilder();
  315.  
  316.         List<String> images = new ArrayList<>();
  317.  
  318.         //1。从自媒体文章的内容中提取文本和图片
  319.         if (StringUtils.isNotBlank(wmNews.getContent())) {
  320.             List<Map> maps = JSONArray.parseArray(wmNews.getContent(), Map.class);
  321.             for (Map map : maps) {
  322.                 if (map.get("type").equals("text")) {
  323.                     stringBuilder.append(map.get("value"));
  324.                 }
  325.  
  326.                 if (map.get("type").equals("image")) {
  327.                     images.add((String) map.get("value"));
  328.                 }
  329.             }
  330.         }
  331.         //2.提取文章的封面图片
  332.         if (StringUtils.isNotBlank(wmNews.getImages())) {
  333.             String[] split = wmNews.getImages().split(",");
  334.             images.addAll(Arrays.asList(split));
  335.         }
  336.  
  337.         Map<String, Object> resultMap = new HashMap<>();
  338.         resultMap.put("content", stringBuilder.toString());
  339.         resultMap.put("images", images);
  340.         return resultMap;
  341.  
  342.     }
  343. }

9)文章详情-静态文件生成

9.1)思路分析

文章端创建app相关文章时，生成文章详情静态页上传到MinIO中

9.2)实现步骤

1.新建ArticleFreemarkerService创建静态文件并上传到minIO中

  1.  
  2. package com.heima.article.service;
  3. import com.heima.model.article.pojos.ApArticle;
  4.  
  5. public interface ArticleFreemarkerService {
  6.  
  7.     /**
  8.      * 生成静态文件上传到minIO中
  9.      * @param apArticle
  10.      * @param content
  11.      */
  12.     public void buildArticleToMinIO(ApArticle apArticle,String content);
  13. }

实现

  1.  
  2. package com.heima.article.service.impl;
  3. import java.util.Map;
  4.  
  5. @Service
  6. @Slf4j
  7. @Transactional
  8. public class ArticleFreemarkerServiceImpl implements ArticleFreemarkerService {
  9.  
  10.     @Autowired
  11.     private ApArticleContentMapper apArticleContentMapper;
  12.  
  13.     @Autowired
  14.     private Configuration configuration;
  15.  
  16.     @Autowired
  17.     private FileStorageService fileStorageService;
  18.  
  19.     @Autowired
  20.     private ApArticleService apArticleService;
  21.  
  22.     /**
  23.      * 生成静态文件上传到minIO中
  24.      * @param apArticle
  25.      * @param content
  26.      */
  27.     @Async
  28.     @Override
  29.     public void buildArticleToMinIO(ApArticle apArticle, String content) {
  30.         //已知文章的id
  31.         //4.1 获取文章内容
  32.         if(StringUtils.isNotBlank(content)){
  33.             //4.2 文章内容通过freemarker生成html文件
  34.             Template template = null;
  35.             StringWriter out = new StringWriter();
  36.             try {
  37.                 template = configuration.getTemplate("article.ftl");
  38.                 //数据模型
  39.                 Map<String,Object> contentDataModel = new HashMap<>();
  40.                 contentDataModel.put("content", JSONArray.parseArray(content));
  41.                 //合成
  42.                 template.process(contentDataModel,out);
  43.             } catch (Exception e) {
  44.                 e.printStackTrace();
  45.             }
  46.  
  47.             //4.3 把html文件上传到minio中
  48.             InputStream in = new ByteArrayInputStream(out.toString().getBytes());
  49.             String path = fileStorageService.uploadHtmlFile("", apArticle.getId() + ".html", in);
  50.  
  51.  
  52.             //4.4 修改ap_article表，保存static_url字段
  53.             apArticleService.update(Wrappers.<ApArticle>lambdaUpdate().eq(ApArticle::getId,apArticle.getId())
  54.                     .set(ApArticle::getStaticUrl,path));
  55.         }
  56.     }
  57. }

2.在ApArticleService的saveArticle实现方法中添加调用生成文件的方法

  1.  
  2. /**
  3.      * 保存app端相关文章
  4.      * @param dto
  5.      * @return
  6.      */
  7. @Override
  8. public ResponseResult saveArticle(ArticleDto dto) {
  9.  
  10.     //        try {
  11.     //            Thread.sleep(3000);
  12.     //        } catch (InterruptedException e) {
  13.     //            e.printStackTrace();
  14.     //        }
  15.     //1.检查参数
  16.     if(dto == null){
  17.         return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
  18.     }
  19.  
  20.     ApArticle apArticle = new ApArticle();
  21.     BeanUtils.copyProperties(dto,apArticle);
  22.  
  23.     //2.判断是否存在id
  24.     if(dto.getId() == null){
  25.         //2.1 不存在id  保存  文章  文章配置  文章内容
  26.  
  27.         //保存文章
  28.         save(apArticle);
  29.  
  30.         //保存配置
  31.         ApArticleConfig apArticleConfig = new ApArticleConfig(apArticle.getId());
  32.         apArticleConfigMapper.insert(apArticleConfig);
  33.  
  34.         //保存 文章内容
  35.         ApArticleContent apArticleContent = new ApArticleContent();
  36.         apArticleContent.setArticleId(apArticle.getId());
  37.         apArticleContent.setContent(dto.getContent());
  38.         apArticleContentMapper.insert(apArticleContent);
  39.  
  40.     }else {
  41.         //2.2 存在id   修改  文章  文章内容
  42.  
  43.         //修改  文章
  44.         updateById(apArticle);
  45.  
  46.         //修改文章内容
  47.         ApArticleContent apArticleContent = apArticleContentMapper.selectOne(Wrappers.<ApArticleContent>lambdaQuery().eq(ApArticleContent::getArticleId, dto.getId()));
  48.         apArticleContent.setContent(dto.getContent());
  49.         apArticleContentMapper.updateById(apArticleContent);
  50.     }
  51.  
  52.     //异步调用 生成静态文件上传到minio中
  53.     articleFreemarkerService.buildArticleToMinIO(apArticle,dto.getContent());
  54.  
  55.  
  56.     //3.结果返回  文章的id
  57.     return ResponseResult.okResult(apArticle.getId());
  58. }

3.文章微服务开启异步调用

05延迟任务精准发布文章

1)文章定时发布

2)延迟任务概述

2.1)什么是延迟任务

  • 定时任务：有固定周期的，有明确的触发时间

  • 延迟队列：没有固定的开始时间，它常常是由一个事件触发的，而在这个事件触发之后的一段时间内触发另一个事件，任务可以立即执行，也可以延迟

应用场景：

场景一：

订单下单之后30分钟后，如果用户没有付钱，则系统自动取消订单；如果期间下单成功，任务取消

场景二：接口对接出现网络问题，1分钟后重试，如果失败，2分钟重试，直到出现阈值终止

2.2)技术对比
2.2.1)DelayQueue

JDK自带DelayQueue 是一个支持延时获取元素的阻塞队列， 内部采用优先队列 PriorityQueue 存储元素，同时元素必须实现 Delayed 接口；在创建元素时可以指定多久才可以从队列中获取当前元素，只有在延迟期满时才能从队列中提取元素

DelayQueue属于排序队列，它的特殊之处在于队列的元素必须实现Delayed接口，该接口需要实现compareTo和getDelay方法

getDelay方法：获取元素在队列中的剩余时间，只有当剩余时间为0时元素才可以出队列。

compareTo方法：用于排序，确定元素出队列的顺序。

实现：

1：在测试包jdk下创建延迟任务元素对象DelayedTask，实现compareTo和getDelay方法，

2：在main方法中创建DelayQueue并向延迟队列中添加三个延迟任务，

3：循环的从延迟队列中拉取任务

  1.  
  2. public class DelayedTask  implements Delayed{
  3.     
  4.     // 任务的执行时间
  5.     private int executeTime = 0;
  6.     
  7.     public DelayedTask(int delay){
  8.         Calendar calendar = Calendar.getInstance();
  9.         calendar.add(Calendar.SECOND,delay);
  10.         this.executeTime = (int)(calendar.getTimeInMillis() /1000 );
  11.     }
  12.  
  13.     /**
  14.      * 元素在队列中的剩余时间
  15.      * @param unit
  16.      * @return
  17.      */
  18.     @Override
  19.     public long getDelay(TimeUnit unit) {
  20.         Calendar calendar = Calendar.getInstance();
  21.         return executeTime - (calendar.getTimeInMillis()/1000);
  22.     }
  23.  
  24.     /**
  25.      * 元素排序
  26.      * @param o
  27.      * @return
  28.      */
  29.     @Override
  30.     public int compareTo(Delayed o) {
  31.         long val = this.getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
  32.         return val == 0 ? 0 : ( val < 0 ? -1: 1 );
  33.     }
  34.  
  35.  
  36.     public static void main(String[] args) {
  37.         DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>();
  38.         
  39.         queue.add(new DelayedTask(5));
  40.         queue.add(new DelayedTask(10));
  41.         queue.add(new DelayedTask(15));
  42.  
  43.         System.out.println(System.currentTimeMillis()/1000+" start consume ");
  44.         while(queue.size() != 0){
  45.             DelayedTask delayedTask = queue.poll();
  46.             if(delayedTask !=null ){
  47.                 System.out.println(System.currentTimeMillis()/1000+" cosume task");
  48.             }
  49.             //每隔一秒消费一次
  50.             try {
  51.                 Thread.sleep(1000);
  52.             } catch (InterruptedException e) {
  53.                 e.printStackTrace();
  54.             }
  55.         }     
  56.     }
  57. }

DelayQueue实现完成之后思考一个问题：

使用线程池或者原生DelayQueue程序挂掉之后，任务都是放在内存，需要考虑未处理消息的丢失带来的影响，如何保证数据不丢失，需要持久化（磁盘）

2.2.2)RabbitMQ实现延迟任务

  • TTL：Time To Live (消息存活时间)

  • 死信队列：Dead Letter Exchange(死信交换机)，当消息成为Dead message后，可以重新发送另一个交换机（死信交换机）

2.2.3)redis实现

zset数据类型的去重有序（分数排序）特点进行延迟。例如：时间戳作为score进行排序

3)redis实现延迟任务

实现思路

问题思路

1.为什么任务需要存储在数据库中？

延迟任务是一个通用的服务，任何需要延迟得任务都可以调用该服务，需要考虑数据持久化的问题，存储数据库中是一种数据安全的考虑。

2.为什么redis中使用两种数据类型，list和zset？

效率问题，算法的时间复杂度； list是双向链表

3.在添加zset数据的时候，为什么不需要预加载？

任务模块是一个通用的模块，项目中任何需要延迟队列的地方，都可以调用这个接口，要考虑到数据量的问题，如果数据量特别大，为了防止zset阻塞，只需要把未来几分钟要执行的数据存入缓存即可。

锐评：完全为了学list zset而编出来的场景，实际工作中延迟队列要设计成这样只能说太蠢了

实际工作绝对用MQ

4)延迟任务服务实现

4.1)搭建heima-leadnews-schedule模块

leadnews-schedule是一个通用的服务，单独创建模块来管理任何类型的延迟任务

①：导入资料文件夹的heima-leadnews-schedule模块到heima-leadnews-service下，如下图所示：

②：添加bootstrap.yml

  1. server:
  2.   port: 51701
  3. spring:
  4.   application:
  5.     name: leadnews-schedule
  6.   cloud:
  7.     nacos:
  8.       discovery:
  9.         server-addr: 192.168.200.130:8848
  10.       config:
  11.         server-addr: 192.168.200.130:8848
  12.         file-extension: yml

③：在nacos中添加对应配置，并添加数据库及mybatis-plus的配置

  1. spring:
  2.   datasource:
  3.     driver-class-name: com.mysql.jdbc.Driver
  4.     url: jdbc:mysql://localhost:3306/leadnews_schedule?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC
  5.     username: root
  6.     password: root
  7. # 设置Mapper接口所对应的XML文件位置，如果你在Mapper接口中有自定义方法，需要进行该配置
  8. mybatis-plus:
  9.   mapper-locations: classpath*:mapper/*.xml
  10.   # 设置别名包扫描路径，通过该属性可以给包中的类注册别名
  11.   type-aliases-package: com.heima.model.schedule.pojos
4.2)数据库准备

导入资料中leadnews_schedule数据库

taskinfo 任务表

实体类

  1.  
  2. package com.heima.model.schedule.pojos;
  3. import java.io.Serializable;
  4. import java.util.Date;
  5.  
  6. /**
  7.  * <p>
  8.  * 
  9.  * </p>
  10.  *
  11.  * @author itheima
  12.  */
  13. @Data
  14. @TableName("taskinfo")
  15. public class Taskinfo implements Serializable {
  16.  
  17.     private static final long serialVersionUID = 1L;
  18.  
  19.     /**
  20.      * 任务id
  21.      */
  22.     @TableId(type = IdType.ID_WORKER)
  23.     private Long taskId;
  24.  
  25.     /**
  26.      * 执行时间
  27.      */
  28.     @TableField("execute_time")
  29.     private Date executeTime;
  30.  
  31.     /**
  32.      * 参数
  33.      */
  34.     @TableField("parameters")
  35.     private byte[] parameters;
  36.  
  37.     /**
  38.      * 优先级
  39.      */
  40.     @TableField("priority")
  41.     private Integer priority;
  42.  
  43.     /**
  44.      * 任务类型
  45.      */
  46.     @TableField("task_type")
  47.     private Integer taskType;
  48.  
  49.  
  50. }

taskinfo_logs 任务日志表

实体类

  1.  
  2. package com.heima.model.schedule.pojos;
  3. import java.io.Serializable;
  4. import java.util.Date;
  5.  
  6. /**
  7.  * <p>
  8.  * 
  9.  * </p>
  10.  *
  11.  * @author itheima
  12.  */
  13. @Data
  14. @TableName("taskinfo_logs")
  15. public class TaskinfoLogs implements Serializable {
  16.  
  17.     private static final long serialVersionUID = 1L;
  18.  
  19.     /**
  20.      * 任务id
  21.      */
  22.     @TableId(type = IdType.ID_WORKER)
  23.     private Long taskId;
  24.  
  25.     /**
  26.      * 执行时间
  27.      */
  28.     @TableField("execute_time")
  29.     private Date executeTime;
  30.  
  31.     /**
  32.      * 参数
  33.      */
  34.     @TableField("parameters")
  35.     private byte[] parameters;
  36.  
  37.     /**
  38.      * 优先级
  39.      */
  40.     @TableField("priority")
  41.     private Integer priority;
  42.  
  43.     /**
  44.      * 任务类型
  45.      */
  46.     @TableField("task_type")
  47.     private Integer taskType;
  48.  
  49.     /**
  50.      * 版本号,用乐观锁
  51.      */
  52.     @Version
  53.     private Integer version;
  54.  
  55.     /**
  56.      * 状态 0=int 1=EXECUTED 2=CANCELLED
  57.      */
  58.     @TableField("status")
  59.     private Integer status;
  60.  
  61.  
  62. }
乐观锁/悲观锁

悲观锁效率低；

乐观锁支持：

  1.  
  2. /**
  3.      * mybatis-plus乐观锁支持
  4.      * @return
  5.      */
  6. @Bean
  7. public MybatisPlusInterceptor optimisticLockerInterceptor(){
  8.     MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
  9.     interceptor.addInnerInterceptor(new OptimisticLockerInnerInterceptor());
  10.     return interceptor;
  11. }
4.3)安装redis

①拉取镜像

docker pull redis

② 创建容器

docker run -d --name redis --restart=always -p 6379:6379 redis --requirepass "leadnews"

③链接测试

打开资料中的Redis Desktop Manager，输入host、port、password链接测试

能链接成功，即可

4.4)项目集成redis

① 在项目导入redis相关依赖，已经完成

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-data-redis</artifactId>
  4. </dependency>
  5. <!-- redis依赖commons-pool 这个依赖一定要添加 -->
  6. <dependency>
  7.     <groupId>org.apache.commons</groupId>
  8.     <artifactId>commons-pool2</artifactId>
  9. </dependency>

② 在heima-leadnews-schedule中集成redis,添加以下nacos配置，链接上redis

  1. spring:
  2.   redis:
  3.     host: 192.168.200.130
  4.     password: leadnews
  5.     port: 6379

③ 拷贝资料文件夹下的类：CacheService到heima-leadnews-common模块下，并添加自动配置

④：测试

  1.  
  2. package com.heima.schedule.test;
  3. import java.util.Set;
  4.  
  5. @SpringBootTest(classes = ScheduleApplication.class)
  6. @RunWith(SpringRunner.class)
  7. public class RedisTest {
  8.  
  9.     @Autowired
  10.     private CacheService cacheService;
  11.  
  12.     @Test
  13.     public void testList(){
  14.  
  15.         //在list的左边添加元素
  16. //        cacheService.lLeftPush("list_001","hello,redis");
  17.  
  18.         //在list的右边获取元素，并删除
  19.         String list_001 = cacheService.lRightPop("list_001");
  20.         System.out.println(list_001);
  21.     }
  22.  
  23.     @Test
  24.     public void testZset(){
  25.         //添加数据到zset中  分值
  26.         /*cacheService.zAdd("zset_key_001","hello zset 001",1000);
  27.         cacheService.zAdd("zset_key_001","hello zset 002",8888);
  28.         cacheService.zAdd("zset_key_001","hello zset 003",7777);
  29.         cacheService.zAdd("zset_key_001","hello zset 004",999999);*/
  30.  
  31.         //按照分值获取数据
  32.         Set<String> zset_key_001 = cacheService.zRangeByScore("zset_key_001", 0, 8888);
  33.         System.out.println(zset_key_001);
  34.     }
  35. }
4.5)添加任务

①：拷贝mybatis-plus生成的文件，mapper

②：创建task类，用于接收添加任务的参数

  1.  
  2. package com.heima.model.schedule.dtos;
  3. import lombok.Data;
  4. import java.io.Serializable;
  5.  
  6. @Data
  7. public class Task implements Serializable {
  8.  
  9.     /**
  10.      * 任务id
  11.      */
  12.     private Long taskId;
  13.     /**
  14.      * 类型
  15.      */
  16.     private Integer taskType;
  17.  
  18.     /**
  19.      * 优先级
  20.      */
  21.     private Integer priority;
  22.  
  23.     /**
  24.      * 执行id
  25.      */
  26.     private long executeTime;
  27.  
  28.     /**
  29.      * task参数
  30.      */
  31.     private byte[] parameters;
  32.     
  33. }

③：创建TaskService

  1.  
  2. package com.heima.schedule.service;
  3. import com.heima.model.schedule.dtos.Task;
  4. /**
  5.  * 对外访问接口
  6.  */
  7. public interface TaskService {
  8.  
  9.     /**
  10.      * 添加任务
  11.      * @param task   任务对象
  12.      * @return       任务id
  13.      */
  14.     public long addTask(Task task) ;
  15.  
  16. }

实现：

  1.  
  2. package com.heima.schedule.service.impl;
  3. import java.util.Calendar;
  4. import java.util.Date;
  5.  
  6. @Service
  7. @Transactional
  8. @Slf4j
  9. public class TaskServiceImpl implements TaskService {
  10.     /**
  11.      * 添加延迟任务
  12.      *
  13.      * @param task
  14.      * @return
  15.      */
  16.     @Override
  17.     public long addTask(Task task) {
  18.         //1.添加任务到数据库中
  19.  
  20.         boolean success = addTaskToDb(task);
  21.  
  22.         if (success) {
  23.             //2.添加任务到redis
  24.             addTaskToCache(task);
  25.         }
  26.         return task.getTaskId();
  27.     }
  28.  
  29.     @Autowired
  30.     private CacheService cacheService;
  31.  
  32.     /**
  33.      * 把任务添加到redis中
  34.      *
  35.      * @param task
  36.      */
  37.     private void addTaskToCache(Task task) {
  38.  
  39.         String key = task.getTaskType() + "_" + task.getPriority();
  40.  
  41.         //获取5分钟之后的时间  毫秒值
  42.         Calendar calendar = Calendar.getInstance();
  43.         calendar.add(Calendar.MINUTE, 5);
  44.         long nextScheduleTime = calendar.getTimeInMillis();
  45.  
  46.         //2.1 如果任务的执行时间小于等于当前时间，存入list
  47.         if (task.getExecuteTime() <= System.currentTimeMillis()) {
  48.             cacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
  49.         } else if (task.getExecuteTime() <= nextScheduleTime) {
  50.             //2.2 如果任务的执行时间大于当前时间 && 小于等于预设时间（未来5分钟） 存入zset中
  51.             cacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
  52.         }
  53.     }
  54.  
  55.     @Autowired
  56.     private TaskinfoMapper taskinfoMapper;
  57.  
  58.     @Autowired
  59.     private TaskinfoLogsMapper taskinfoLogsMapper;
  60.  
  61.     /**
  62.      * 添加任务到数据库中
  63.      *
  64.      * @param task
  65.      * @return
  66.      */
  67.     private boolean addTaskToDb(Task task) {
  68.  
  69.         boolean flag = false;
  70.  
  71.         try {
  72.             //保存任务表
  73.             Taskinfo taskinfo = new Taskinfo();
  74.             BeanUtils.copyProperties(task, taskinfo);
  75.             taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
  76.             taskinfoMapper.insert(taskinfo);
  77.  
  78.             //设置taskID
  79.             task.setTaskId(taskinfo.getTaskId());
  80.  
  81.             //保存任务日志数据
  82.             TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
  83.             BeanUtils.copyProperties(taskinfo, taskinfoLogs);
  84.             taskinfoLogs.setVersion(1);
  85.             taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
  86.             taskinfoLogsMapper.insert(taskinfoLogs);
  87.  
  88.             flag = true;
  89.         } catch (Exception e) {
  90.             e.printStackTrace();
  91.         }
  92.  
  93.         return flag;
  94.     }
  95. }

ScheduleConstants常量类

  1.  
  2. package com.heima.common.constants;
  3.  
  4. public class ScheduleConstants {
  5.  
  6.     //task状态
  7.     public static final int SCHEDULED=0;   //初始化状态
  8.  
  9.     public static final int EXECUTED=1;       //已执行状态
  10.  
  11.     public static final int CANCELLED=2;   //已取消状态
  12.  
  13.     public static String FUTURE="future_";   //未来数据key前缀
  14.  
  15.     public static String TOPIC="topic_";     //当前数据key前缀
  16. }

④：测试

4.6)取消任务

在TaskService中添加方法

  1.  
  2. /**
  3.      * 取消任务
  4.      * @param taskId        任务id
  5.      * @return              取消结果
  6.      */
  7. public boolean cancelTask(long taskId);

实现

  1.  
  2. /**
  3.      * 取消任务
  4.      * @param taskId
  5.      * @return
  6.      */
  7. @Override
  8. public boolean cancelTask(long taskId) {
  9.  
  10.     boolean flag = false;
  11.  
  12.     //删除任务，更新日志
  13.     Task task = updateDb(taskId,ScheduleConstants.EXECUTED);
  14.  
  15.     //删除redis的数据
  16.     if(task != null){
  17.         removeTaskFromCache(task);
  18.         flag = true;
  19.     }
  20.     return false;
  21. }
  22.  
  23. /**
  24.      * 删除redis中的任务数据
  25.      * @param task
  26.      */
  27. private void removeTaskFromCache(Task task) {
  28.  
  29.     String key = task.getTaskType()+"_"+task.getPriority();
  30.  
  31.     if(task.getExecuteTime()<=System.currentTimeMillis()){
  32.         cacheService.lRemove(ScheduleConstants.TOPIC+key,0,JSON.toJSONString(task));
  33.     }else {
  34.         cacheService.zRemove(ScheduleConstants.FUTURE+key, JSON.toJSONString(task));
  35.     }
  36. }
  37.  
  38. /**
  39.      * 删除任务，更新任务日志状态
  40.      * @param taskId
  41.      * @param status
  42.      * @return
  43.      */
  44. private Task updateDb(long taskId, int status) {
  45.     Task task = null;
  46.     try {
  47.         //删除任务
  48.         taskinfoMapper.deleteById(taskId);
  49.  
  50.         TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
  51.         taskinfoLogs.setStatus(status);
  52.         taskinfoLogsMapper.updateById(taskinfoLogs);
  53.  
  54.         task = new Task();
  55.         BeanUtils.copyProperties(taskinfoLogs,task);
  56.         task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
  57.     }catch (Exception e){
  58.         log.error("task cancel exception taskid={}",taskId);
  59.     }
  60.     return task;
  61. }

测试

4.7)消费任务

在TaskService中添加方法

  1.  
  2. /**
  3.  * 按照类型和优先级来拉取任务
  4.  * @param type
  5.  * @param priority
  6.  * @return
  7.  */
  8. public Task poll(int type,int priority);

实现

  1.  
  2. /**
  3.      * 按照类型和优先级拉取任务
  4.      * @return
  5.      */
  6. @Override
  7. public Task poll(int type,int priority) {
  8.     Task task = null;
  9.     try {
  10.         String key = type+"_"+priority;
  11.         String task_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
  12.         if(StringUtils.isNotBlank(task_json)){
  13.             task = JSON.parseObject(task_json, Task.class);
  14.             //更新数据库信息
  15.             updateDb(task.getTaskId(),ScheduleConstants.EXECUTED);
  16.         }
  17.     }catch (Exception e){
  18.         e.printStackTrace();
  19.         log.error("poll task exception");
  20.     }
  21.  
  22.     return task;
  23. }
4.8)未来数据定时刷新
4.8.1)reids key值匹配

方案1：keys 模糊匹配

keys的模糊匹配功能很方便也很强大，但是在生产环境需要慎用！开发中使用keys的模糊匹配却发现redis的CPU使用率极高，所以公司的redis生产环境将keys命令禁用了！redis是单线程，会被堵塞

方案2：scan

SCAN 命令是一个基于游标的迭代器，SCAN命令每次被调用之后， 都会向用户返回一个新的游标， 用户在下次迭代时需要使用这个新游标作为SCAN命令的游标参数， 以此来延续之前的迭代过程。

代码案例：

  1.  
  2. @Test
  3. public void testKeys(){
  4.     Set<String> keys = cacheService.keys("future_*");
  5.     System.out.println(keys);
  6.  
  7.     Set<String> scan = cacheService.scan("future_*");
  8.     System.out.println(scan);
  9. }
4.8.2)reids管道

普通redis客户端和服务器交互模式 性能很低

Pipeline请求模型

官方测试结果数据对比

测试案例对比：

  1.  
  2. //耗时6151
  3. @Test
  4. public  void testPiple1(){
  5.     long start =System.currentTimeMillis();
  6.     for (int i = 0; i <10000 ; i++) {
  7.         Task task = new Task();
  8.         task.setTaskType(1001);
  9.         task.setPriority(1);
  10.         task.setExecuteTime(new Date().getTime());
  11.         cacheService.lLeftPush("1001_1", JSON.toJSONString(task));
  12.     }
  13.     System.out.println("耗时"+(System.currentTimeMillis()- start));
  14. }
  15.  
  16. @Test
  17. public void testPiple2(){
  18.     long start  = System.currentTimeMillis();
  19.     //使用管道技术
  20.     List<Object> objectList = cacheService.getstringRedisTemplate().executePipelined(new RedisCallback<Object>() {
  21.         @Nullable
  22.         @Override
  23.         public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
  24.             for (int i = 0; i <10000 ; i++) {
  25.                 Task task = new Task();
  26.                 task.setTaskType(1001);
  27.                 task.setPriority(1);
  28.                 task.setExecuteTime(new Date().getTime());
  29.                 redisConnection.lPush("1001_1".getBytes(), JSON.toJSONString(task).getBytes());
  30.             }
  31.             return null;
  32.         }
  33.     });
  34.     System.out.println("使用管道技术执行10000次自增操作共耗时:"+(System.currentTimeMillis()-start)+"毫秒");
  35. }
4.8.3)未来数据定时刷新-功能完成

在TaskService中添加方法

  1.  
  2. @Scheduled(cron = "0 */1 * * * ?")//定时 (每分钟执行一次
  3. //{秒数} {分钟} {小时} {日期} {月份} {星期} {年份(可为空)}
  4. public void refresh() {
  5.     System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");
  6.  
  7.     // 获取所有未来数据集合的key值
  8.     Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_*
  9.     for (String futureKey : futureKeys) { // future_250_250
  10.  
  11.         String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];
  12.         //获取该组key下当前需要消费的任务数据
  13.         Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
  14.         if (!tasks.isEmpty()) {
  15.             //将这些任务数据添加到消费者队列中
  16.             cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
  17.             System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
  18.         }
  19.     }
  20. }

在引导类中添加开启任务调度注解：@EnableScheduling

4.9)分布式锁解决集群下的方法抢占执行
4.9.1)问题描述

启动两台heima-leadnews-schedule服务，每台服务都会去执行refresh定时任务方法

4.9.2)分布式锁

分布式锁：控制分布式系统有序的去对共享资源进行操作，通过互斥来保证数据的一致性。

解决方案：

4.9.3)redis分布式锁

sexnx （SET if Not eXists）命令在指定的 key 不存在时，为 key 设置指定的值。

这种加锁的思路是，如果 key 不存在则为 key 设置 value，如果 key 已存在则 SETNX 命令不做任何操作

  • 客户端A请求服务器设置key的值，如果设置成功就表示加锁成功

  • 客户端B也去请求服务器设置key的值，如果返回失败，那么就代表加锁失败

  • 客户端A执行代码完成，删除锁

  • 客户端B在等待一段时间后再去请求设置key的值，设置成功

  • 客户端B执行代码完成，删除锁

4.9.4)在工具类CacheService中添加方法
  1.  
  2. /**
  3.  * 加锁
  4.  *
  5.  * @param name
  6.  * @param expire
  7.  * @return
  8.  */
  9. public String tryLock(String name, long expire) {
  10.     name = name + "_lock";
  11.     String token = UUID.randomUUID().toString();
  12.     RedisConnectionFactory factory = stringRedisTemplate.getConnectionFactory();
  13.     RedisConnection conn = factory.getConnection();
  14.     try {
  15.  
  16.         //参考redis命令：
  17.         //set key value [EX seconds] [PX milliseconds] [NX|XX]
  18.         Boolean result = conn.set(
  19.                 name.getBytes(),
  20.                 token.getBytes(),
  21.                 Expiration.from(expire, TimeUnit.MILLISECONDS),
  22.                 RedisStringCommands.SetOption.SET_IF_ABSENT //NX
  23.         );
  24.         if (result != null && result)
  25.             return token;
  26.     } finally {
  27.         RedisConnectionUtils.releaseConnection(conn, factory,false);
  28.     }
  29.     return null;
  30. }

修改未来数据定时刷新的方法，如下：

  1.  
  2. /**
  3.  * 未来数据定时刷新
  4.  */
  5. @Scheduled(cron = "0 */1 * * * ?")
  6. public void refresh(){
  7.  
  8.     String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
  9.     if(StringUtils.isNotBlank(token)){
  10.         log.info("未来数据定时刷新---定时任务");
  11.  
  12.         //获取所有未来数据的集合key
  13.         Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");
  14.         for (String futureKey : futureKeys) {//future_100_50
  15.  
  16.             //获取当前数据的key  topic
  17.             String topicKey = ScheduleConstants.TOPIC+futureKey.split(ScheduleConstants.FUTURE)[1];
  18.  
  19.             //按照key和分值查询符合条件的数据
  20.             Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
  21.  
  22.             //同步数据
  23.             if(!tasks.isEmpty()){
  24.                 cacheService.refreshWithPipeline(futureKey,topicKey,tasks);
  25.                 log.info("成功的将"+futureKey+"刷新到了"+topicKey);
  26.             }
  27.         }
  28.     }
  29. }
4.10)数据库同步到redis
  1.  
  2. @Scheduled(cron = "0 */5 * * * ?")
  3. @PostConstruct
  4. public void reloadData() {
  5.     clearCache();
  6.     log.info("数据库数据同步到缓存");
  7.     Calendar calendar = Calendar.getInstance();
  8.     calendar.add(Calendar.MINUTE, 5);
  9.  
  10.     //查看小于未来5分钟的所有任务
  11.     List<Taskinfo> allTasks = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime,calendar.getTime()));
  12.     if(allTasks != null && allTasks.size() > 0){
  13.         for (Taskinfo taskinfo : allTasks) {
  14.             Task task = new Task();
  15.             BeanUtils.copyProperties(taskinfo,task);
  16.             task.setExecuteTime(taskinfo.getExecuteTime().getTime());
  17.             addTaskToCache(task);
  18.         }
  19.     }
  20. }
  21.  
  22. private void clearCache(){
  23.     // 删除缓存中未来数据集合和当前消费者队列的所有key
  24.     Set<String> futurekeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_
  25.     Set<String> topickeys = cacheService.scan(ScheduleConstants.TOPIC + "*");// topic_
  26.     cacheService.delete(futurekeys);
  27.     cacheService.delete(topickeys);
  28. }

5)延迟队列解决精准时间发布文章

5.1)延迟队列服务提供对外接口

提供远程的feign接口，在heima-leadnews-feign-api编写类如下：

  1.  
  2. package com.heima.apis.schedule;
  3. import org.springframework.web.bind.annotation.RequestBody;
  4.  
  5. @FeignClient("leadnews-schedule")
  6. public interface IScheduleClient {
  7.  
  8.     /**
  9.      * 添加任务
  10.      * @param task   任务对象
  11.      * @return       任务id
  12.      */
  13.     @PostMapping("/api/v1/task/add")
  14.     public ResponseResult  addTask(@RequestBody Task task);
  15.  
  16.     /**
  17.      * 取消任务
  18.      * @param taskId        任务id
  19.      * @return              取消结果
  20.      */
  21.     @GetMapping("/api/v1/task/cancel/{taskId}")
  22.     public ResponseResult cancelTask(@PathVariable("taskId") long taskId);
  23.  
  24.     /**
  25.      * 按照类型和优先级来拉取任务
  26.      * @param type
  27.      * @param priority
  28.      * @return
  29.      */
  30.     @GetMapping("/api/v1/task/poll/{type}/{priority}")
  31.     public ResponseResult poll(@PathVariable("type") int type,@PathVariable("priority")  int priority);
  32. }

在heima-leadnews-schedule微服务下提供对应的实现

  1.  
  2. package com.heima.schedule.feign;
  3. import org.springframework.web.bind.annotation.*;
  4.  
  5. @RestController
  6. public class ScheduleClient  implements IScheduleClient {
  7.  
  8.     @Autowired
  9.     private TaskService taskService;
  10.  
  11.     /**
  12.      * 添加任务
  13.      * @param task 任务对象
  14.      * @return 任务id
  15.      */
  16.     @PostMapping("/api/v1/task/add")
  17.     @Override
  18.     public ResponseResult addTask(@RequestBody Task task) {
  19.         return ResponseResult.okResult(taskService.addTask(task));
  20.     }
  21.  
  22.     /**
  23.      * 取消任务
  24.      * @param taskId 任务id
  25.      * @return 取消结果
  26.      */
  27.     @GetMapping("/api/v1/task/cancel/{taskId}")
  28.     @Override
  29.     public ResponseResult cancelTask(@PathVariable("taskId") long taskId) {
  30.         return ResponseResult.okResult(taskService.cancelTask(taskId));
  31.     }
  32.  
  33.     /**
  34.      * 按照类型和优先级来拉取任务
  35.      * @param type
  36.      * @param priority
  37.      * @return
  38.      */
  39.     @GetMapping("/api/v1/task/poll/{type}/{priority}")
  40.     @Override
  41.     public ResponseResult poll(@PathVariable("type") int type, @PathVariable("priority") int priority) {
  42.         return ResponseResult.okResult(taskService.poll(type,priority));
  43.     }
  44. }
5.2)发布文章集成添加延迟队列接口

在创建WmNewsTaskService

  1.  
  2. package com.heima.wemedia.service;
  3.  
  4. import com.heima.model.wemedia.pojos.WmNews;
  5.  
  6.  
  7. public interface WmNewsTaskService {
  8.  
  9.     /**
  10.      * 添加任务到延迟队列中
  11.      * @param id  文章的id
  12.      * @param publishTime  发布的时间  可以做为任务的执行时间
  13.      */
  14.     public void addNewsToTask(Integer id, Date publishTime);
  15. }

实现：

  1.  
  2. package com.heima.wemedia.service.impl;
  3. import org.springframework.stereotype.Service;
  4.  
  5. @Service
  6. @Slf4j
  7. public class WmNewsTaskServiceImpl  implements WmNewsTaskService {
  8.  
  9.  
  10.     @Autowired
  11.     private IScheduleClient scheduleClient;
  12.  
  13.     /**
  14.      * 添加任务到延迟队列中
  15.      * @param id          文章的id
  16.      * @param publishTime 发布的时间  可以做为任务的执行时间
  17.      */
  18.     @Override
  19.     @Async
  20.     public void addNewsToTask(Integer id, Date publishTime) {
  21.  
  22.         log.info("添加任务到延迟服务中----begin");
  23.  
  24.         Task task = new Task();
  25.         task.setExecuteTime(publishTime.getTime());
  26.         task.setTaskType(TaskTypeEnum.NEWS_SCAN_TIME.getTaskType());
  27.         task.setPriority(TaskTypeEnum.NEWS_SCAN_TIME.getPriority());
  28.         WmNews wmNews = new WmNews();
  29.         wmNews.setId(id);
  30.         task.setParameters(ProtostuffUtil.serialize(wmNews));
  31.  
  32.         scheduleClient.addTask(task);
  33.  
  34.         log.info("添加任务到延迟服务中----end");
  35.  
  36.     }
  37.     
  38. }

枚举类：

  1.  
  2. package com.heima.model.common.enums;
  3.  
  4. import lombok.AllArgsConstructor;
  5. import lombok.Getter;
  6.  
  7. @Getter
  8. @AllArgsConstructor
  9. public enum TaskTypeEnum {
  10.  
  11.     NEWS_SCAN_TIME(1001, 1,"文章定时审核"),
  12.     REMOTEERROR(1002, 2,"第三方接口调用失败，重试");
  13.     private final int taskType; //对应具体业务
  14.     private final int priority; //业务不同级别
  15.     private final String desc; //描述信息
  16. }
序列化工具对比

  • JdkSerialize：java内置的序列化能将实现了Serilazable接口的对象进行序列化和反序列化， ObjectOutputStream的writeObject()方法可序列化对象生成字节数组

  • Protostuff：google开源的protostuff采用更为紧凑的二进制数组，表现更加优异，然后使用protostuff的编译工具生成pojo类

拷贝资料中的两个类到heima-leadnews-utils下

Protostuff需要引导依赖：

  1. <dependency>
  2.     <groupId>io.protostuff</groupId>
  3.     <artifactId>protostuff-core</artifactId>
  4.     <version>1.6.0</version>
  5. </dependency>
  6.  
  7. <dependency>
  8.     <groupId>io.protostuff</groupId>
  9.     <artifactId>protostuff-runtime</artifactId>
  10.     <version>1.6.0</version>
  11. </dependency>

修改发布文章代码：

把之前的异步调用修改为调用延迟任务

  1.  
  2. @Autowired
  3. private WmNewsTaskService wmNewsTaskService;
  4.  
  5. /**
  6.      * 发布修改文章或保存为草稿
  7.      * @param dto
  8.      * @return
  9.      */
  10. @Override
  11. public ResponseResult submitNews(WmNewsDto dto) {
  12.  
  13.     //0.条件判断
  14.     if(dto == null || dto.getContent() == null){
  15.         return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);
  16.     }
  17.  
  18.     //1.保存或修改文章
  19.  
  20.     WmNews wmNews = new WmNews();
  21.     //属性拷贝 属性名词和类型相同才能拷贝
  22.     BeanUtils.copyProperties(dto,wmNews);
  23.     //封面图片  list---> string
  24.     if(dto.getImages() != null && dto.getImages().size() > 0){
  25.         //[1dddfsd.jpg,sdlfjldk.jpg]-->   1dddfsd.jpg,sdlfjldk.jpg
  26.         String imageStr = StringUtils.join(dto.getImages(), ",");
  27.         wmNews.setImages(imageStr);
  28.     }
  29.     //如果当前封面类型为自动 -1
  30.     if(dto.getType().equals(WemediaConstants.WM_NEWS_TYPE_AUTO)){
  31.         wmNews.setType(null);
  32.     }
  33.  
  34.     saveOrUpdateWmNews(wmNews);
  35.  
  36.     //2.判断是否为草稿  如果为草稿结束当前方法
  37.     if(dto.getStatus().equals(WmNews.Status.NORMAL.getCode())){
  38.         return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
  39.     }
  40.  
  41.     //3.不是草稿，保存文章内容图片与素材的关系
  42.     //获取到文章内容中的图片信息
  43.     List<String> materials =  ectractUrlInfo(dto.getContent());
  44.     saveRelativeInfoForContent(materials,wmNews.getId());
  45.  
  46.     //4.不是草稿，保存文章封面图片与素材的关系，如果当前布局是自动，需要匹配封面图片
  47.     saveRelativeInfoForCover(dto,wmNews,materials);
  48.  
  49.     //审核文章
  50.     //        wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
  51.     wmNewsTaskService.addNewsToTask(wmNews.getId(),wmNews.getPublishTime());
  52.  
  53.     return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
  54.  
  55. }
5.3)消费任务进行审核文章

WmNewsTaskService中添加方法


<code class="language-plaintext hljs">/**
 * 消费延迟队列数据
 */
public void scanNewsByTask();</code>

实现

  1.  
  2. @Autowired
  3. private WmNewsAutoScanServiceImpl wmNewsAutoScanService;
  4.  
  5. /**
  6.      * 消费延迟队列数据
  7.      */
  8. @Scheduled(fixedRate = 1000)
  9. @Override
  10. @SneakyThrows
  11. public void scanNewsByTask() {
  12.  
  13.     log.info("文章审核---消费任务执行---begin---");
  14.  
  15.     ResponseResult responseResult = scheduleClient.poll(TaskTypeEnum.NEWS_SCAN_TIME.getTaskType(), TaskTypeEnum.NEWS_SCAN_TIME.getPriority());
  16.     if(responseResult.getCode().equals(200) && responseResult.getData() != null){
  17.         String json_str = JSON.toJSONString(responseResult.getData());
  18.         Task task = JSON.parseObject(json_str, Task.class);
  19.         byte[] parameters = task.getParameters();
  20.         WmNews wmNews = ProtostuffUtil.deserialize(parameters, WmNews.class);
  21.         System.out.println(wmNews.getId()+"-----------");
  22.         wmNewsAutoScanService.autoScanWmNews(wmNews.getId());
  23.     }
  24.     log.info("文章审核---消费任务执行---end---");
  25. }

在WemediaApplication自媒体的引导类中添加开启任务调度注解@EnableScheduling

06kafka及异步通知文章上下架

1)自媒体文章上下架

需求分析

2)kafka概述

消息中间件对比

消息中间件对比-选择建议

消息中间件

建议

Kafka

追求高吞吐量，适合产生大量数据的互联网服务的数据收集业务

RocketMQ

可靠性要求很高的金融互联网领域,稳定性高，经历了多次阿里双11考验

RabbitMQ

性能较好，社区活跃度高，数据量没有那么大，优先选择功能比较完备的RabbitMQ

kafka介绍

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

kafka官网：Apache Kafka

kafka介绍-名词解释

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

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

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

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

3)kafka安装配置

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

  • Docker安装zookeeper

下载镜像：

docker pull zookeeper:3.4.14

创建容器

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

  • Docker安装kafka

下载镜像：

docker pull wurstmeister/kafka:2.12-2.3.1

创建容器

  1. docker run -d --name kafka \
  2. --env KAFKA_ADVERTISED_HOST_NAME=192.168.200.130 \
  3. --env KAFKA_ZOOKEEPER_CONNECT=192.168.200.130:2181 \
  4. --env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.200.130:9092 \
  5. --env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
  6. --env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
  7. --net=host wurstmeister/kafka:2.12-2.3.1

云主机无法使用--net

4)kafka入门

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

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

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

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

（2）生产者发送消息

  1.  
  2. package com.heima.kafka.sample;
  3. import java.util.Properties;
  4.  
  5. /**
  6.  * 生产者
  7.  */
  8. public class ProducerQuickStart {
  9.  
  10.     public static void main(String[] args) {
  11.         //1.kafka的配置信息
  12.         Properties properties = new Properties();
  13.         //kafka的连接地址
  14.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
  15.         //发送失败，失败的重试次数
  16.         properties.put(ProducerConfig.RETRIES_CONFIG,5);
  17.         //消息key的序列化器
  18.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
  19.         //消息value的序列化器
  20.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
  21.  
  22.         //2.生产者对象
  23.         KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);
  24.  
  25.         //封装发送的消息
  26.         ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");
  27.  
  28.         //3.发送消息
  29.         producer.send(record);
  30.  
  31.         //4.关闭消息通道，必须关闭，否则消息发送不成功
  32.         producer.close();
  33.     }
  34.  
  35. }

（3）消费者接收消息

  1.  
  2. package com.heima.kafka.sample;
  3. import java.util.Properties;
  4.  
  5. /**
  6.  * 消费者
  7.  */
  8. public class ConsumerQuickStart {
  9.  
  10.     public static void main(String[] args) {
  11.         //1.添加kafka的配置信息
  12.         Properties properties = new Properties();
  13.         //kafka的连接地址
  14.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
  15.         //消费者组
  16.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");
  17.         //消息的反序列化器
  18.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  19.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  20.  
  21.         //2.消费者对象
  22.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
  23.  
  24.         //3.订阅主题
  25.         consumer.subscribe(Collections.singletonList("itheima-topic"));
  26.  
  27.         //当前线程一直处于监听状态
  28.         while (true) {
  29.             //4.获取消息
  30.             ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
  31.             for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
  32.                 System.out.println(consumerRecord.key());
  33.                 System.out.println(consumerRecord.value());
  34.             }
  35.         }
  36.  
  37.     }
  38.  
  39. }

总结

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

  • 生产者发送消息，多个消费者订阅同一个主题，所有消费者都能收到消息（一对多）多个组

分区机制—topic剖析

5)kafka高可用设计

5.1)集群

  • Kafka 的服务器端由被称为 Broker 的服务进程构成，即一个 Kafka 集群由多个 Broker 组成

  • 这样如果集群中某一台机器宕机，其他机器上的 Broker 也依然能够对外提供服务。这其实就是 Kafka 提供高可用的手段之一

5.2)备份机制(Replication）

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

Kafka 定义了两类副本：

  • 领导者副本（Leader Replica）

  • 追随者副本（Follower Replica）

备份机制—同步方式

ISR（in-sync replica）需要同步复制保存的follower

如果leader失效后，需要选出新的leader，选举的原则如下：

第一：选举时优先从ISR中选定，因为这个列表中follower的数据是与leader同步的

第二：如果ISR列表中的follower都不行了，就只能从其他follower中选取

极端情况，就是所有副本都失效了，这时有两种方案

第一：等待ISR中的一个活过来，选为Leader，数据可靠，但活过来的时间不确定

第二：选择第一个活过来的Replication，不一定是ISR中的，选为leader，以最快速度恢复可用性，但数据不一定完整

6)kafka生产者详解

6.1)发送类型

  • 同步发送

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

  1.  
  2. RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
  3. System.out.println(recordMetadata.offset());

  • 异步发送

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

  1.  
  2. //异步消息发送
  3. producer.send(kvProducerRecord, new Callback() {
  4.     @Override
  5.     public void onCompletion(RecordMetadata recordMetadata, Exception e) {
  6.         if(e != null){
  7.             System.out.println("记录异常信息到日志表中");
  8.         }
  9.         System.out.println(recordMetadata.offset());
  10.     }
  11. });
6.2)参数详解
  • ack确认机制

代码的配置方式：

  1.  
  2. //ack配置  消息确认机制
  3. prop.put(ProducerConfig.ACKS_CONFIG,"all");

参数的选择说明

确认机制

说明

acks=0

生产者在成功写入消息之前不会等待（不需要）任何来自服务器的响应,消息有丢失的风险，但是速度最快

acks=1（默认值）

只要集群Leader节点收到消息，生产者就会收到一个来自服务器的成功响应

acks=all

只有当所有参与赋值的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应

  • retries 重试次数

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

代码中配置方式：

  1.  
  2. //重试次数
  3. prop.put(ProducerConfig.RETRIES_CONFIG,10);
  • 消息压缩

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

代码中配置方式：

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

压缩算法

说明

snappy

占用较少的 CPU， 却能提供较好的性能和相当可观的压缩比， 如果看重性能和网络带宽，建议采用

lz4

占用较少的 CPU， 压缩和解压缩速度较快，压缩比也很客观

gzip

占用较多的 CPU，但会提供更高的压缩比，网络带宽有限，可以使用这种算法

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

7)kafka消费者详解

7.1)消费者组

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

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

  • 所有的消费者都在一个组中，那么这就变成了queue模型 消息队列 一对一

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

7.2)消息有序性

应用场景：

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

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

topic分区中消息只能由消费者组中的唯一一个消费者处理，所以消息肯定是按照先后顺序进行处理的。但是它也仅仅是保证Topic的一个分区顺序处理，不能保证跨分区的消息先后处理顺序。 所以，如果你想要顺序的处理Topic的所有消息，那就只提供一个分区。

7.3)提交和偏移量

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

消费者会往一个叫做_consumer_offset的特殊主题发送消息，消息里包含了每个分区的偏移量。如果消费者发生崩溃或有新的消费者加入群组，就会触发再均衡

正常的情况

如果消费者2挂掉以后，会发生再均衡，消费者2负责的分区会被其他消费者进行消费

再均衡后不可避免会出现一些问题

问题一：

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

问题二：

如果提交的偏移量5大于客户端最后一个消息11的偏移量，那么处于两个偏移量之间的消息将会丢失。

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

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

  • 自动提交偏移量

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

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

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

  • 异步提交

  • 同步和异步组合提交

1.提交当前偏移量（同步提交）

把enable.auto.commit设置为false,让应用程序决定何时提交偏移量。使用commitSync()提交偏移量，commitSync()将会提交poll返回的最新的偏移量，所以在处理完所有记录后要确保调用了commitSync()方法。否则还是会有消息丢失的风险。

只要没有发生不可恢复的错误，commitSync()方法会一直尝试直至提交成功，如果提交失败也可以记录到错误日志里。

  1.  
  2. while (true){
  3.     ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
  4.     for (ConsumerRecord<String, String> record : records) {
  5.         System.out.println(record.value());
  6.         System.out.println(record.key());
  7.         try {
  8.             consumer.commitSync();//同步提交当前最新的偏移量
  9.         }catch (CommitFailedException e){
  10.             System.out.println("记录提交失败的异常："+e);
  11.         }
  12.     }
  13. }
2.异步提交

手动提交有一个缺点，那就是当发起提交调用时应用会阻塞。当然我们可以减少手动提交的频率，但这个会增加消息重复的概率（和自动提交一样）。另外一个解决办法是，使用异步提交的API ：commitAsync()

  1.  
  2. while (true){
  3.     ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
  4.     for (ConsumerRecord<String, String> record : records) {
  5.         System.out.println(record.value());
  6.         System.out.println(record.key());
  7.     }
  8.     consumer.commitAsync(new OffsetCommitCallback() {
  9.         @Override
  10.         public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
  11.             if(e!=null){
  12.                 System.out.println("记录错误的提交偏移量："+ map+",异常信息"+e);
  13.             }
  14.         }
  15.     });
  16. }
3.同步和异步组合提交

异步提交也有个缺点，那就是如果服务器返回提交失败，异步提交不会进行重试。

相比较起来，同步提交会进行重试直到成功或者最后抛出异常给应用。异步提交没有实现重试是因为，如果同时存在多个异步提交，进行重试可能会导致位移覆盖。

举个例子，假如我们发起了一个异步提交commitA，此时的提交位移为2000，随后又发起了一个异步提交commitB且位移为3000；commitA提交失败但commitB提交成功，此时commitA进行重试并成功的话，会将实际上将已经提交的位移从3000回滚到2000，导致消息重复消费。

  1.  
  2. try {
  3.     while (true){
  4.         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
  5.         for (ConsumerRecord<String, String> record : records) {
  6.             System.out.println(record.value());
  7.             System.out.println(record.key());
  8.         }
  9.         consumer.commitAsync();
  10.     }
  11. }catch (Exception e){+
  12.     e.printStackTrace();
  13.     System.out.println("记录错误信息："+e);
  14. }finally {
  15.     try {
  16.         consumer.commitSync();
  17.     }finally {
  18.         consumer.close();
  19.     }
  20. }

8)springboot集成kafka

8.1)入门

1.导入spring-kafka依赖信息

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

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

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

3.消息生产者

  1.  
  2. package com.heima.kafka.controller;
  3. import org.springframework.web.bind.annotation.RestController;
  4.  
  5. @RestController
  6. public class HelloController {
  7.  
  8.     @Autowired
  9.     private KafkaTemplate<String,String> kafkaTemplate;
  10.  
  11.     @GetMapping("/hello")
  12.     public String hello(){
  13.         kafkaTemplate.send("itcast-topic","黑马程序员");
  14.         return "ok";
  15.     }
  16. }

4.消息消费者

  1.  
  2. package com.heima.kafka.listener;
  3.  
  4. import org.springframework.kafka.annotation.KafkaListener;
  5. import org.springframework.stereotype.Component;
  6. import org.springframework.util.StringUtils;
  7.  
  8. @Component
  9. public class HelloListener {
  10.  
  11.     @KafkaListener(topics = "itcast-topic")
  12.     public void onMessage(String message){
  13.         if(!StringUtils.isEmpty(message)){
  14.             System.out.println(message);
  15.         }
  16.     }
  17. }
8.2)传递消息为对象

目前springboot整合后的kafka，因为序列化器是StringSerializer，这个时候如果需要传递对象可以有两种方式

方式一：可以自定义序列化器，对象类型众多，这种方式通用性不强，本章节不介绍

方式二：可以把要传递的对象进行转json字符串，接收消息后再转为对象即可，本项目采用这种方式

  • 发送消息

  1.  
  2. @GetMapping("/hello")
  3. public String hello(){
  4.     User user = new User();
  5.     user.setUsername("xiaowang");
  6.     user.setAge(18);
  7.     
  8.     kafkaTemplate.send("user-topic", JSON.toJSONString(user));
  9.  
  10.     return "ok";
  11. }

  • 接收消息

  1.  
  2. package com.heima.kafka.listener;
  3. import org.springframework.util.StringUtils;
  4.  
  5. @Component
  6. public class HelloListener {
  7.  
  8.     @KafkaListener(topics = "user-topic")
  9.     public void onMessage(String message){
  10.         if(!StringUtils.isEmpty(message)){
  11.             User user = JSON.parseObject(message, User.class);
  12.             System.out.println(user);
  13.         }
  14.     }
  15. }

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

9.1)需求分析

  • 已发表且已上架的文章可以下架

  • 已发表且已下架的文章可以上架

9.2)流程说明
9.3)接口定义

说明

接口路径

/api/v1/news/down_or_up

请求方式

POST

参数

DTO

响应结果

ResponseResult

DTO

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

ResponseResult

9.4)自媒体文章上下架-功能实现

9.4.1)接口定义

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

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

在WmNewsDto中新增enable属性 ，完整的代码如下：

  1.  
  2. package com.heima.model.wemedia.dtos;
  3.  
  4. import lombok.Data;
  5.  
  6. import java.util.Date;
  7. import java.util.List;
  8.  
  9. @Data
  10. public class WmNewsDto {
  11.     
  12.     private Integer id;
  13.      /**
  14.      * 标题
  15.      */
  16.     private String title;
  17.      /**
  18.      * 频道id
  19.      */
  20.     private Integer channelId;
  21.      /**
  22.      * 标签
  23.      */
  24.     private String labels;
  25.      /**
  26.      * 发布时间
  27.      */
  28.     private Date publishTime;
  29.      /**
  30.      * 文章内容
  31.      */
  32.     private String content;
  33.      /**
  34.      * 文章封面类型  0 无图 1 单图 3 多图 -1 自动
  35.      */
  36.     private Short type;
  37.      /**
  38.      * 提交时间
  39.      */
  40.     private Date submitedTime; 
  41.      /**
  42.      * 状态 提交为1  草稿为0
  43.      */
  44.     private Short status;
  45.      
  46.      /**
  47.      * 封面图片列表 多张图以逗号隔开
  48.      */
  49.     private List<String> images;
  50.  
  51.     /**
  52.      * 上下架 0 下架  1 上架
  53.      */
  54.     private Short enable;
  55. }

9.4.2)业务层编写

在WmNewsService新增方法

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

实现方法

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

9.4.3)控制器

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

9.4.4)测试

9.5)消息通知article端文章上下架

9.5.1)在heima-leadnews-common模块下导入kafka依赖

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

9.5.2)在自媒体端的nacos配置中心配置kafka的生产者

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

9.5.3)在自媒体端文章上下架后发送消息

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

常量类：

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

9.5.4)在article端的nacos配置中心配置kafka的消费者

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

9.5.5)在article端编写监听，接收数据

  1.  
  2. package com.heima.article.listener;
  3.  
  4. import com.alibaba.fastjson.JSON;
  5. import com.heima.article.service.ApArticleConfigService;
  6. import lombok.extern.slf4j.Slf4j;
  7. import org.apache.commons.lang3.StringUtils;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.kafka.annotation.KafkaListener;
  10. import org.springframework.stereotype.Component;
  11.  
  12. import java.util.Map;
  13.  
  14. @Component
  15. @Slf4j
  16. public class ArtilceIsDownListener {
  17.  
  18.     @Autowired
  19.     private ApArticleConfigService apArticleConfigService;
  20.  
  21.     @KafkaListener(topics = WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC)
  22.     public void onMessage(String message){
  23.         if(StringUtils.isNotBlank(message)){
  24.             Map map = JSON.parseObject(message, Map.class);
  25.             apArticleConfigService.updateByMap(map);
  26.             log.info("article端文章配置修改，articleId={}",map.get("articleId"));
  27.         }
  28.     }
  29. }

9.5.6)修改ap_article_config表的数据

新建ApArticleConfigService

  1.  
  2. package com.heima.article.service;
  3.  
  4. import com.baomidou.mybatisplus.extension.service.IService;
  5. import com.heima.model.article.pojos.ApArticleConfig;
  6.  
  7. import java.util.Map;
  8.  
  9. public interface ApArticleConfigService extends IService<ApArticleConfig> {
  10.  
  11.     /**
  12.      * 修改文章配置
  13.      * @param map
  14.      */
  15.     public void updateByMap(Map map);
  16. }

实现类：

  1.  
  2. package com.heima.article.service.impl;
  3.  
  4. import com.baomidou.mybatisplus.core.toolkit.Wrappers;
  5. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
  6. import com.heima.article.mapper.ApArticleConfigMapper;
  7. import com.heima.article.service.ApArticleConfigService;
  8. import com.heima.model.article.pojos.ApArticleConfig;
  9. import lombok.extern.slf4j.Slf4j;
  10. import org.springframework.stereotype.Service;
  11. import org.springframework.transaction.annotation.Transactional;
  12.  
  13. import java.util.Map;
  14.  
  15. @Service
  16. @Slf4j
  17. @Transactional
  18. public class ApArticleConfigServiceImpl extends ServiceImpl<ApArticleConfigMapper, ApArticleConfig> implements ApArticleConfigService {
  19.  
  20.  
  21.     /**
  22.      * 修改文章配置
  23.      * @param map
  24.      */
  25.     @Override
  26.     public void updateByMap(Map map) {
  27.         //0 下架 1 上架
  28.         Object enable = map.get("enable");
  29.         boolean isDown = true;
  30.         if(enable.equals(1)){
  31.             isDown = false;
  32.         }
  33.         //修改文章配置
  34.         update(Wrappers.<ApArticleConfig>lambdaUpdate().eq(ApArticleConfig::getArticleId,map.get("articleId")).set(ApArticleConfig::getIsDown,isDown));
  35.  
  36.     }
  37. }
声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号