1、阻塞队列image.png

测试代码：

  1. public class BlockingQueueTests {
  3.     public static void main(String[] args) {
  4.         BlockingQueue queue = new ArrayBlockingQueue(10);
  5.         new Thread(new Producer(queue)).start();
  6.         new Thread(new Consumer(queue)).start();
  7.         new Thread(new Consumer(queue)).start();
  8.         new Thread(new Consumer(queue)).start();
  9.     }
  11. }
  13. class Producer implements Runnable {
  15.     private BlockingQueue<Integer> queue;
  17.     public Producer(BlockingQueue<Integer> queue) {
  18.         this.queue = queue;
  19.     }
  21.     @Override
  22.     public void run() {
  23.         try {
  24.             for (int i = 0; i < 100; i++) {
  25.                 Thread.sleep(20);
  26.                 queue.put(i);
  27.                 System.out.println(Thread.currentThread().getName() + "生产:" + queue.size());
  28.             }
  29.         } catch (Exception e) {
  30.             e.printStackTrace();
  31.         }
  32.     }
  34. }
  36. class Consumer implements Runnable {
  38.     private BlockingQueue<Integer> queue;
  40.     public Consumer(BlockingQueue<Integer> queue) {
  41.         this.queue = queue;
  42.     }
  44.     @Override
  45.     public void run() {
  46.         try {
  47.             while (true) {
  48.                 Thread.sleep(new Random().nextInt(1000));
  49.                 queue.take();
  50.                 System.out.println(Thread.currentThread().getName() + "消费:" + queue.size());
  51.             }
  52.         } catch (Exception e) {
  53.             e.printStackTrace();
  54.         }
  55.     }
  56. }

2、Kafka

1、下载官网：kafka.apache.org
2、解压缩后到指定目录下d:/AAcommunity/kafka
修改配置config中zookeeper.properties的dataDir改变存放位置d:/AAjava/work/data/zookeeper
修改配置config中server.properties的log.dirs改变存放位置d:/AAjava/work/data/kafka-logs
image.png
测试：
1、新窗口启动启动zookeeper
进入到kafka安装目录下，执行bin\windows\zookeeper-server-start.bat config\zookeeper.properties指定配置文件
2、新窗口启动启动kafka
进入到kafka安装目录下，执行bin\windows\kafka-server-start.bat config\server.properties指定配置文件
3、新窗口启动创建主题test
进入到windows目录下 ==>kafka-topics.bat —create —bootstrap-server localhost:9092 —replication-factor 1 —partitions 1 —topic test 没有提示即创建成功
4、查看是否创建成功
kafka-topics.bat —list —bootstrap-server localhost:9092
5、生产者发送消息
kafka-console-producer.bat —broker-list localhost:9092 —topic test
回车发送消息内容
hello
world
6、新窗口启动消费者读取消息
进入到windows目录下
kafka-console-consumer.bat —bootstrap-server localhost:9092 —topic test —from-beginning

3、spring整合kafka

1、导入依赖

  1. <dependency>
  2.     <groupId>org.springframework.kafka</groupId>
  3.     <artifactId>spring-kafka</artifactId>
  4. </dependency>

2、配置kafka

  1. #配置KafkaProperties
  2. spring.kafka.bootstrap-servers=localhost:9092
  3. spring.kafka.consumer.group-id=test-consumer-group
  4. spring.kafka.consumer.enable-auto-commit=true
  5. spring.kafka.consumer.auto-commit-interval=3000

3、测试代码
Producer

  1. @Component
  2. public class Producer{
  4.     @Autowired
  5.     private KafkaTemplate kafkaTemplate;
  7.     public void sendMessage(String title,String content){
  8.         kafkaTemplate.send(title,content);
  9.     }
  10. }

Consumer

  1. @Component
  2. public class Consumer{
  4.     @KafkaListener(topics = {"test"})
  5.     public void readMessage(ConsumerRecord record){
  6.         System.out.println(record.value());
  7.     }
  8. }

KafkaTset

  1. public class KafkaTest {
  3.     @Resource
  4.     private Producer producer;
  6.     @Test
  7.     public void testKafka(){
  8.         producer.sendMessage("test","kafka");
  9.         producer.sendMessage("test","test");
  11.         try {
  12.             Thread.sleep(10*900);
  13.         } catch (InterruptedException e) {
  14.             e.printStackTrace();
  15.         }
  16.     }
  17. }

