kafka及异步通知文章上下架

1)自媒体文章上下架

需求分析

2)kafka概述

消息中间件对比

特 性

ActiveMQ

RabbitMQ

RocketMQ

Kafka

开 发 语 言

java

erlang

java

scala

单 机 吞 吐 量

万级

万级

10万级

100万级

时 效 性

ms

us

ms

ms级以内

可 用 性

高（主从）

高（主从）

非常高（分布 式）

非常高（分布 式）

功 能 特 性

成熟的产品、较全的 文档、各种协议支持 好

并发能力强、 性能好、延迟 低

MQ功能比较 完善，扩展性 佳

只支持主要的MQ功能， 主要应用于大数据领域

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

消息中间 件

建议

Kafka

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

RocketMQ

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

RabbitMQ

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

kafka介绍

Kafka 是一个分布式流媒体平台,类似于消息队列或企业消息传递系统。kafka官网：http://kafka.apach e.org/

kafka介绍-名词解释

  • producer：发布消息的对象称之为主题生产者（Kafka topic producer）
  • topic：Kafka将消息分门别类，每一类的消息称之为一个主题（Topic）
  • consumer：订阅消息并处理发布的消息的对象称之为主题消费者（consumers）
  • broker：已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理（Broker）。消费者可以订阅一个或多个主题（topic），并从Broker拉数据，从而消费这些已发布的消息。

3)kafka安装配置

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

https://blog.csdn.net/m0_70325779/article/details/137248462

4)kafka入门

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

（1）创建kafka-demo项目，导入依赖
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId>
</dependency>
（2）生产者发送消息
/*** 生产者*/
public class ProducerQuickStart {public static void main(String[] args) throws ExecutionException, 
InterruptedException {//1.kafka链接配置信息Properties prop = new Properties();//kafka链接地址prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");//key和value的序列化prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.se
rialization.StringSerializer");prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.
serialization.StringSerializer");//2.创建kafka生产者对象KafkaProducer<String,String> producer = new KafkaProducer<String,String>
(prop);//3.发送消息/*** 第一个参数 ：topic* 第二个参数：消息的key* 第三个参数：消息的value*/ProducerRecord<String,String> kvProducerRecord = new 
ProducerRecord<String,String>("topic-first","key-001","hello kafka");//同步发送消息RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();System.out.println(recordMetadata.offset());//4.关闭消息通道 必须要关闭，否则消息发送不成功producer.close();}
}
（3）消费者接收消息
/*** 消费者*/
public class ConsumerQuickStart {public static void main(String[] args) {//1.kafka的配置信息Properties prop = new Properties();//链接地址prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");//key和value的反序列化器prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
"org.apache.kafka.common.serialization.StringDeserializer");prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
"org.apache.kafka.common.serialization.StringDeserializer");//设置消费者组prop.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");//2.创建消费者对象KafkaConsumer<String, String> consumer = new KafkaConsumer<String, 
String>(prop);//3.订阅主题consumer.subscribe(Collections.singletonList("topic-first"));//4.拉取消息while (true) {ConsumerRecords<String, String> consumerRecords = 
consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> consumerRecord : 
consumerRecords) {System.out.println(consumerRecord.key());System.out.println(consumerRecord.value());}}}
}
使用情景：
  • 生产者发送消息，多个消费者订阅同一个主题(多个消费者都是一个组)只能有一个消费者收到消息（一对一）
  • 生产者发送消息，多个消费者订阅同一个主题(多个消费者不是一个组)所有消费者都能收到消息（一对多）

springboot集成kafka入门

1.导入spring-kafka依赖信息
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- kafkfa --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><exclusions><exclusion><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId></dependency>
</dependencies>
2.在resources下创建文件application.yml
server:port: 9991
spring:application:name: kafka-demokafka:bootstrap-servers: 192.168.200.130:9092producer:retries: 10key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:group-id: ${spring.application.name}-testkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: 
org.apache.kafka.common.serialization.StringDeserializer
3.消息生产者
4.消息消费者

传递消息为对象

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

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

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

  • 发送消息
  • 接收消息

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

需求分析

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

流程说明

接口定义

说明

接口路径

/api/v1/news/down_or_up

请求方式

POST

参数

DTO

响应结果

ResponseResult

DTO
@Data
public class WmNewsDto {private Integer id;/*** 是否上架 0 下架 1 上架*/private Short enable;}
ResponseResult

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

接口定义

water-wemedia工程

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

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

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

