第5章 Kafka，构建TB级异步消息系统【仿牛客网社区论坛项目】

2024-05-15 08:04
文章标签 系统 构建 项目 异步 消息 kafka tb 社区论坛 仿牛 客网

本文主要是介绍第5章 Kafka，构建TB级异步消息系统【仿牛客网社区论坛项目】，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

第5章 Kafka，构建TB级异步消息系统【仿牛客网社区论坛项目】

  • 前言
  • 推荐
  • 项目总结
    • 第5章 Kafka，构建TB级异步消息系统
      • 1.阻塞队列
      • 2. Kafka入门
      • 3.Spring整合Kafka
      • 4.发送系统通知
      • 5.显示系统通知
  • 最后

前言

2023-4-30 20:42:51

以下内容源自【Java面试项目】
仅供学习交流使用

推荐

仿牛客网项目【面试】

项目总结

第5章 Kafka，构建TB级异步消息系统

1.阻塞队列

2. Kafka入门

3.Spring整合Kafka

4.发送系统通知

package com.jsss.community.event;import com.alibaba.fastjson.JSONObject;
import com.jsss.community.entity.Event;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;@Component
public class EventProducer {@Autowiredprivate KafkaTemplate kafkaTemplate;// 处理事件public void fireEvent(Event event) {// 将事件发布到指定的主题kafkaTemplate.send(event.getTopic(), JSONObject.toJSONString(event));}}
package com.jsss.community.event;import com.alibaba.fastjson.JSONObject;
import com.jsss.community.entity.DiscussPost;
import com.jsss.community.entity.Event;
import com.jsss.community.entity.Message;
import com.jsss.community.service.DiscussPostService;import com.jsss.community.service.ElasticsearchService;
import com.jsss.community.service.MessageService;
import com.jsss.community.util.CommunityConstant;
import com.jsss.community.util.CommunityUtil;import com.qiniu.common.QiniuException;
import com.qiniu.common.Zone;
import com.qiniu.http.Response;
import com.qiniu.storage.Configuration;
import com.qiniu.storage.UploadManager;
import com.qiniu.util.Auth;
import com.qiniu.util.StringMap;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.KafkaListeners;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;import java.io.File;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;@Component
public class EventConsumer implements CommunityConstant {private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);@Autowiredprivate MessageService messageService;@Autowiredprivate DiscussPostService discussPostService;@Autowiredprivate ElasticsearchService elasticsearchService;@Value("${wk.image.command}")private String wkImageCommand;@Value("${wk.image.storage}")private String wkImageStorage;//七牛云@Value("${qiniu.key.access}")private String accessKey;@Value("${qiniu.key.secret}")private String secretKey;@Value("${qiniu.bucket.share.name}")private String shareBucketName;//定时任务@Autowiredprivate ThreadPoolTaskScheduler taskScheduler;//消费通知事件@KafkaListener(topics = {TOPIC_COMMENT, TOPIC_LIKE, TOPIC_FOLLOW})public void handleCommentMessage(ConsumerRecord record) {if (record == null || record.value() == null) {logger.error("消息的内容为空!");return;}Event event = JSONObject.parseObject(record.value().toString(), Event.class);if (event == null) {logger.error("消息格式错误!");return;}// 发送站内通知Message message = new Message();message.setFromId(SYSTEM_USER_ID);message.setToId(event.getEntityUserId());//复用message conversation_id 做 topicmessage.setConversationId(event.getTopic());message.setCreateTime(new Date());//拼接语句  用户nowcoder 评论了你的 帖子 , 点击查看 !Map<String, Object> content = new HashMap<>();content.put("userId", event.getUserId());content.put("entityType", event.getEntityType());content.put("entityId", event.getEntityId());//附加信息 一律存入contentif (!event.getData().isEmpty()) {for (Map.Entry<String, Object> entry : event.getData().entrySet()) {content.put(entry.getKey(), entry.getValue());}}message.setContent(JSONObject.toJSONString(content));//复用message conversation_id 做 topic//92,1,111,like,"{""entityType"":1,""entityId"":237,""postId"":237,""userId"":112}",1,2019-04-13 22:20:58messageService.addMessage(message);}// 消费发帖事件@KafkaListener(topics = {TOPIC_PUBLISH})public void handlePublishMessage(ConsumerRecord record) {if (record == null || record.value() == null) {logger.error("消息的内容为空!");return;}Event event = JSONObject.parseObject(record.value().toString(), Event.class);if (event == null) {logger.error("消息格式错误!");return;}DiscussPost post = discussPostService.findDiscussPostById(event.getEntityId());elasticsearchService.saveDiscussPost(post);}// 消费删帖事件@KafkaListener(topics = {TOPIC_DELETE})public void handleDeleteMessage(ConsumerRecord record) {if (record == null || record.value() == null) {logger.error("消息的内容为空!");return;}Event event = JSONObject.parseObject(record.value().toString(), Event.class);if (event == null) {logger.error("消息格式错误!");return;}elasticsearchService.deleteDiscussPost(event.getEntityId());}// 消费分享事件@KafkaListener(topics = TOPIC_SHARE)public void handleShareMessage(ConsumerRecord record) {if (record == null || record.value() == null) {logger.error("消息的内容为空!");return;}Event event = JSONObject.parseObject(record.value().toString(), Event.class);if (event == null) {logger.error("消息格式错误!");return;}String htmlUrl = (String) event.getData().get("htmlUrl");String fileName = (String) event.getData().get("fileName");String suffix = (String) event.getData().get("suffix");String cmd = wkImageCommand + " --quality 75 "+ htmlUrl + " " + wkImageStorage + "/" + fileName + suffix;try {Runtime.getRuntime().exec(cmd);logger.info("生成长图成功: " + cmd);} catch (IOException e) {logger.error("生成长图失败: " + e.getMessage());}// 启用定时器,监视该图片,一旦生成了,则上传至七牛云.UploadTask task = new UploadTask(fileName, suffix);Future future = taskScheduler.scheduleAtFixedRate(task, 500);task.setFuture(future);}class UploadTask implements Runnable {// 文件名称private String fileName;// 文件后缀private String suffix;// 启动任务的返回值private Future future;// 开始时间private long startTime;// 上传次数private int uploadTimes;public UploadTask(String fileName, String suffix) {this.fileName = fileName;this.suffix = suffix;this.startTime = System.currentTimeMillis();}public void setFuture(Future future) {this.future = future;}@Overridepublic void run() {// 生成失败if (System.currentTimeMillis() - startTime > 30000) {logger.error("执行时间过长,终止任务:" + fileName);future.cancel(true);return;}// 上传失败if (uploadTimes >= 3) {logger.error("上传次数过多,终止任务:" + fileName);future.cancel(true);return;}String path = wkImageStorage + "/" + fileName + suffix;File file = new File(path);if (file.exists()) {logger.info(String.format("开始第%d次上传[%s].", ++uploadTimes, fileName));// 设置响应信息StringMap policy = new StringMap();policy.put("returnBody", CommunityUtil.getJSONString(0));// 生成上传凭证Auth auth = Auth.create(accessKey, secretKey);String uploadToken = auth.uploadToken(shareBucketName, fileName, 3600, policy);// 指定上传机房UploadManager manager = new UploadManager(new Configuration(Zone.zone1()));try {// 开始上传图片Response response = manager.put(path, fileName, uploadToken, null, "image/" + suffix, false);// 处理响应结果JSONObject json = JSONObject.parseObject(response.bodyString());if (json == null || json.get("code") == null || !json.get("code").toString().equals("0")) {logger.info(String.format("第%d次上传失败[%s].", uploadTimes, fileName));} else {logger.info(String.format("第%d次上传成功[%s].", uploadTimes, fileName));future.cancel(true);}} catch (QiniuException e) {logger.info(String.format("第%d次上传失败[%s].", uploadTimes, fileName));}} else {logger.info("等待图片生成[" + fileName + "].");}}}}

5.显示系统通知

