仿牛客社区开发--系统通知模块

Java多线程间通信的方式：如果是采用synchronized保证线程安全的话，使用wait()、notify()、notifyAll()就可以完成多线程之间的通信，如果是采用lock保证线程安全的话，使用await()，signal()，signalAll()，
除此之外，还有BlockingQueue，最经典的就是生产者消费者模型，当生产者线程生产线程时，如果队列已满，则生产者线程被阻塞；当消费者线程使用线程时，如果队列为空，则消费者线程被阻塞。

有关阻塞队列的示例：
test包下新建生产者、消费者类，实现Runnable接口，重写run()方法，生产者线程生产100个数据，放入阻塞队列，消费者线程从阻塞队列中取出线程，并进行测试

public class BlockingQueueTests {
    public static void main(String[] args) {
        BlockingQueue queue = new ArrayBlockingQueue(10); //队列的容量默认为10，即最多只能存10个数
        new Thread(new Producer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
        new Thread(new Consumer(queue)).start();
    }
}

class Producer implements Runnable {

    private BlockingQueue queue;

    public Producer(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < 100; i++) {
                Thread.sleep(20);
                queue.put(i);
                System.out.println(Thread.currentThread().getName() + "生产:" + queue.size());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

class Consumer implements Runnable {
    private BlockingQueue queue;

    public Consumer(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                Thread.sleep(new Random().nextInt(1000));
                queue.take();
                System.out.println(Thread.currentThread().getName() + "消费：" + queue.size());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

修改zookeeper.properties下的dataDir路径
修改server.properties下的log.dirs
启动服务，先进入D:workkafka_2.12-2.2.2
启动 zookeeper:
binwindowszookeeper-server-start.bat configzookeeper.properties
启动kafka:
binwindowskafka-server-start.bat configserver.properties

maven搜索spring kafka，添加相关的依赖

		
			org.springframework.kafka
			spring-kafka
		

application.properties添加相关配置

# KafkaProperties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=community-consumer-group
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=3000

测试类代码，生产者发消息是主动去调的，消费者处理消息是被动自动去调的

@RunWith(SpringRunner.class)
@SpringBootTest
@ContextConfiguration(classes = CommunityApplication.class)
public class KafkaTests {
    @Autowired
    private KafkaProducer kafkaProducer;

    @Test
    public void testKafka() {
        kafkaProducer.sendMessage("test", "你好");
        kafkaProducer.sendMessage("test", "在吗");

        try {
            Thread.sleep(1000 * 10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

@Component
class KafkaProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void sendMessage(String topic, String content) {
        kafkaTemplate.send(topic, content);
    }
}

@Component
class KafkaConsumer {
    @KafkaListener(topics = {"test"})
    public void handleMessage(ConsumerRecord record) {
        System.out.println(record.value());
    }
}

发送系统通知是非常频繁的行为，重点要考虑性能的问题，为了保证性能，这里用到kafka消息队列，一旦事件发生，将其包装成消息，放进消息队列，生产者线程就可以处理下一条消息，后续的业务由消费者处理。生产者消费者处理业务是可以同时进行的，这种方式称为异步。
采用事件驱动的方式，评论、点赞、关注都是一个事件，以事件为目的进一步封装，首先封装好事件对象，然后开发事件的生产者、消费者。

entity包下新建Event实体类

public class Event {
    private String topic;
    private int userId;
    private int entityType;
    private int entityId;
    private int entityUserId;
    private Map data = new HashMap<>(); //事件对象要具有通用性，在处理其他的事件中也需要记录，但是无法预知数据类型，所以定义一个map

    public String getTopic() {
        return topic;
    }

    public Event setTopic(String topic) {
        this.topic = topic;
        return this;
    }

    public int getUserId() {
        return userId;
    }

    public Event setUserId(int userId) {
        this.userId = userId;
        return this;
    }

    public int getEntityType() {
        return entityType;
    }

    public Event setEntityType(int entityType) {
        this.entityType = entityType;
        return this;
    }

    public int getEntityId() {
        return entityId;
    }

    public Event setEntityId(int entityId) {
        this.entityId = entityId;
        return this;
    }

    public int getEntityUserId() {
        return entityUserId;
    }

    public Event setEntityUserId(int entityUserId) {
        this.entityUserId = entityUserId;
        return this;
    }

    public Map getData() {
        return data;
    }

    public Event setData(String key, Object value) {
        this.data.put(key, value);
        return this;
    }
}

新建event包，开发事件的生产者与消费者

@Component
public class EventProducer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    // 处理事件
    public void fireEvent(Event event) {
        // 将事件发布到指定的主题
        kafkaTemplate.send(event.getTopic(), JSONObject.toJSONString(event));
    }
}

@Component
public class EventConsumer implements CommunityConstant {
	private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);

    @Autowired
    private MessageService messageService;

	@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.setConversationId(event.getTopic());
        message.setCreateTime(new Date());

        Map content = new HashMap<>();
        content.put("userId", event.getUserId());
        content.put("entityType", event.getEntityType());
        content.put("entityId", event.getEntityId());

        if (!event.getData().isEmpty()) {
            for (Map.Entry entry : event.getData().entrySet()) {
                content.put(entry.getKey(), entry.getValue());
            }
        }
		message.setContent(JSONObject.toJSONString(content));
        messageService.addMessage(message);
	}   

	
}

处理CommentController、LikeController、FollowController中的事件请求

@Controller
@RequestMapping("/comment")
public class CommentController {

    @Autowired
    private CommentService commentService;

    @Autowired
    private HostHolder hostHolder;

    @Autowired
    private EventProducer eventProducer;

    @RequestMapping(path = "/add/{discussPostId}", method = RequestMethod.POST)
    public String addComment(@PathVariable("discussPostId") int discussPostId, Comment comment) {
        comment.setUserId(hostHolder.getUser().getId());
        comment.setStatus(0);
        comment.setCreateTime(new Date());
        commentService.addComment(comment);

        // 触发评论事件
        Event event = new Event()
                .setTopic(TOPIC_COMMENT)
                .setUserId(hostHolder.getUser().getId())
                .setEntityType(comment.getEntityType())
                .setEntityId(comment.getEntityId())
                .setData("postId", discussPostId);
        if (comment.getEntityType() == ENTITY_TYPE_POST) {
            DiscussPost target = discussPostService.findDiscussPostById(comment.getEntityId());
            event.setEntityUserId(target.getUserId());
        } else if (comment.getEntityType() == ENTITY_TYPE_COMMENT) {
            Comment target = commentService.findCommentById(comment.getEntityId());
            event.setEntityUserId(target.getUserId());
        }
        eventProducer.fireEvent(event)；

        return "redirect:/discuss/detail/" + discussPostId;
    }
}

CommentMapper中新增selectCommentById方法

@Mapper
public interface CommentMapper {
    List selectCommentsByEntity(int entityType, int entityId, int offset, int limit);

    int selectCountByEntity(int entityType, int entityId);

    int insertComment(Comment comment);

    Comment selectCommentById(int id);
}

comment-mapper中书写sql语句

    
        select 
        from comment
        where id = #{id}

    
 
commentService中新增findCommentById方法
 
    public Comment findCommentById(int id) {
        return commentMapper.selectCommentById(id);
    }
 
LikeController中处理点赞事件
 
   @RequestMapping(path = "/like", method = RequestMethod.POST)
    @ResponseBody
    public String like(int entityType, int entityId, int entityUserId, int postId) {
        User user = hostHolder.getUser();

        // 点赞
        likeService.like(user.getId(), entityType, entityId, entityUserId);

        // 数量
        long likeCount = likeService.findEntityLikeCount(entityType, entityId);
        // 状态
        int likeStatus = likeService.findEntityLikeStatus(user.getId(), entityType, entityId);
        // 返回的结果，统一封装到map里返回
        Map map = new HashMap<>();
        map.put("likeCount", likeCount);
        map.put("likeStatus", likeStatus);

        // 触发点赞事件
        if (likeStatus == 1) {
            Event event = new Event()
                    .setTopic(TOPIC_LIKE)
                    .setUserId(hostHolder.getUser().getId())
                    .setEntityType(entityType)
                    .setEntityId(entityId)
                    .setEntityUserId(entityUserId)
                    .setData("postId", postId);
            eventProducer.fireEvent(event);
        }
        return CommunityUtil.getJSONString(0, null, map); //正确返回0，提示返回null
    }
 
FollowController中处理关注事件
 
    @RequestMapping(path = "/follow", method = RequestMethod.POST)
    @ResponseBody
    public String follow(int entityType, int entityId) {
        User user = hostHolder.getUser();

        followService.follow(user.getId(), entityType, entityId);

        // 触发关注事件
        Event event = new Event()
                .setTopic(TOPIC_FOLLOW)
                .setUserId(hostHolder.getUser().getId())
                .setEntityType(entityType)
                .setEntityId(entityId)
                .setEntityUserId(entityId);
        eventProducer.fireEvent(event);

        return CommunityUtil.getJSONString(0, "已关注!");
    }
 
修改discuss-detail.html
 修改discuss.js文件
 
function like(btn, entityType, entityId, entityUserId, postId) {
    $.post(
        CONTEXT_PATH + "/like",
        {"entityType": entityType, "entityId": entityId, "entityUserId": entityUserId, "postId": postId},
        function (data) {
            data = $.parseJSON(data);
            if (data.code == 0) {
                $(btn).children("i").text(data.likeCount);
                $(btn).children("b").text(data.likeStatus == 1 ? '已赞' : "赞");
            } else {
                alert(data.msg);
            }
        }
    );
}
 
5.显示系统通知 

 私信显示的不同的会话，通知显示的是点赞、关注、评论三类通知，只显示最新的消息。
 
5.1 开发系统通知 
数据访问层：MessageMapper里新增查询某一主题下最新通知、包含通知数量、未读通知数量的方法
 
@Mapper
public interface MessageMapper {

    //查询当前用户的会话列表，针对每个会话只返回一条最新的私信
    List selectConversations(int userId, int offset, int limit);

    //查询当前用户的会话数量
    int selectConversationCount(int userId);

    //查询某个会话所包含的私信列表
    List selectLetters(String conversationId, int offset, int limit);

    //查询某个会话所包含的私信数量
    int selectLetterCount(String conversationId);

    //查询未读私信的数量
    int selectLetterUnreadCount(int userId, String conversationId);

    //新增消息
    int insertMessage(Message message);

    //修改消息的状态
    int updateStatus(List ids, int status);

    //查询某个主题下最新的通知
    Message selectLatestNotice(int userId, String topic);

    //查询某个主题所包含的通知数量
    int selectNoticeCount(int userId, String topic);

    //查询未读的通知数量
    int selectNoticeUnreadCount(int userId, String topic);
}
 
message-mapper.xml中书写sql语句
 
    
        select 
        from message
        where id in (
            select max(id) from message
            where status != 2
            and from_id = 1
            and to_id = #{userId}
            and conversation_id = #{topic}
        )
    

    
        select count(id) from message
        where status != 2
        and from_id = 1
        and to_id = #{userId}
        and conversation_id = #{topic}
    

    
        select count(id) from message
        where status = 0
        and from_id = 1
        and to_id = #{userId}
        
            and conversation_id = #{topic}
        
    

业务层，新增方法调用mapper中的方法

    public Message findLatestNotice(int userId, String topic) {
        return messageMapper.selectLatestNotice(userId, topic);
    }

    public int findNoticeCount(int userId, String topic) {
        return messageMapper.selectNoticeCount(userId, topic);
    }

    public int findNoticeUnreadCount(int userId, String topic) {
        return messageMapper.selectNoticeUnreadCount(userId, topic);
    }

表现层：处理查询评论类、点赞类、关注类的请求

    @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 messageVO = new HashMap<>();
            messageVO.put("message", message);
            String content = HtmlUtils.htmlUnescape(message.getContent());
            Map 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 messageVO = new HashMap<>();
            messageVO.put("message", message);
            String content = HtmlUtils.htmlUnescape(message.getContent());
            Map 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 messageVO = new HashMap<>();
            messageVO.put("message", message);
            String content = HtmlUtils.htmlUnescape(message.getContent());
            Map 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";
    }

修改letter.html的系统通知

						

							
								系统通知27
						

使用Thymeleaf管理notice.html，并修改通知列表

				
				

					
					

						3
						


						

							

								评论
								2019-04-28 14:13:25
							
							

								
									用户
									nowcoder
									评论了你的帖子 ...
								
								

									
共 3 条会话
								
							
						
					
					
					

						3
						


						

							

								赞
								2019-04-28 14:13:25
							
							

								
									用户
									nowcoder
									点赞了你的帖子 ...
								
								

									
共 3 条会话
								
							
						
					
					
					

						3
						


						

							

								关注
								2019-04-28 14:13:25
							
							

								
									用户
									nowcoder 关注了你 ...
								
								

									
共 3 条会话
								
							
						
					
					
				

数据访问层：MessageMapper中新增查询通知列表的方法

    //查询某个主题所包含的通知列表
    List selectNotices(int userId, String topic, int offset, int limit);

message-mapper中书写sql语句

    
        select 
        from message
        where status != 2
        and from_id = 1
        and to_id = #{userId}
        and conversation_id = #{topic}
        order by create_time desc
        limit #{offset},#{limit}
    
 
业务层：新增查询通知列表的方法
 
    public List findNotices(int userId, String topic, int offset, int limit) {
        return messageMapper.selectNotices(userId, topic, offset, limit);
    }
 
MessageController中处理查询通知列表的请求
 
    @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 noticeList = messageService.findNotices(user.getId(), topic, page.getOffset(), page.getLimit());
        List> noticeVOList = new ArrayList<>();
        if (noticeList != null) {
            for (Message notice : noticeList) {
                Map map = new HashMap<>();
                //通知
                map.put("notice", notice);
                //内容
                String content = HtmlUtils.htmlUnescape(notice.getContent());
                Map 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 ids = getLetterIds(noticeList);
        if (!ids.isEmpty()) {
            messageService.readMessage(ids);
        }
        return "/site/notice-detail";
    }
 
使用Thymeleaf管理notice-detail.html，并修改通知列表
 
			
			

				

					
					

						

							落基山脉下的闲人
							2019-04-25 15:49:32
							
								×
							
						
						

								
									用户
									nowcoder
									评论了你的帖子,
									点击查看 !
								
							
									用户
									nowcoder
									点赞了你的帖子,
									点击查看 !
								
							
									用户
									nowcoder
									关注了你,
									点击查看 !
								
						
					
				
			
 
使用Interceptor显示消息的未读数量
 controller下Interceptor新增MessageInterceptor
 
@Component
public class MessageInterceptor implements HandlerInterceptor {

    @Autowired
    private HostHolder hostHolder;

    @Autowired
    private MessageService messageService;

    @Override
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
        User user = hostHolder.getUser();
        if (user != null && modelAndView != null) {
            int letterUnreadCount = messageService.findLetterUnreadCount(user.getId(), null);
            int noticeUnreadCount = messageService.findNoticeUnreadCount(user.getId(), null);
            modelAndView.addObject("allUnreadCount", letterUnreadCount + noticeUnreadCount);
        }
    }
}
 
修改WebMvcConfig
 
        registry.addInterceptor(messageInterceptor)
                .excludePathPatterns("*.css", "*.js", "*.png", "*.jpg", "*.jpeg");
 
修改index.html


转载请注明：文章转载自 www.051e.com
本文地址：http://www.051e.com/it/1047811.html


   上一篇  跨境电商wms系统功能分析
 
   下一篇  Orb
  






云平台相关栏目本月热门文章

  
1JPA+QueryDSL
  
2Cookie、Session、Token之间的区别到底在哪？
  
3springBoot使用自定义注解实现复杂校验
  
4Mybatis-plus学习笔记---乐观锁、逻辑删除、代码自动生成器等
  
5反射Field、Method、Constructor
  
6Servlet+mybatis整合Web项目
  
7SLF4j的介绍与使用+SpringBoot日志配置
  
8MybatisPlus【SpringBoot】 2 入门案例
  
9IntelliJ IDEA 2022免费激活码正规申请方法（IDEA官方激活注册码）
  
10Qt多线程编程简介
  
11商品管理模块微服务demo
  
12移动开发技术作业2
  
13SpringCloud 之OpenFeign 自定义配置和使用/自定义拦截器
  
14关于前端安全性那些事
  
15JSP超市管理系统myeclipse定制开发SQLServer数据库网页模式java编程jdbc
  
16java---数位dp---计数问题（每日一道算法2022.10.21）
  
17在javaweb项目中进行添加的时候，下拉框选中的数据怎末进行保存呢
  
18移动开发课程-作业二
  
19基本数据类型转换（java）
  
20Spring Boot工程配置多数据源，Cannot resolve configuration property dynamic







热门相关搜索



北京印刷学院排名
北方工业大学排名
北京航空航天大学排名
首都经济贸易大学排名
中国传媒大学排名
首都师范大学排名
中国地质大学(北京)排名
北京信息科技大学排名
中央民族大学排名
中国戏曲学院排名
河北政法职业学院排名
河北经贸大学排名
天津中德应用技术大学排名
天津医学高等专科学校排名
天津美术学院排名
天津音乐学院排名
天津工业大学排名
北京工业大学耿丹学院排名
北京警察学院排名
天津科技大学排名
铁岭卫生职业学院排名
沈阳北软信息职业技术学院排名
吉林艺术学院排名
吉林体育学院排名
白城师范学院排名
通化师范学院排名
吉林农业大学排名
长春电子科技学院排名
吉林工业职业技术学院排名
吉林司法警官职业学院排名
吉林警察学院排名
长春健康职业学院排名
吉林水利电力职业学院排名
大庆师范学院排名
黑龙江大学排名
哈尔滨剑桥学院排名
哈尔滨铁道职业技术学院排名
哈尔滨应用职业技术学院排名
黑龙江民族职业学院排名
七台河职业学院排名
黑龙江生态工程职业学院排名
华东师范大学排名
上海健康医学院排名
华东理工大学排名
上海交通大学排名
同济大学排名
复旦大学排名
上海杉达学院排名
上海戏剧学院排名
上海音乐学院排名







学习工具
代数计算器
三角函数
解析几何
立体几何


知识解答
教育知识
百科知识
生活知识
常识知识


写作必备
作文大全
作文素材
句子大全

实用范文


关于我们
关于我们
联系我们
网站地图




终身学习网 版权所有 ©2023-2025  051e.com     ICP备案号：京ICP备12030808号
 






关于我们
文章归档
网站地图
联系我们
版权所有 ©2023-2025  051e.com
ICP备案号：京ICP备12030808号