4、发送系统通知

1、Event

  1. public class Event {
  3.     private String topic;//主题
  4.     private int userId;//当前用户id
  5.     private int entityType;//实体类型
  6.     private int entityId;//实体id
  7.     private int entityUserId;//实体用户id
  8.     private Map<String, Object> data = new HashMap<>();
  10.     public String getTopic() {
  11.         return topic;
  12.     }
  14.     public Event setTopic(String topic) {
  15.         this.topic = topic;
  16.         return this;
  17.     }
  19.     public int getUserId() {
  20.         return userId;
  21.     }
  23.     public Event setUserId(int userId) {
  24.         this.userId = userId;
  25.         return this;
  26.     }
  28.     public int getEntityType() {
  29.         return entityType;
  30.     }
  32.     public Event setEntityType(int entityType) {
  33.         this.entityType = entityType;
  34.         return this;
  35.     }
  37.     public int getEntityId() {
  38.         return entityId;
  39.     }
  41.     public Event setEntityId(int entityId) {
  42.         this.entityId = entityId;
  43.         return this;
  44.     }
  46.     public int getEntityUserId() {
  47.         return entityUserId;
  48.     }
  50.     public Event setEntityUserId(int entityUserId) {
  51.         this.entityUserId = entityUserId;
  52.         return this;
  53.     }
  55.     public Map<String, Object> getData() {
  56.         return data;
  57.     }
  59.     public Event setData(String key, Object value) {
  60.         this.data.put(key, value);
  61.         return this;
  62.     }
  63. }

2、CommunityConstant添加四个常量

  1.  /**
  2.      * 主题:评论
  3.      */
  4.     String TOPIC_COMMENT = "comment";
  6.     /**
  7.      * 主题:点赞
  8.      */
  9.     String TOPIC_LIKE = "like";
  11.     /**
  12.      * 主题:关注
  13.      */
  14.     String TOPIC_FOLLOW = "follow";
  16.     /**
  17.      * 系统用户id
  18.      */
  19.     int SYSTEM_USER_ID = 1;

3、EventProducer

  1. @Component
  2. public class EventProducer {
  4.     @Autowired
  5.     private KafkaTemplate template;
  6.     //将整个message传入，后面页面再按需选择需要的信息
  7.     public void fireEvent(Event event){
  8.         template.send(event.getTopic(), JSONObject.toJSONString(event));
  9.     }
  10. }

4、EventConsumer

  1. @Component
  2. public class EventConsumer implements CommunityConstant {
  4.     private static final Logger logger = LoggerFactory.getLogger(EventConsumer.class);
  6.     @Autowired
  7.     private MessageService messageService;
  8.     @Autowired
  9.     private DiscussPostService discussPostService;
  10.     @Autowired
  11.     private ElasticsearchService elasticsearchService;
  13.     @KafkaListener(topics = {TOPIC_COMMENT, TOPIC_LIKE, TOPIC_FOLLOW})
  14.     public void handleCommentMessagee(ConsumerRecord record) {
  15.         if (record == null || record.value() == null) {
  16.             logger.error("消息的内容为空！");
  17.             return;
  18.         }
  20.         Event event = JSONObject.parseObject(record.value().toString(), Event.class);
  21.         if (event == null) {
  22.             logger.error("消息格式错误");
  23.             return;
  24.         }
  26.         //发送站内消息
  27.         Message message = new Message();
  28.         message.setFromId(SYSTEM_USER_ID);
  29.         message.setToId(event.getEntityUserId());
  30.         message.setConversationId(event.getTopic());//存储主题
  31.         message.setCreateTime(new Date());
  33.         Map<String, Object> content = new HashMap<>();
  34.         content.put("userId", event.getUserId());
  35.         content.put("entityType", event.getEntityType());
  36.         content.put("entityId", event.getEntityId());
  38.         //将map的数据存进content
  39.         if (!event.getData().isEmpty()) {
  40.             for (Map.Entry<String, Object> entry : event.getData().entrySet()) {
  41.                 content.put(entry.getKey(), entry.getValue());
  42.             }
  43.         }
  45.         //最后将content存进message的Content
  46.         message.setContent(JSONObject.toJSONString(content));
  47.         messageService.addMessage(message);
  48.     }
  49. }