@Data
public class WmNewsDto {private Integer id;/*** 标题*/private String title;/*** 频道id*/private Integer channelId;/*** 标签*/private String labels;/*** 发布时间*/private Date publishTime;/*** 文章内容*/private String content;/*** 文章封面类型 0 无图 1 单图 3 多图 -1 自动*/private Short type;/*** 提交时间*/private Date submitedTime; /*** 状态 提交为1 草稿为0*/private Short status;/*** 封面图片列表 多张图以逗号隔开*/private List<String> images;/*** 上下架 0 下架 1 上架*/private Short enable;
}
业务层编写

在WmNewsService新增方法

/*** 文章的上下架* @param dto* @return*/
public ResponseResult downOrUp(WmNewsDto dto);
实现方法
/*** 文章的上下架* @param dto* @return*/
@Override
public ResponseResult downOrUp(WmNewsDto dto) {//1.检查参数if(dto.getId() == null){return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID);}//2.查询文章WmNews wmNews = getById(dto.getId());if(wmNews == null){return ResponseResult.errorResult(AppHttpCodeEnum.DATA_NOT_EXIST,"文章不
存在");}//3.判断文章是否已发布if(!wmNews.getStatus().equals(WmNews.Status.PUBLISHED.getCode())){return ResponseResult.errorResult(AppHttpCodeEnum.PARAM_INVALID,"当前文章
不是发布状态，不能上下架");}//4.修改文章enableif(dto.getEnable() != null && dto.getEnable() > -1 && dto.getEnable() < 2){update(Wrappers.
<WmNews>lambdaUpdate().set(WmNews::getEnable,dto.getEnable())
.eq(WmNews::getId,wmNews.getId()));}return ResponseResult.okResult(AppHttpCodeEnum.SUCCESS);
}
控制器
@PostMapping("/down_or_up")
public ResponseResult downOrUp(@RequestBody WmNewsDto dto){return wmNewsService.downOrUp(dto);
}
测试
消息通知article端文章上下架
在water-common模块下导入kafka依赖
<!-- kafkfa -->
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId>
</dependency>
在自媒体端的nacos配置中心配置kafka的生产者
spring:kafka:bootstrap-servers: localhost:9092producer:retries: 10key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer
在自媒体端文章上下架后发送消息
//发送消息，通知article端修改文章配置
if(wmNews.getArticleId() != null){Map<String,Object> map = new HashMap<>();map.put("articleId",wmNews.getArticleId());map.put("enable",dto.getEnable());kafkaTemplate.send(WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC,JSON.toJSONS
tring(map));
}
常量类：
public class WmNewsMessageConstants {public static final String 
WM_NEWS_UP_OR_DOWN_TOPIC="wm.news.up.or.down.topic";
}
在article端的nacos配置中心配置kafka的消费者
spring:kafka:bootstrap-servers: localhost:9092consumer:group-id: ${spring.application.name}key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: 
org.apache.kafka.common.serialization.StringDeserializer
在article端编写监听，接收数据
@Component
@Slf4j
public class ArtilceIsDownListener {@Autowiredprivate ApArticleConfigService apArticleConfigService;@KafkaListener(topics = WmNewsMessageConstants.WM_NEWS_UP_OR_DOWN_TOPIC)public void onMessage(String message){if(StringUtils.isNotBlank(message)){Map map = JSON.parseObject(message, Map.class);apArticleConfigService.updateByMap(map);log.info("article端文章配置修改，articleId={}",map.get("articleId"));}}
}

修改ap_article_config表的数据新建ApArticleConfigService