   // 得到通知列表@RequestMapping(path = "/notice/list", method = RequestMethod.GET)public String getNoticeList(Model model) {User user = hostHolder.getUser();// 查询评论类通知Message message = messageService.findLatestNotice(user.getId(), TOPIC_COMMENT);if (message != null) {Map<String, Object> messageVO = new HashMap<>();messageVO.put("message", message);//还原commentString content = HtmlUtils.htmlUnescape(message.getContent());Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);messageVO.put("user", userService.findUserById((Integer) data.get("userId")));messageVO.put("entityType", data.get("entityType"));messageVO.put("entityId", data.get("entityId"));messageVO.put("postId", data.get("postId"));int count = messageService.findNoticeCount(user.getId(), TOPIC_COMMENT);messageVO.put("count", count);int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_COMMENT);messageVO.put("unread", unread);model.addAttribute("commentNotice", messageVO);}// 查询点赞类通知message = messageService.findLatestNotice(user.getId(), TOPIC_LIKE);if (message != null) {Map<String, Object> messageVO = new HashMap<>();messageVO.put("message", message);String content = HtmlUtils.htmlUnescape(message.getContent());Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);messageVO.put("user", userService.findUserById((Integer) data.get("userId")));messageVO.put("entityType", data.get("entityType"));messageVO.put("entityId", data.get("entityId"));messageVO.put("postId", data.get("postId"));int count = messageService.findNoticeCount(user.getId(), TOPIC_LIKE);messageVO.put("count", count);int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_LIKE);messageVO.put("unread", unread);model.addAttribute("likeNotice", messageVO);}// 查询关注类通知message = messageService.findLatestNotice(user.getId(), TOPIC_FOLLOW);if (message != null) {Map<String, Object> messageVO = new HashMap<>();messageVO.put("message", message);String content = HtmlUtils.htmlUnescape(message.getContent());Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);messageVO.put("user", userService.findUserById((Integer) data.get("userId")));messageVO.put("entityType", data.get("entityType"));messageVO.put("entityId", data.get("entityId"));int count = messageService.findNoticeCount(user.getId(), TOPIC_FOLLOW);messageVO.put("count", count);int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_FOLLOW);messageVO.put("unread", unread);model.addAttribute("followNotice", messageVO);}// 查询未读消息数量int letterUnreadCount = messageService.findLetterUnreadCount(user.getId(), null);model.addAttribute("letterUnreadCount", letterUnreadCount);int noticeUnreadCount = messageService.findNoticeUnreadCount(user.getId(), null);model.addAttribute("noticeUnreadCount", noticeUnreadCount);return "site/notice";}// 得到通知详情@RequestMapping(path = "/notice/detail/{topic}", method = RequestMethod.GET)public String getNoticeDetail(@PathVariable("topic") String topic, Page page, Model model) {User user = hostHolder.getUser();//分页信息page.setLimit(5);page.setPath("/notice/detail/" + topic);page.setRows(messageService.findNoticeCount(user.getId(), topic));//通知信息List<Message> noticeList = messageService.findNotices(user.getId(), topic, page.getOffset(), page.getLimit());List<Map<String, Object>> noticeVoList = new ArrayList<>();if (noticeList != null) {for (Message notice : noticeList) {Map<String, Object> map = new HashMap<>();// 通知map.put("notice", notice);// 内容String content = HtmlUtils.htmlUnescape(notice.getContent());Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);map.put("user", userService.findUserById((Integer) data.get("userId")));map.put("entityType", data.get("entityType"));map.put("entityId", data.get("entityId"));map.put("postId", data.get("postId"));// 通知作者map.put("fromUser", userService.findUserById(notice.getFromId()));noticeVoList.add(map);}}model.addAttribute("notices", noticeVoList);// 设置已读List<Integer> ids = getLetterIds(noticeList);if (!ids.isEmpty()) {messageService.readMessage(ids);}return "site/notice-detail";}

最后

这篇博客能写好的原因是：站在巨人的肩膀上

这篇博客要写好的目的是：做别人的肩膀

开源：为爱发电

学习：为我而行

这篇关于第5章 Kafka，构建TB级异步消息系统【仿牛客网社区论坛项目】的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/991267。

相关文章

基于 Cursor 开发 Spring Boot 项目详细攻略

基于 Cursor 开发 Spring Boot 项目详细攻略

《基于Cursor开发SpringBoot项目详细攻略》Cursor是集成GPT4、Claude3.5等LLM的VSCode类AI编程工具,支持SpringBoot项目开发全流程,涵盖环境配... 目录cursor是什么？基于 Cursor 开发 Spring Boot 项目完整指南1. 环境准备2. 创建
阅读更多...
Three.js构建一个 3D 商品展示空间完整实战项目

Three.js构建一个 3D 商品展示空间完整实战项目

《Three.js构建一个3D商品展示空间完整实战项目》Three.js是一个强大的JavaScript库,专用于在Web浏览器中创建3D图形,：本文主要介绍Three.js构建一个3D商品展... 目录引言项目核心技术1. 项目架构与资源组织2. 多模型切换、交互热点绑定3. 移动端适配与帧率优化4. 可
阅读更多...
sky-take-out项目中Redis的使用示例详解

sky-take-out项目中Redis的使用示例详解

《sky-take-out项目中Redis的使用示例详解》SpringCache是Spring的缓存抽象层,通过注解简化缓存管理,支持Redis等提供者,适用于方法结果缓存、更新和删除操作,但无法实现... 目录Spring Cache主要特性核心注解1.@Cacheable2.@CachePut3.@Ca
阅读更多...
JWT + 拦截器实现无状态登录系统

JWT + 拦截器实现无状态登录系统

《JWT+拦截器实现无状态登录系统》JWT（JSONWebToken）提供了一种无状态的解决方案：用户登录后,服务器返回一个Token,后续请求携带该Token即可完成身份验证,无需服务器存储会话... 目录✅ 引言 一、JWT 是什么？ 二、技术选型 三、项目结构 四、核心代码实现4.1 添加依赖（pom
阅读更多...
Java Kafka消费者实现过程

Java Kafka消费者实现过程

《JavaKafka消费者实现过程》Kafka消费者通过KafkaConsumer类实现,核心机制包括偏移量管理、消费者组协调、批量拉取消息及多线程处理,手动提交offset确保数据可靠性,自动提交... 目录基础KafkaConsumer类分析关键代码与核心算法2.1 订阅与分区分配2.2 拉取消息2.3
阅读更多...
基于Python实现自动化邮件发送系统的完整指南

基于Python实现自动化邮件发送系统的完整指南

《基于Python实现自动化邮件发送系统的完整指南》在现代软件开发和自动化流程中,邮件通知是一个常见且实用的功能,无论是用于发送报告、告警信息还是用户提醒,通过Python实现自动化的邮件发送功能都能... 目录一、前言：二、项目概述三、配置文件 `.env` 解析四、代码结构解析1. 导入模块2. 加载环
阅读更多...
linux系统上安装JDK8全过程

linux系统上安装JDK8全过程

《linux系统上安装JDK8全过程》文章介绍安装JDK的必要性及Linux下JDK8的安装步骤,包括卸载旧版本、下载解压、配置环境变量等,强调开发需JDK,运行可选JRE,现JDK已集成JRE... 目录为什么要安装jdk？1.查看linux系统是否有自带的jdk：2.下载jdk压缩包2.解压3.配置环境
阅读更多...
SpringBoot通过main方法启动web项目实践

SpringBoot通过main方法启动web项目实践

《SpringBoot通过main方法启动web项目实践》SpringBoot通过SpringApplication.run()启动Web项目,自动推断应用类型,加载初始化器与监听器,配置Spring... 目录1. 启动入口：SpringApplication.run()2. SpringApplicat
阅读更多...
Python异步编程之await与asyncio基本用法详解

Python异步编程之await与asyncio基本用法详解

《Python异步编程之await与asyncio基本用法详解》在Python中,await和asyncio是异步编程的核心工具,用于高效处理I/O密集型任务（如网络请求、文件读写、数据库操作等）,接... 目录一、核心概念二、使用场景三、基本用法1. 定义协程2. 运行协程3. 并发执行多个任务四、关键
阅读更多...
Python利用PySpark和Kafka实现流处理引擎构建指南

Python利用PySpark和Kafka实现流处理引擎构建指南

《Python利用PySpark和Kafka实现流处理引擎构建指南》本文将深入解剖基于Python的实时处理黄金组合：Kafka（分布式消息队列）与PySpark（分布式计算引擎）的化学反应,并构建一... 目录引言：数据洪流时代的生存法则第一章 Kafka：数据世界的中央神经系统消息引擎核心设计哲学高吞吐
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2