5、CommentMapper添加根据id查询帖子

  1. Comment findCommentById(int id);

6、comment-mapper.xml

  1.  <select id="findCommentById" resultType="Comment">
  2.         select <include refid="selectFields"></include>
  3.         from comment
  4.         where id=#{id}
  5.     </select>

7、CommentService

  1.  public Comment findCommentById(int entityId) {
  2.         return commentMapper.findCommentById(entityId);
  3.     }

8、CommentController

  1.          // 触发评论事件
  2.         Event event = new Event()
  3.                 .setTopic(TOPIC_COMMENT)
  4.                 .setUserId(hostHolder.getUser().getId())
  5.                 .setEntityType(comment.getEntityType())
  6.                 .setEntityId(comment.getEntityId())
  7.                 .setData("postId", discussPostId);
  9.         if (comment.getEntityType() == ENTITY_TYPE_POST) {
  10.             DiscussPost target = discussPostService.selectDiscussPostById(comment.getEntityId());
  11.             event.setEntityUserId(target.getUserId());
  12.         } else if (comment.getEntityType() == ENTITY_TYPE_COMMENT) {
  13.             Comment target = commentService.findCommentById(comment.getEntityId());
  14.             event.setEntityUserId(target.getUserId());
  15.         }
  16.         eventProducer.fireEvent(event);

9、重构LikeController：添加参数postId

  1.         // 触发点赞事件
  2.         if (likeStatus == 1) {
  3.             Event event = new Event()
  4.                     .setTopic(TOPIC_LIKE)
  5.                     .setUserId(hostHolder.getUser().getId())
  6.                     .setEntityType(entityType)
  7.                     .setEntityId(entityId)
  8.                     .setEntityUserId(entityUserId)
  9.                     .setData("postId", postId);
  10.             eventProducer.fireEvent(event);
  11.         }

10、FollowController

  1.          // 触发关注事件
  2.         Event event = new Event()
  3.                 .setTopic(TOPIC_FOLLOW)
  4.                 .setUserId(hostHolder.getUser().getId())
  5.                 .setEntityType(entityType)
  6.                 .setEntityId(entityId)
  7.                 .setEntityUserId(entityId);
  8.         eventProducer.fireEvent(event);

11、页面（略）
discuss-detail.html
discuss.js

12、空指针异常
ServiceLogAspect.java

  1.   if(attributes==null){
  2.         return;
  3.      }

5、显示系统通知

• 通知列表

  • 显示评论、点赞、关注三种类型的通知
    • 通知详情
  • 分页显示某一类主题所包含的通知
    • 未读消息
  • 在页面头部显示所有的未读消息数量

1、MessageMapper

  1.  /**
  2.      * 查询某个主题下最新的通知
  3.      * @param userId
  4.      * @param topic
  5.      * @return
  6.      */
  7.     Message selectLatestNotice(int userId, String topic);
  9.     /**
  10.      * 查询某个主题所包含的通知数量
  11.      * @param userId
  12.      * @param topic
  13.      * @return
  14.      */
  15.     int selectNoticeCount(int userId, String topic);
  17.     /**
  18.      * 查询未读的通知的数量
  19.      * @param userId
  20.      * @param topic
  21.      * @return
  22.      */
  23.     int selectNoticeUnreadCount(int userId, String topic);
  25.     /**
  26.      * 查询某个主题所包含的通知列表
  27.      * @param userId
  28.      * @param topic
  29.      * @param offset
  30.      * @param limit
  31.      * @return
  32.      */
  33.     List<Message> selectNotices(int userId, String topic, int offset, int limit);

2、message-mapper.xml

  1. <select id="selectLatestNotice" resultType="Message">
  2.         select <include refid="selectFields"></include>
  3.         from message
  4.         where id in (
  5.         select max(id) from message
  6.         where status != 2
  7.         and from_id = 1
  8.         and to_id = #{userId}
  9.         and conversation_id = #{topic}
  10.         )
  11.     </select>
  13.     <select id="selectNoticeCount" resultType="int">
  14.         select count(id) from message
  15.         where status != 2
  16.         and from_id = 1
  17.         and to_id = #{userId}
  18.         and conversation_id = #{topic}
  19.     </select>
  21.     <select id="selectNoticeUnreadCount" resultType="int">
  22.         select count(id) from message
  23.         where status = 0
  24.         and from_id = 1
  25.         and to_id = #{userId}
  26.         <if test="topic!=null">
  27.             and conversation_id = #{topic}
  28.         </if>
  29.     </select>
  31.     <select id="selectNotices" resultType="Message">
  32.         select <include refid="selectFields"></include>
  33.         from message
  34.         where status != 2
  35.         and from_id = 1
  36.         and to_id = #{userId}
  37.         and conversation_id = #{topic}
  38.         order by create_time desc
  39.         limit #{offset}, #{limit}
  40.     </select>