public interface ApArticleConfigService extends IService<ApArticleConfig> {/*** 修改文章配置* @param map*/public void updateByMap(Map map);
}
实现类：
@Service
@Slf4j
@Transactional
public class ApArticleConfigServiceImpl extends 
ServiceImpl<ApArticleConfigMapper, ApArticleConfig> implements 
ApArticleConfigService {/*** 修改文章配置* @param map*/@Overridepublic void updateByMap(Map map) {//0 下架 1 上架Object enable = map.get("enable");boolean isDown = true;if(enable.equals(1)){isDown = false;}//修改文章配置update(Wrappers.
<ApArticleConfig>lambdaUpdate().eq(ApArticleConfig::getArticleId,map.get("articl
eId")).set(ApArticleConfig::getIsDown,isDown));}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/51952.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

如何从 Bak 文件中恢复 SQL数据库？（3种方法）

如何从 Bak 文件中恢复 SQL数据库？（3种方法）

如何从 .bak 文件恢复 SQL数据库&#xff1f; 在数据库管理和维护过程中&#xff0c;数据的安全性和完整性至关重要。备份文件&#xff08;.bak 文件&#xff09;是 SQL Server 中常用的数据库备份格式&#xff0c;它包含了数据库的完整副本&#xff0c;用于在数据丢失、系统故…
阅读更多...
4. 第一个3D案例—创建3D场景

4. 第一个3D案例—创建3D场景

入门Three.js的第一步&#xff0c;就是认识场景Scene、相机Camera、渲染器Renderer三个基本概念&#xff0c;接下来&#xff0c;咱们通过三小节课&#xff0c;大家演示“第一个3D案例”完成实现过程。 学习建议&#xff1a;只要你能把第一个3D案例搞明白&#xff0c;后面学习就…
阅读更多...
SEO之网站结构优化（十四-内部链接及权重分配3）

SEO之网站结构优化（十四-内部链接及权重分配3）

初创企业搭建网站的朋友看1号文章&#xff1b;想学习云计算&#xff0c;怎么入门看2号文章谢谢支持&#xff1a; 1、我给不会敲代码又想搭建网站的人建议 2、“新手上云”能够为你开启探索云世界的第一步 博客&#xff1a;阿幸SEO~探索搜索排名之道 7、锚文字分布及变化 前面…
阅读更多...
部署 Web 项目到 Linux，可以使他人也访问项目的方法

部署 Web 项目到 Linux，可以使他人也访问项目的方法

目录 一、环境配置 二、建构项目并打包 三、上传Jar包到服务器, 并运行 3.1 上传Jar包 3.2 运行 jar 包 3.3 开放端口号 四、其他问题 4.1 运行异常问题 4.2 杀掉进程 五、总结 一、环境配置 如果本地项目是SpringBoot项目&#xff0c;使用的数据库是MySQL&#xff…
阅读更多...
APP 数据抓取 - Charles 抓包工具的使用（Charles 端口配置、CA 证书配置、Charles Android 模拟器配置）

APP 数据抓取 - Charles 抓包工具的使用（Charles 端口配置、CA 证书配置、Charles Android 模拟器配置）

前言说明 此文章是我在学习 Charles APP 抓包时编写&#xff0c;内容都是亲测有效&#xff0c;文章内容也有参考其他人&#xff0c;参考文章如下&#xff1a; Android 手机使用 charles 抓 https 请求&#xff08;保姆级教程&#xff09;网易 mumu 模拟器安装下载 charles 的…
阅读更多...
计算机网络（八股文）

计算机网络（八股文）

这里写目录标题 计算机网络一、网络分层模型1. TCP/IP四层架构和OSI七层架构⭐️⭐️⭐️⭐️⭐️2. 为什么网络要分层&#xff1f;⭐️⭐️⭐️3. 各层都有那些协议&#xff1f;⭐️⭐️⭐️⭐️ 二、HTTP【重要】1. http状态码&#xff1f;⭐️⭐️⭐️2. 从输入URL到页面展示…
阅读更多...
2024年9月3日嵌入式学习

2024年9月3日嵌入式学习

数据结构 1定义 一组用来保存一种或者多种特定关系的数据的集合&#xff08;组织和存储数据&#xff09; 程序的设计&#xff1a;将现实中大量而复杂的问题以特定的数据类型和特定的存储结构存储在内存中&#xff0c; 并在此基础上实现某个特定的功能的操作&am…
阅读更多...
「Python程序设计」条件控制：if-elif-else语句

「Python程序设计」条件控制：if-elif-else语句

我们在进行程序设计的过程中&#xff0c;基本上遵循的过程是&#xff0c;找出变量和常量&#xff0c;通过python编程语言&#xff0c;设置变量和常量&#xff0c;以及考虑是否需要赋予初始值。 设计变量和常量&#xff0c;其实就是为了模拟和计算我们的现实世界中&#xff0c;…
阅读更多...
学习笔记--Docker

学习笔记--Docker

安装 1.卸载旧版 首先如果系统中已经存在旧的Docker&#xff0c;则先卸载&#xff1a; yum remove docker \docker-client \docker-client-latest \docker-common \docker-latest \docker-latest-logrotate \docker-logrotate \docker-engine 2.配置Docker的yum库 首先要安…
阅读更多...
IntelliJ IDEA 自定义字体大小

IntelliJ IDEA 自定义字体大小

常用编程软件自定义字体大全首页 文章目录 前言具体操作1. 打开设置对话框2. 设置编辑器字体3. 设置编译软件整体字体 前言 IntelliJ IDEA 自定义字体大小&#xff0c;统一设置为 JetBrains Mono 具体操作 【File】>【Settings...】>【Editor】>【Font】 统一设置…
阅读更多...
C++：list篇

C++：list篇

前言: 观看C的list前需要对链表有一些了解&#xff0c;如C语言的链表结构。本片仅介绍list容器中常用的接口函数概念以及使用。 list的概念&#xff1a; 简而言之&#xff0c;C的list是一个双向带哨兵位的链表容器模板 list的构造&#xff1a; 1.list():默认构造 2.li…
阅读更多...
认识git和git的基本使用,本地仓库，远程仓库和克隆远程仓库

认识git和git的基本使用,本地仓库，远程仓库和克隆远程仓库

本地仓库 #安装git https://git-scm.com/download/win #git是什么&#xff1f;有什么用&#xff1f; git相当于一个版本控制系统&#xff0c;版本控制是一种记录一个或若干文件内容变化&#xff0c;以便将来查阅特定版本修订情况的系统。 作用: 记录&#xff08;项目&#…
阅读更多...
物联网（IoT）支持的小型水处理厂实时硬件在环（HIL）仿真

物联网（IoT）支持的小型水处理厂实时硬件在环（HIL）仿真

这篇论文的标题是《Real-Time Hardware-In-The-Loop Simulation of IoT-Enabled Mini Water Treatment Plant》&#xff0c;作者是 Mohamad Taib Miskon 等人&#xff0c;发表在 2024 年 IEEE 自动控制与智能系统国际会议&#xff08;I2CACIS&#xff09;上。以下是该论文的主要…
阅读更多...
【Agent】Agent Q: Advanced Reasoning and Learning for Autonomous AI Agents

【Agent】Agent Q: Advanced Reasoning and Learning for Autonomous AI Agents

1、问题背景 传统的训练Agent方法是在静态数据集上进行监督预训练&#xff0c;这种方式对于要求Agent能够自主的在动态环境中可进行复杂决策的能力存在不足。例如&#xff0c;要求Agent在web导航等动态设置中执行复杂决策。 现有的方式是用高质量数据进行微调来增强Agent在动…
阅读更多...
vector 常见函数

vector 常见函数

目录 一.vector 构造函数 二 . Iterators 迭代器&#xff08;random access iterator&#xff09; 三.Capacity: 空间 3.1 resize 3.2 reserve 四.Element access: 元素访问 方式 4.1 operator[] 类似于数组的 [] 4.2 front 和back 五.Modifiers: 六.vector 的 二…
阅读更多...
18、Gemini-Pentest-v2

18、Gemini-Pentest-v2

难度 中 目标 root权限 一个flag 靶机启动环境为VMware kali 192.168.152.56 靶机 192.168.152.63 信息收集 web测试 访问80端口 上面介绍了一下这个系统是一个内部系统&#xff0c;让员工查看他们的个人资料还可以导出为PDF 页面还有一个链接是UserList可以访问但是页面什…
阅读更多...
ES数据写入过程

ES数据写入过程

1. 写入请求 当一个写入请求&#xff08;如 Index、Update 或 Delete 请求&#xff09;通过REST API发送到Elasticsearch时&#xff0c;通常包含一个文档的内容&#xff0c;以及该文档的索引和ID。 2. 请求路由 协调节点&#xff1a;首先&#xff0c;请求会到达一个协调节点…
阅读更多...
微服务日常总结

微服务日常总结

1.当我们在开发中&#xff0c;需要连接多个库时&#xff0c;可以在yml中进行配置。 当在查询的时候&#xff0c;跨库时&#xff0c;需要通过DS 注解来指定&#xff0c;需要yml配置需要保持一致。 2. 当我们想把数据存入到clob类型中&#xff0c;需要再字段 的占位符后面加上j…
阅读更多...
Zookeeper 官方示例2-SyncPrimitive 代码解读（二）

Zookeeper 官方示例2-SyncPrimitive 代码解读（二）

测试命令 java jar .\ZookeeperDemo-0.0.1-SNAPSHOT.jar bTest 192.168.206.100:2181 2 1. Barrier&#xff08;阻塞原语&#xff09; 1.1 概念 [!quote] A barrier is a primitive that enables a group of processes to synchronize the beginning and the end of a comput…
阅读更多...
vulhub xxe靶机

vulhub xxe靶机

先用御剑扫描出ip然后进入网页 进入robots.txt里面会发现俩个目录然后我们进去xxe里面 进入xxe页面进行登录&#xff0c;burp抓包 然后进入重放器 可以看到关于密码和用户名的是xml,那么就可以考虑用xxe注入 <?xml version"1.0" ?> <!DOCTYPE r [ <!…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023