BBS论坛项目相关-11：Kafka异步消息队列相关

阻塞队列：BlockingQueue

解决线程通信问题
阻塞方法：put，take
生产者消费者模式：即N个线程进行生产，同时N个线程进行消费，两种角色通过内存缓冲区进行通信，
生产者：产生数据的线程
消费者：使用数据的线程
实现类：ArrayBlockingQueue，LinkedBlockingQueue，PriorityBlockingQueue等

生产者生产数据放在阻塞队列中

class Producer implements Runnable {private BlockingQueue<Integer> queue;public Producer(BlockingQueue<Integer> queue) {this.queue = queue;}@Overridepublic 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<Integer> queue;public Consumer(BlockingQueue<Integer> queue) {this.queue = queue;}@Overridepublic 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();}}
}

主线程
生产者和消费者公用一个阻塞队列

public class BlockingQueueTests {public static void main(String[] args) {BlockingQueue queue = new ArrayBlockingQueue(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();}}

消息队列

使用消息队列的好处：
解耦和扩展性，允许独立的扩展或修改，两边的处理过程，只要确保它们遵守同样的接口约束。消息队列可作为一个接口层，解耦重要的业务流程，只需要遵守约定，针对数据编程即可获取扩展能力。
健壮性：消息队列可以堆积请求，所以消费端业务即使短时间死掉，也不会影响业务的正常运行
冗余：可以采用一对多的方式，一个生产者发布消息，可以被多个订阅topic的服务消费到，供多个毫无关联的业务使用
缓冲，有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
灵活性&峰值处理能力，使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃
异步通信，消息队列提供异步处理机制，允许用户把消息放入队列，并不立即处理它，往队列里放消息，在需要的时候再去处理。
消息队列的两种模式：
1 点对点（一对一，消费者主动拉取数据，消息得到后消息清除）
2 发布/订阅模式（一对多，消费者消费数据之后不会清除消息）【分为两种：消费者主动拉取数据：生产者需要时常询问有没有消息；生产者推动数据队列：由于消费者速度不一致，可能导致资源浪费或消费者崩掉】

Kafka

一个分布式发布-订阅消息系统。分布式，可划分的，冗余备份的持久性的日志服务，主要用于处理流式数据。
应用：消息系统，日志收集，用户行为跟踪，流式处理
特点：高吞吐量，消息持久化，高可靠性，高扩展性
Kafka术语 ：Broker、Zookeeper、Topic、partition、Offeset、Leader Replica、Follow Replica

创建主题：kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --totpic test
查看主题：kafka-topics.bat --list --bootstrap-server localhost:9092
生产者发送消息：kafka-console-producer.bat --broker-list localhost:9092 --topic test
消费者命令查看命令：kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning

Spring整合Kafka

引入依赖：spring-kafka
配置kafka：配置server，consumer

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

访问kafka：
生产者：kafkaTemplate.send(topic,content);
消费者：
@KafkaListener(topics={””})
public void handleMassage(ConsumerRecord record){}
生产者发消息是主动发送，消费者接收消息是被动接收2，只要有消息就接收。

public class KafkaTests {@Autowiredprivate KafkaProducer kafkaProducer;@Testpublic void testKafka() {kafkaProducer.sendMessage("test", "你好");kafkaProducer.sendMessage("test", "在吗");try {Thread.sleep(1000 * 10);} catch (InterruptedException e) {e.printStackTrace();}}
}
@Component
class KafkaProducer {@Autowiredprivate 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());}
}

发送系统通知

对不同事件创建一个主题，系统发布消息后就不用管了，可以处理下一条消息，消费者自己拉取消息。生产者和消费者处理消息时是并发的，可以同时进行。
触发事件：评论后，发布通知；点赞后，发布通知；关注后，发布通知
处理事件：封装事件对象，开发事件的生产者，开发事件的消费者

创建一个事件类
包含topic，userId，entityId，entityType，entityUserId等属性，将整个事件的触发对象userid，发生事件的对象entityId，entityType以及该对象的作者entityUserId全都包含在里面，便于之后通知调用，并用一个map来存放可能出现的额外数据，便于扩展。

public class Event {private String topic;private int userId;private int entityType;private int entityId;private int entityUserId;private Map<String, Object> data = new HashMap<>();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<String, Object> getData() {return data;}public Event setData(String key, Object value) {this.data.put(key, value);return this;}
}