3、MessageService

  1.  public Message findLatestNotice(int userId, String topic) {
  2.         return messageMapper.selectLatestNotice(userId, topic);
  3.     }
  5.     public int findNoticeCount(int userId, String topic) {
  6.         return messageMapper.selectNoticeCount(userId, topic);
  7.     }
  9.     public int findNoticeUnreadCount(int userId, String topic) {
  10.         return messageMapper.selectNoticeUnreadCount(userId, topic);
  11.     }
  13.     public List<Message> findNotices(int userId, String topic, int offset, int limit) {
  14.         return messageMapper.selectNotices(userId, topic, offset, limit);
  15.     }

4、MessageController
注意再私信接口中也要添加上查询未读通知数量：
int noticeUnreadCount = messageService.findNoticeUnreadCount(user.getId(), null);
model.addAttribute(“noticeUnreadCount”, noticeUnreadCount);

  1. /**
  2.      * 通知接口
  3.      * @param model
  4.      * @return
  5.      */
  6.     @GetMapping("/notice/list")
  7.     public String getNoticeList(Model model) {
  8.         User user = hostHolder.getUser();
  10.         // 查询评论类通知
  11.         Message message = messageService.findLatestNotice(user.getId(), TOPIC_COMMENT);
  12.         Map<String, Object> messageVO = new HashMap<>();
  13.         if (message != null) {
  14.             messageVO.put("message", message);
  16.             String content = HtmlUtils.htmlUnescape(message.getContent());
  17.             Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);
  19.             messageVO.put("user", userService.selectUserById((Integer) data.get("userId")));
  20.             messageVO.put("entityType", data.get("entityType"));
  21.             messageVO.put("entityId", data.get("entityId"));
  22.             messageVO.put("postId", data.get("postId"));
  24.             int count = messageService.findNoticeCount(user.getId(), TOPIC_COMMENT);
  25.             messageVO.put("count", count);
  27.             int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_COMMENT);
  28.             messageVO.put("unread", unread);
  29.         }
  30.         model.addAttribute("commentNotice", messageVO);
  32.         // 查询点赞类通知
  33.         message = messageService.findLatestNotice(user.getId(), TOPIC_LIKE);
  34.         messageVO = new HashMap<>();
  35.         if (message != null) {
  36.             messageVO.put("message", message);
  38.             String content = HtmlUtils.htmlUnescape(message.getContent());
  39.             Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);
  41.             messageVO.put("user", userService.selectUserById((Integer) data.get("userId")));
  42.             messageVO.put("entityType", data.get("entityType"));
  43.             messageVO.put("entityId", data.get("entityId"));
  44.             messageVO.put("postId", data.get("postId"));
  46.             int count = messageService.findNoticeCount(user.getId(), TOPIC_LIKE);
  47.             messageVO.put("count", count);
  49.             int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_LIKE);
  50.             messageVO.put("unread", unread);
  51.         }
  52.         model.addAttribute("likeNotice", messageVO);
  54.         // 查询关注类通知
  55.         message = messageService.findLatestNotice(user.getId(), TOPIC_FOLLOW);
  56.         messageVO = new HashMap<>();
  57.         if (message != null) {
  58.             messageVO.put("message", message);
  60.             String content = HtmlUtils.htmlUnescape(message.getContent());
  61.             Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);
  63.             messageVO.put("user", userService.selectUserById((Integer) data.get("userId")));
  64.             messageVO.put("entityType", data.get("entityType"));
  65.             messageVO.put("entityId", data.get("entityId"));
  67.             int count = messageService.findNoticeCount(user.getId(), TOPIC_FOLLOW);
  68.             messageVO.put("count", count);
  70.             int unread = messageService.findNoticeUnreadCount(user.getId(), TOPIC_FOLLOW);
  71.             messageVO.put("unread", unread);
  72.         }
  73.         model.addAttribute("followNotice", messageVO);
  75.         // 查询未读消息数量
  76.         int letterUnreadCount = messageService.findLetterUnreadCount(user.getId(), null);
  77.         model.addAttribute("letterUnreadCount", letterUnreadCount);
  78.         int noticeUnreadCount = messageService.findNoticeUnreadCount(user.getId(), null);
  79.         model.addAttribute("noticeUnreadCount", noticeUnreadCount);
  81.         return "/site/notice";
  82.     }
  84.     /**
  85.      * 通知详情接口
  86.      * @param topic
  87.      * @param page
  88.      * @param model
  89.      * @return
  90.      */
  91.     @GetMapping("/notice/detail/{topic}")
  92.     public String getNoticeDetail(@PathVariable("topic") String topic, Page page, Model model) {
  93.         User user = hostHolder.getUser();
  95.         page.setLimit(5);
  96.         page.setPath("/notice/detail/" + topic);
  97.         page.setRows(messageService.findNoticeCount(user.getId(), topic));
  99.         List<Message> noticeList = messageService.findNotices(user.getId(), topic, page.getOffset(), page.getLimit());
  100.         List<Map<String, Object>> noticeVoList = new ArrayList<>();
  101.         if (noticeList != null) {
  102.             for (Message notice : noticeList) {
  103.                 Map<String, Object> map = new HashMap<>();
  104.                 // 通知
  105.                 map.put("notice", notice);
  106.                 // 内容
  107.                 String content = HtmlUtils.htmlUnescape(notice.getContent());
  108.                 Map<String, Object> data = JSONObject.parseObject(content, HashMap.class);
  109.                 map.put("user", userService.selectUserById((Integer) data.get("userId")));
  110.                 map.put("entityType", data.get("entityType"));
  111.                 map.put("entityId", data.get("entityId"));
  112.                 map.put("postId", data.get("postId"));
  113.                 // 通知作者
  114.                 map.put("fromUser", userService.selectUserById(notice.getFromId()));
  116.                 noticeVoList.add(map);
  117.             }
  118.         }
  119.         model.addAttribute("notices", noticeVoList);
  121.         // 设置已读
  122.         List<Integer> ids = getLetterIds(noticeList);
  123.         if (!ids.isEmpty()) {
  124.             messageService.readMessage(ids);
  125.         }
  127.         return "/site/notice-detail";
  128.     }

