仿牛客网项目---消息队列的实现

本篇文章讲一讲我们的项目中用到的消息队列。

 

1.阻塞队列

2.kafka

我的项目为什么要用消息队列？

如果采用消息队列，那么评论、点赞、关注三类不同的事，可以定义三类不同的主题（评论、点赞、关注），发生相应的事件可以将其包装成一条消息放入对应的队列里。那么当前的线程可以继续处理下一条消息，不用处理后续的业务（后续由消费者线程处理）。面向事件驱动编程。

3.发布系统通知

下面讲讲怎么发布系统通知。

在entity包下创建event类：

public class Event {
 
    private String topic;  //表示事件的主题的字符串
    private int userId;     //表示与事件相关的用户的ID
    private int entityType;   //表示与事件相关的实体类型
    private int entityId;      //表示与事件相关的实体的ID
    private int entityUserId;    //表示与实体相关的用户的ID
    private Map data = new HashMap<>();   //一个Map对象，用于存储与事件相关的附加数据
 
    //省略了很多get()方法和set()方法
}

通过使用这个Event类，我就可以创建一个事件对象，并设置事件的主题、用户ID、实体类型、实体ID、实体用户ID以及其他相关的附加数据。

接着在在community的event目录下创建EventProducer、EventCnsumer类。

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

通过调用fireEvent方法，我们就可以将一个事件对象发送到Kafka消息队列中的指定主题。这样，消费者可以从该主题订阅事件，并对事件进行相应的处理逻辑。

@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);
    }
}

在handleCommentMessage方法中，首先检查接收到的消息是否为空，然后将消息内容解析为Event对象。如果解析失败，会记录错误并返回。

接下来，根据接收到的事件信息，构建一个站内通知的Message对象。设置发送者ID为系统用户ID，接收者ID为事件关联的实体用户ID，会话ID为事件主题，创建时间为当前时间。

然后，构建一个存储事件相关信息的content映射。

最后，将content转换为JSON字符串，并将其设置为站内通知的内容。

说了一堆乱七八糟的，反正就是这段代码提供了一个事件消费者类，用于订阅和处理从Kafka消息队列中接收到的指定事件主题的消息。它可以根据事件的类型执行相应的处理操作，例如发送站内通知等。

在评论、点赞、关注视图层增加方法。

CommentController。LikeController。Followcontroller。

4.显示系统通知

消息未读使用拦截器实现，因为每个请求都需要记录次数。

@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);
        }
    }
}

上面的这段代码提供了一个消息拦截器类，用于在请求处理完成后，对返回的ModelAndView对象进行拦截和处理，以实现消息的统计和展示功能。