第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。

相关文章

Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

Python FastAPI+Celery+RabbitMQ实现分布式图片水印处理系统

《PythonFastAPI+Celery+RabbitMQ实现分布式图片水印处理系统》这篇文章主要为大家详细介绍了PythonFastAPI如何结合Celery以及RabbitMQ实现简单的分布式... 实现思路FastAPI 服务器Celery 任务队列RabbitMQ 作为消息代理定时任务处理完整
阅读更多...
Linux系统中卸载与安装JDK的详细教程

Linux系统中卸载与安装JDK的详细教程

《Linux系统中卸载与安装JDK的详细教程》本文详细介绍了如何在Linux系统中通过Xshell和Xftp工具连接与传输文件,然后进行JDK的安装与卸载,安装步骤包括连接Linux、传输JDK安装包... 目录1、卸载1.1 linux删除自带的JDK1.2 Linux上卸载自己安装的JDK2、安装2.1
阅读更多...
SpringKafka消息发布之KafkaTemplate与事务支持功能

SpringKafka消息发布之KafkaTemplate与事务支持功能

《SpringKafka消息发布之KafkaTemplate与事务支持功能》通过本文介绍的基本用法、序列化选项、事务支持、错误处理和性能优化技术,开发者可以构建高效可靠的Kafka消息发布系统,事务支... 目录引言一、KafkaTemplate基础二、消息序列化三、事务支持机制四、错误处理与重试五、性能优
阅读更多...
SpringIntegration消息路由之Router的条件路由与过滤功能

SpringIntegration消息路由之Router的条件路由与过滤功能

《SpringIntegration消息路由之Router的条件路由与过滤功能》本文详细介绍了Router的基础概念、条件路由实现、基于消息头的路由、动态路由与路由表、消息过滤与选择性路由以及错误处理... 目录引言一、Router基础概念二、条件路由实现三、基于消息头的路由四、动态路由与路由表五、消息过滤
阅读更多...
一文教你如何将maven项目转成web项目

一文教你如何将maven项目转成web项目

《一文教你如何将maven项目转成web项目》在软件开发过程中,有时我们需要将一个普通的Maven项目转换为Web项目,以便能够部署到Web容器中运行,本文将详细介绍如何通过简单的步骤完成这一转换过程... 目录准备工作步骤一：修改​​pom.XML​​1.1 添加​​packaging​​标签1.2 添加
阅读更多...
一文详解如何从零构建Spring Boot Starter并实现整合

一文详解如何从零构建Spring Boot Starter并实现整合

《一文详解如何从零构建SpringBootStarter并实现整合》SpringBoot是一个开源的Java基础框架,用于创建独立、生产级的基于Spring框架的应用程序,：本文主要介绍如何从... 目录一、Spring Boot Starter的核心价值二、Starter项目创建全流程2.1 项目初始化（
阅读更多...
tomcat多实例部署的项目实践

tomcat多实例部署的项目实践

《tomcat多实例部署的项目实践》Tomcat多实例是指在一台设备上运行多个Tomcat服务,这些Tomcat相互独立,本文主要介绍了tomcat多实例部署的项目实践,具有一定的参考价值,感兴趣的可... 目录1.创建项目目录，测试文China编程件2js.创建实例的安装目录3.准备实例的配置文件4.编辑实例的
阅读更多...
Python 中的异步与同步深度解析(实践记录)

Python 中的异步与同步深度解析(实践记录)

《Python中的异步与同步深度解析(实践记录)》在Python编程世界里,异步和同步的概念是理解程序执行流程和性能优化的关键,这篇文章将带你深入了解它们的差异,以及阻塞和非阻塞的特性,同时通过实际... 目录python中的异步与同步：深度解析与实践异步与同步的定义异步同步阻塞与非阻塞的概念阻塞非阻塞同步
阅读更多...
使用Java实现通用树形结构构建工具类

使用Java实现通用树形结构构建工具类

《使用Java实现通用树形结构构建工具类》这篇文章主要为大家详细介绍了如何使用Java实现通用树形结构构建工具类,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录完整代码一、设计思想与核心功能二、核心实现原理1. 数据结构准备阶段2. 循环依赖检测算法3. 树形结构构建4. 搜索子
阅读更多...
Java 中实现异步的多种方式

Java 中实现异步的多种方式

《Java中实现异步的多种方式》文章介绍了Java中实现异步处理的几种常见方式,每种方式都有其特点和适用场景,通过选择合适的异步处理方式,可以提高程序的性能和可维护性,感兴趣的朋友一起看看吧... 目录1. 线程池（ExecutorService）2. CompletableFuture3. ForkJoi
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2