5、页面(略）
letter.html 修改通知的路径
notice.html
notice-detail.html
index.html 修改数量

6、添加拦截器
MessageInterceptor

  1. @Component
  2. public class MessageInterceptor implements HandlerInterceptor {
  4.     @Autowired
  5.     private HostHolder hostHolder;
  7.     @Autowired
  8.     private MessageService messageService;
  10.     @Override
  11.     public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
  12.         User user = hostHolder.getUser();
  13.         if (user != null && modelAndView != null) {
  14.             int letterUnreadCount = messageService.findLetterUnreadCount(user.getId(), null);
  15.             int noticeUnreadCount = messageService.findNoticeUnreadCount(user.getId(), null);
  16.             modelAndView.addObject("allUnreadCount", letterUnreadCount + noticeUnreadCount);
  17.         }
  18.     }
  19. }

WebMvcConfig

  1. @Configuration
  2. public class WebMvcConfig implements WebMvcConfigurer {
  4.     @Autowired
  5.     private LoginTicketInterceptor loginTicketInterceptor;
  6.     @Autowired
  7.     private LoginRequiredInterceptor loginRequiredInterceptor;
  8.     @Autowired
  9.     private MessageInterceptor messageInterceptor;
  11.     @Override
  12.     public void addInterceptors(InterceptorRegistry registry) {
  13.         registry.addInterceptor(loginTicketInterceptor)
  14.                 .excludePathPatterns("/**/*.css","/**/*.js","/**/*.png","/**/*.jpg","/**/*.jpeg");
  15.         registry.addInterceptor(loginRequiredInterceptor)
  16.                 .excludePathPatterns("/**/*.css","/**/*.js","/**/*.png","/**/*.jpg","/**/*.jpeg");
  17.         registry.addInterceptor(messageInterceptor)
  18.                 .excludePathPatterns("/**/*.css","/**/*.js","/**/*.png","/**/*.jpg","/**/*.jpeg");
  19.     }
  20. }