可通过set设置topic和userId等，用一个map<String,Object>类来存放数据，返回this，可实现链式调用。
在处理系统消息时，有了事件类，就可以创建一个事件生产类直接将这个事件类对象用json字符串发送到队列，消费者自行拉取后就能从中获取所需要的数据，并返回给前端对应的消息。无论是评论点赞还是关注等，触发事件发生后只需要将所需要发送的数据用事件类对象封装传入事件生产类的发送事件方法，即可发送对应消息。
Produce

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

Consumer
消费者监听各个主题，对每个主题发过来的消息进行处理。
对主题可以在常量接口中设置常量，方便管理。

public interface CommunityConstant {/*** 激活成功*/int ACTIVATION_SUCCESS = 0;/*** 重复激活*/int ACTIVATION_REPEAT = 1;/*** 激活失败*/int ACTIVATION_FAILURE = 2;/*** 默认状态的登录凭证的超时时间*/int DEFAULT_EXPIRED_SECONDS = 3600 * 12;/*** 记住状态的登录凭证超时时间*/int REMEMBER_EXPIRED_SECONDS = 3600 * 24 * 100;/*** 实体类型: 帖子*/int ENTITY_TYPE_POST = 1;/*** 实体类型: 评论*/int ENTITY_TYPE_COMMENT = 2;/*** 实体类型: 用户*/int ENTITY_TYPE_USER = 3;/*** 主题: 评论*/String TOPIC_COMMENT = "comment";/*** 主题: 点赞*/String TOPIC_LIKE = "like";/*** 主题: 关注*/String TOPIC_FOLLOW = "follow";/*** 主题: 发帖*/String TOPIC_PUBLISH = "publish";/*** 主题: 删帖*/String TOPIC_DELETE = "delete";/*** 主题: 分享*/String TOPIC_SHARE = "share";}

系统通知也属于私信功能，只是from_id为固定1系统id，发送站内消息，由数据库中1系统用户发送，此时conversationId就不需要再拼接from_id和to_id，可以用来存储topic主题字符串，content用于存储需要在前端页面显示的内容的json字符串，传到前端进行拼接。私信功能模块实现

public class Message {private int id;private int fromId;private int toId;private String conversationId;private String content;private int status;private Date createTime;
}

消费者监听所有主题，对ConsumerRecord接收的消息进行处理，发送过来的是一个json字符串，先转换为even事件对象。
构建消息对象，from_id为系统，to_id为发送过来的事件中的entityUserid，conversationId为topic，创建时间为当前时间。需要拼接的是消息内容
消息内容可用一个map来接收用于拼接的数据，用户id，entityType，entityId等。
event中可能会有其他数据，遍历event的map中的数据，存在消息内容的map中。存完后就可以将转为json字符串存入message的content中。
message消息全部内容已填充完毕，调用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<String, Object> 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<String, Object> entry : event.getData().entrySet()) {content.put(entry.getKey(), entry.getValue());}}message.setContent(JSONObject.toJSONString(content));messageService.addMessage(message);}

消费者事件不需要我们主动调用，它是被动触发的，我们只需要在触发事件的时候调用生产者方法生产消息，消费者就会发送站内消息。
在评论时触发事件
创建event对象，设置主题comment，设置触发事件的用户，即当前持有用户发出评论，设置评论的entityId和entityType，在系统发送的消息中可以点击查看具体内容，需要链接到该评论的帖子，所以还需要把帖子id传入。
对于entityUserid需要进行判断，因为可能是对帖子评论，也可能是对帖子的评论进行回复，所以需要对评论的entityType进行判断，如果是对帖子评论，则通过帖子表查询帖子的userid，如果是对评论的回复，则通过评论表进行查询评论的userid。
event对象补充完毕之后就将时间发送给kafka消息生产者，消息发送就不用管了，消费者会自动触发。

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

点赞消息系统通知
点赞也是同样的逻辑，在用户点赞后将事件发送给kafka消息生产者。对于点赞的系统消息中查看详情需要到达帖子的详情页面，所以也要将postId传入。

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<String, Object> 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);}

关注事件系统通知
点击详情只需要点进关注者的个人主页，不需要传入帖子id

@RequestMapping(path = "/follow", method = RequestMethod.POST)@ResponseBodypublic 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, "已关注!");}

注意：
前面对service所有方法用AOP做了一个统一日志记录，里面通过request来获得用户ip，实际上添加了kafka生产者与消费者功能后，消费者也调用了service方法，但此时没有请求，request为null，会空指针异常，所以需要进行判断处理。

@Before("pointcut()")public void before(JoinPoint joinPoint) {// 用户[1.2.3.4],在[xxx],访问了[com.nowcoder.community.service.xxx()].ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();if (attributes == null) {return;}HttpServletRequest request = attributes.getRequest();String ip = request.getRemoteHost();String now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());String target = joinPoint.getSignature().getDeclaringTypeName() + "." + joinPoint.getSignature().getName();logger.info(String.format("用户[%s],在[%s],访问了[%s].", ip, now, target));}

显示系统通知

通知存在message中
查询某个主题下的最新通知

<select id="selectLatestNotice" resultType="Message">select <include refid="selectFields"></include>from messagewhere id in (select max(id) from messagewhere status != 2and from_id = 1and to_id = #{userId}and conversation_id = #{topic})</select>

查询某个主题下包含的通知数

 <select id="selectNoticeCount" resultType="int">select count(id) from messagewhere status != 2and from_id = 1and to_id = #{userId}and conversation_id = #{topic}</select>

查询未读通知的数量

<select id="selectNoticeUnreadCount" resultType="int">select count(id) from messagewhere status = 0and from_id = 1and to_id = #{userId}<if test="topic!=null">and conversation_id = #{topic}</if></select>

controller层
获得通知的message后，需要对message进行处理，因为传过来的message的内容是一个json字符串，里面存储的是消息通知内容，如XX给你的帖子XX点赞消息中存储了XX的用户id，被点赞对象的entityType，entityId，以及entityUserid，如果需要跳转到帖子，还会有postId。所以需要将json字符串还原，需要先将字符串中的转义字符进行反转义，然后将字符串还原为对象，消费者对其进行封装时用的是map，所以又将字符串还原为map，然后取出map中的内容补充放入一个新的map中（如根据用户id查询用户，将用户对象放入新map），和message，以及查询的相关消息通知数量，未读消息数量一起返回给页面。

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

前端同样使用遍历map获取。

通知详情页

查看通知详情页时，需要通过url将topic传入，SQL查找关于这个topic的通知消息，按照创建时间倒序排列。

<select id="selectNotices" resultType="Message">select <include refid="selectFields"></include>from messagewhere status != 2and from_id = 1and to_id = #{userId}and conversation_id = #{topic}order by create_time desclimit #{offset}, #{limit}</select>

同样需要先获得当前持有用户，根据用户id为to_id，系统1为from_id，再加上topic为conversationid，查询得到一个List，同样需要用一个List<Map<String, Object>>结构去存储每条消息的具体细节。
遍历list，每次循环创建一个map，将通知对象放入map中，并将通知的内容还原为map对象，取出里面的相关信息，获取用户id（实际上是触发这个事件的用户，即对你点赞评论的用户），entityType，entityId（对方对你进行操作的实体，帖子或评论），postId也一并取出，为null时，说明不需要postid，对前端显示也不影响。
同时当查询通知详情页需要将未读通知全设置为已读。可以直接用私信那块写的已读逻辑，先获得所有未读通知的list，传入list，<foreach>标签设为已读。BBS项目相关-私信模块

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

总消息

对于最上方一栏总的未读消息处理，因为需要随时显示，所以用拦截器处理。对每个请求查询未读私信数量和未读通知数量。

@Component
public class MessageInterceptor implements HandlerInterceptor {@Autowiredprivate HostHolder hostHolder;@Autowiredprivate 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);}}
}

同样需要把拦截器加入配置，对所有动态请求拦截

@Configuration
public class WebMvcConfig implements WebMvcConfigurer {@Autowiredprivate AlphaInterceptor alphaInterceptor;@Autowiredprivate LoginTicketInterceptor loginTicketInterceptor;//    @Autowired
//    private LoginRequiredInterceptor loginRequiredInterceptor;@Autowiredprivate MessageInterceptor messageInterceptor;@Autowiredprivate DataInterceptor dataInterceptor;@Overridepublic void addInterceptors(InterceptorRegistry registry) {registry.addInterceptor(loginTicketInterceptor).excludePathPatterns("/**/*.css", "/**/*.js", "/**/*.png", "/**/*.jpg", "/**/*.jpeg");registry.addInterceptor(messageInterceptor).excludePathPatterns("/**/*.css", "/**/*.js", "/**/*.png", "/**/*.jpg", "/**/*.jpeg");registry.addInterceptor(dataInterceptor).excludePathPatterns("/**/*.css", "/**/*.js", "/**/*.png", "/**/*.jpg", "/**/*.jpeg");}}