❤ 作者主页：欢迎来到我的技术博客😎
❀ 个人介绍：大家好，本人热衷于 Java后端开发，欢迎来交流学习哦！(￣▽￣)~*
🍊 如果文章对您有帮助，记得关注、点赞、收藏、评论⭐️⭐️⭐️
📣 您的支持将是我创作的动力，让我们一起加油进步吧！！！🎉🎉

第三章：Kafka，构建TB级异步消息系统

一、阻塞队列

BlockingQueue
- 解决线程通信的问题。
- 阻塞方法：put、take。
生产者消费者模式
- 生产者：产生数据的线程。
- 消费者：使用数据的线程。
实现类
- ArrayBlockingQueue
- LinkedBlockingQueue
- PriorityBlockingQueue、SynchronousQueue、DelayQueue等。

1. 阻塞队列测试方法

在 test 中添加 BlockingQueueTests类，来表示阻塞队列的测试方法，代码如下：

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

class Producer implements Runnable{
    private BlockingQueue<Integer> queue;

    public Producer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }
    @Override
    public void run(){
        try{
            for(int i = 0; i < 20; 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;
    }

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

2. 测试结果

Thread-0生产：1
Thread-0生产：2
Thread-0生产：3
Thread-0生产：4
Thread-0生产：5
Thread-0生产：6
Thread-0生产：7
Thread-0生产：8
Thread-0生产：9
Thread-0生产：10
Thread-3消费：9
Thread-0生产：10
Thread-1消费：9
Thread-0生产：10
Thread-2消费：9
Thread-0生产：10
Thread-3消费：9
Thread-0生产：10
Thread-1消费：9
Thread-0生产：10
Thread-2消费：9
Thread-0生产：10
Thread-1消费：9
Thread-0生产：10
Thread-1消费：9
Thread-0生产：10
Thread-3消费：9
Thread-0生产：10
Thread-1消费：9
Thread-0生产：10
Thread-1消费：9
Thread-1消费：8
Thread-2消费：7
Thread-1消费：6
Thread-2消费：5
Thread-1消费：4
Thread-3消费：3
Thread-3消费：2
Thread-2消费：1
Thread-3消费：0

二、Kafka入门

Kafka简介
- Kafka是一个分布式的流媒体平台。
- 应用：消息系统、日志收集、用户行为追踪、流式处理。
Kafka特点
- 高吞吐量、消息持久化、高可靠性、高扩展性。
Kafka术语
- Broker、Zookeeper
- Topic、Partition、Offset
- Leader Replica 、Follower Replica

1. Kafka下载

Kafka官网: https://kafka.apache.org/
在这里插入图片描述

2. Kafka安装与配置

下载Kafka的安装包后进行解压，就相当于安装成功了。
需要进行以下配置：
修改 config包下的 zookeeper.properties:
在这里插入图片描述
修改 config包下的 server.properties:

3. Kafka的启动

首先在命令行中启动 Zookeeper:

C:\Users\Andrew> d:
D:\> cd d:\kafka_2.13-3.2.3
d:\kafka_2.13-3.2.3>bin\windows\zookeeper-server-start.bat config\zookeeper.properties

启动成功后不关闭此窗口，重新打开一个新的命令窗口，用于启动 kafka：

C:\Users\Andrew> d:
D:\> cd d:\kafka_2.13-3.2.3
d:\kafka_2.13-3.2.3>bin\windows\kafka-server-start.bat config\server.properties

注意： 当遇到“‘wmic’不是内部或外部命令，也不是可运行程序”。
在C盘下找到wbem文件夹，且里面包含WMIC.exe，将其添加到系统变量path中去。
比如我的路径是：C:\Windows\System32\wbem，在系统变量path中新建该路径。就可以正常启动Kafka了。

4. Kafka使用

创建主题

cd到 …\kafka_2.13-2.8.0\bin\windows这里，然后输入 kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
创建服务器端口号为9092（Kafka默认端口号）的topic，指生产者发布消息存储的位置在该服务器上localhost:9092。--replication-factor 1 指1个副本。--partitions 1 指1个分区。--topic test 指该主题名为 test。
在这里插入图片描述

以生产者身份发送消息

输入：kafka-console-producer.bat --broker-list localhost:9092 --topic test
生产者身份打开服务器列表中为localhost:9092的服务器上的test主题。--broker-list 指服务器列表。
并且输入要发送的消息：
在这里插入图片描述

以消费者身份读取消息

新打开一个命令行窗口，且cd到…\kafka_2.13-2.8.0\bin\windows，并输入：kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning
可以看到生产者发送的消息。并且这个消息队列中可以实时传送消息。
比如在生产者的命令行中继续输入信息，很快在消费者这边也能得到消息。
在这里插入图片描述

三、Spring整合Kafka

引入依赖
```
- `spring-kafka`
```
- 配置Kafka
  - 配置 server、consumer
- 访问Kafka
  - 生产者:
  kafkaTemplate.send(topic, data);
  - 消费者:
  @KafkaListener(topics = {"test"}
  public void handleMessage(ConsumerRecord record) {}

1. 引入依赖

在 pom.xml 添加相关的依赖：

        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.8.6</version>
        </dependency>

2. 配置Kafka

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

3. 测试

在 test 包下添加 KafkaTests 类，代码如下：

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

4. 测试结果

在这里插入图片描述

四、发送系统消息

触发事件
- 评论后，发布通知
- 点赞后，发布通知
- 关注后，发布通知
处理事件
- 封装事件对象
- 开发事件的生产者
- 开发事件的消费者

在这里插入图片描述

1. 实体类

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

创建 event 包，添加 EventProducer 类和 EventConsumer 类，代码如下：

@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<String, Object> content = new HashMap<>();
        content.put("userId", event.getUserId());
        content.put("entityType", event.getEntityType());
        content.put("entityId", event.getEntityId());

        //这里代码要再盘一下 额外数据，比如postId
        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);
    }
}

2. util包

在 CommunityConstant 类中添加新的字段，代码如下：

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

    /**
     * 系统用户ID
     */
    int SYSTEM_USER_ID = 1;

}

3. dao层

CommentMapper类中添加 selectCommentById 方法，代码如下：

@Mapper
public interface CommentMapper {
        List<Comment> 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.xml代码如下：

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.nowcoder.community.dao.CommentMapper">

    <sql id="selectFields">
        id, user_id, entity_type, entity_id, target_id, content, status, create_time
    </sql>

    <sql id="insertFields">
        user_id, entity_type, entity_id, target_id, content, status, create_time
    </sql>

    <select id="selectCommentsByEntity" resultType="com.nowcoder.community.entity.Comment">
        select <include refid="selectFields"></include>
        from comment
        where status = 0
        and entity_type = #{entityType}
        and entity_id = #{entityId}
        order by create_time asc
        limit #{offset}, #{limit}
    </select>

    <select id="selectCountByEntity" resultType="int">
        select count(id)
        from comment
        where status = 0
        and entity_type = #{entityType}
        and entity_id = #{entityId}
    </select>


    <insert id="insertComment" parameterType="com.nowcoder.community.entity.Comment">
        insert into comment(<include refid="insertFields"></include>)
        values(#{userId},#{entityType},#{entityId},#{targetId},#{content},#{status},#{createTime})
    </insert>

    <select id = "selectCommentById" resultType="com.nowcoder.community.entity.Comment">
        select <include refid="selectFields"></include>
        from comment
        where id = #{id}
    </select>

</mapper>

4. service包

在 CommentService 类中添加 findCommentById 方法，代码如下：

@Service
public class CommentService implements CommunityConstant {

    @Autowired
    private CommentMapper commentMapper;

    @Autowired
    private SensitiveFilter sensitiveFilter;

    @Autowired
    private DiscussPostService discussPostService;


    public List<Comment> findCommentsByEntity(int entityType, int entityId, int offset, int limit) {
        return commentMapper.selectCommentsByEntity(entityType, entityId, offset, limit);
    }

    public int findCommentCount(int entityType, int entityId) {
        return commentMapper.selectCountByEntity(entityType, entityId);
    }

    @Transactional(isolation = Isolation.READ_COMMITTED, propagation = Propagation.REQUIRED)
    public int addComment(Comment comment) {
        if (comment == null) {
            throw new IllegalArgumentException("参数不能为空！");
        }
        // 添加评论
        comment.setContent(HtmlUtils.htmlEscape(comment.getContent()));
        comment.setContent(sensitiveFilter.filter(comment.getContent()));
        int rows = commentMapper.insertComment(comment);

        // 更新帖子评论数量
        if (comment.getEntityType() == ENTITY_TYPE_POST) {
            int count = commentMapper.selectCountByEntity(comment.getEntityType(), comment.getEntityId());
            discussPostService.updateCommentCount(comment.getEntityId(), count);
        }
        return  rows;
    }

    public Comment findCommentById(int id) {
        return commentMapper.selectCommentById(id);
    }
}

## 5.controller层
在 commentcontroller 类中添加触发评论事件，代码如下：

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

   @Autowired
   private CommentService commentService;

   @Autowired
   private HostHolder hostHolder;

   @Autowired
   private EventProducer eventProducer;

   @Autowired
   private DiscussPostService discussPostService;

   @RequestMapping(value = "/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;
   }
}

在 likecontroller 类中添加触发点赞事件，代码如下：

@Controller
public class LikeController implements CommunityConstant {

    @Autowired
    private LikeService likeService;

    @Autowired
    private HostHolder hostHolder;

    @Autowired
    private EventProducer eventProducer;


    @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<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);  //postId对于点赞的是帖子而言，这个数据是冗余的
            eventProducer.fireEvent(event);
        }


        return CommunityUtil.getJSONString(0, null, map);
    }
}

在 followeController 中添加触发关注事件，代码如下：

@Controller
public class FollowController implements CommunityConstant {

    @Autowired
    private FollowService followService;

    @Autowired
    private HostHolder hostHolder;

    @Autowired
    private UserService userService;

    @Autowired
    private EventProducer eventProducer;



    @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, "已关注！");
    }

    @RequestMapping(path = "/unfollow", method = RequestMethod.POST)
    @ResponseBody
    public String unfollow(int entityType, int entityId) {
        User user = hostHolder.getUser();

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

        return CommunityUtil.getJSONString(0, "已取消关注!");
    }


    @RequestMapping(path = "/followees/{userId}", method = RequestMethod.GET)
    public String getFollowees(@PathVariable("userId") int userId, Page page, Model model) {
        User user = userService.findUserById(userId);
        if (user == null) {
            throw new RuntimeException("该用户不存在!");
        }
        model.addAttribute("user", user);

        page.setLimit(5);
        page.setPath("/followees/" + userId);
        page.setRows((int) followService.findFolloweeCount(userId, ENTITY_TYPE_USER));

        List<Map<String, Object>> userList = followService.findFollowees(userId, page.getOffset(), page.getLimit());
        if (userList != null) {
            for (Map<String, Object> map : userList) {
                User u = (User) map.get("user");
                map.put("hasFollowed", hasFollowed(u.getId()));
            }
        }
        model.addAttribute("users", userList);

        return "/site/followee";
    }

    @RequestMapping(path = "/followers/{userId}", method = RequestMethod.GET)
    public String getFollowers(@PathVariable("userId") int userId, Page page, Model model) {
        User user = userService.findUserById(userId);
        if (user == null) {
            throw new RuntimeException("该用户不存在!");
        }
        model.addAttribute("user", user);

        page.setLimit(5);
        page.setPath("/followers/" + userId);
        page.setRows((int) followService.findFollowerCount(ENTITY_TYPE_USER, userId));

        List<Map<String, Object>> userList = followService.findFollowers(userId, page.getOffset(), page.getLimit());
        if (userList != null) {
            for (Map<String, Object> map : userList) {
                User u = (User) map.get("user");
                map.put("hasFollowed", hasFollowed(u.getId()));
            }
        }
        model.addAttribute("users", userList);

        return "/site/follower";
    }


    private boolean hasFollowed(int userId) {
        if (hostHolder.getUser() == null) {
            return false;
        }

        return followService.hasFollowed(hostHolder.getUser().getId(), ENTITY_TYPE_USER, userId);
    }


}

6. view层

对 discudd-detail.html 中的like 方法进行重构。
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);
            }
        }
    );
}

在这里插入图片描述

五、显示系统通知

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

1. dao层

在 MessageMapper 类中添加 selectLatestNotice、selectNoticeCount、selectNoticeUnreadCount、selectNotices 方法，代码如下：

@Mapper
public interface MessageMapper {

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

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

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

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

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

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

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

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

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

    // 查询未读的通知的数量
    int selectNoticeUnreadCount(int userId, String topic);

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

}

对应的 message-mapper.xml 代码如下：

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.gerrard.community.dao.MessageMapper">

    <sql id="selectFields">
        id, from_id, to_id, conversation_id, content, status, create_time
    </sql>

    <sql id="insertFields">
        from_id, to_id, conversation_id, content, status, create_time
    </sql>

    <select id="selectConversations" resultType="Message">
        select <include refid="selectFields"></include>
        from message
        where id in (
            select max(id) from message
            where status != 2
            and from_id != 1
            and (from_id = #{userId} or to_id = #{userId})
            group by conversation_id
        )
        order by id desc
        limit #{offset}, #{limit}
    </select>

    <select id="selectConversationCount" resultType="int">
        select count(m.maxid) from (
            select max(id) as maxid from message
            where status != 2
            and from_id != 1
            and (from_id = #{userId} or to_id = #{userId})
            group by conversation_id
        ) as m
    </select>

    <select id="selectLetters" resultType="Message">
        select <include refid="selectFields"></include>
        from message
        where status != 2
        and from_id != 1
        and conversation_id = #{conversationId}
        order by id desc
        limit #{offset}, #{limit}
    </select>

    <select id="selectLetterCount" resultType="int">
        select count(id)
        from message
        where status != 2
        and from_id != 1
        and conversation_id = #{conversationId}
    </select>

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

    <insert id="insertMessage" parameterType="Message" keyProperty="id">
        insert into message(<include refid="insertFields"></include>)
        values(#{fromId},#{toId},#{conversationId},#{content},#{status},#{createTime})
    </insert>

    <update id="updateStatus">
        update message set status = #{status}
        where id in
        <foreach collection="ids" item="id" open="(" separator="," close=")">
            #{id}
        </foreach>
    </update>

    <select id="selectLatestNotice" resultType="Message">
        select <include refid="selectFields"></include>
        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>

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

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

    <select id="selectNotices" resultType="Message">
        select <include refid="selectFields"></include>
        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}
    </select>

</mapper>

2. service层

在 MessageService 类中添加 findLatesNotice、findNoticeCount 、findNoticeUnreadCount、findNotices方法，代码如下：

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.gerrard.community.dao.MessageMapper">

    <sql id="selectFields">
        id, from_id, to_id, conversation_id, content, status, create_time
    </sql>

    <sql id="insertFields">
        from_id, to_id, conversation_id, content, status, create_time
    </sql>

    <select id="selectConversations" resultType="Message">
        select <include refid="selectFields"></include>
        from message
        where id in (
            select max(id) from message
            where status != 2
            and from_id != 1
            and (from_id = #{userId} or to_id = #{userId})
            group by conversation_id
        )
        order by id desc
        limit #{offset}, #{limit}
    </select>

    <select id="selectConversationCount" resultType="int">
        select count(m.maxid) from (
            select max(id) as maxid from message
            where status != 2
            and from_id != 1
            and (from_id = #{userId} or to_id = #{userId})
            group by conversation_id
        ) as m
    </select>

    <select id="selectLetters" resultType="Message">
        select <include refid="selectFields"></include>
        from message
        where status != 2
        and from_id != 1
        and conversation_id = #{conversationId}
        order by id desc
        limit #{offset}, #{limit}
    </select>

    <select id="selectLetterCount" resultType="int">
        select count(id)
        from message
        where status != 2
        and from_id != 1
        and conversation_id = #{conversationId}
    </select>

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

    <insert id="insertMessage" parameterType="Message" keyProperty="id">
        insert into message(<include refid="insertFields"></include>)
        values(#{fromId},#{toId},#{conversationId},#{content},#{status},#{createTime})
    </insert>

    <update id="updateStatus">
        update message set status = #{status}
        where id in
        <foreach collection="ids" item="id" open="(" separator="," close=")">
            #{id}
        </foreach>
    </update>

    <select id="selectLatestNotice" resultType="Message">
        select <include refid="selectFields"></include>
        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>

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

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

    <select id="selectNotices" resultType="Message">
        select <include refid="selectFields"></include>
        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}
    </select>

</mapper>

3. controller层

在 MessageController 类中添加 getNoticeList、getNoticerDetail 方法，代码如下：

@Controller
public class MessageController implements CommunityConstant {

    @Autowired
    private MessageService messageService;

    @Autowired
    private HostHolder hostHolder;

    @Autowired
    private UserService userService;

    // 私信列表
    @RequestMapping(path = "/letter/list", method = RequestMethod.GET)
    public String getLetterList(Model model, Page page) {
        User user = hostHolder.getUser();
        // 分页信息
        page.setLimit(5);
        page.setPath("/letter/list");
        page.setRows(messageService.findConversationCount(user.getId()));

        // 会话列表
        List<Message> conversationList = messageService.findConversations(
                user.getId(), page.getOffset(), page.getLimit());
        List<Map<String, Object>> conversations = new ArrayList<>();
        if (conversationList != null) {
            for (Message message : conversationList) {
                Map<String, Object> map = new HashMap<>();
                map.put("conversation", message);
                map.put("letterCount", messageService.findLetterCount(message.getConversationId()));
                map.put("unreadCount", messageService.findLetterUnreadCount(user.getId(), message.getConversationId()));
                int targetId = user.getId() == message.getFromId() ? message.getToId() : message.getFromId();
                map.put("target", userService.findUserById(targetId));

                conversations.add(map);
            }
        }
        model.addAttribute("conversations", conversations);

        // 查询未读消息数量
        int letterUnreadCount = messageService.findLetterUnreadCount(user.getId(), null);
        model.addAttribute("letterUnreadCount", letterUnreadCount);
        int noticeUnreadCount = messageService.findNoticeUnreadCount(user.getId(), null);
        model.addAttribute("noticeUnreadCount", noticeUnreadCount);

        return "/site/letter";
    }

    @RequestMapping(path = "/letter/detail/{conversationId}", method = RequestMethod.GET)
    public String getLetterDetail(@PathVariable("conversationId") String conversationId, Page page, Model model) {
        // 分页信息
        page.setLimit(5);
        page.setPath("/letter/detail/" + conversationId);
        page.setRows(messageService.findLetterCount(conversationId));

        // 私信列表
        List<Message> letterList = messageService.findLetters(conversationId, page.getOffset(), page.getLimit());
        List<Map<String, Object>> letters = new ArrayList<>();
        if (letterList != null) {
            for (Message message : letterList) {
                Map<String, Object> map = new HashMap<>();
                map.put("letter", message);
                map.put("fromUser", userService.findUserById(message.getFromId()));
                letters.add(map);
            }
        }
        model.addAttribute("letters", letters);

        // 私信目标
        model.addAttribute("target", getLetterTarget(conversationId));

        // 设置已读
        List<Integer> ids = getLetterIds(letterList);
        if (!ids.isEmpty()) {
            messageService.readMessage(ids);
        }

        return "/site/letter-detail";
    }

    private User getLetterTarget(String conversationId) {
        String[] ids = conversationId.split("_");
        int id0 = Integer.parseInt(ids[0]);
        int id1 = Integer.parseInt(ids[1]);

        if (hostHolder.getUser().getId() == id0) {
            return userService.findUserById(id1);
        } else {
            return userService.findUserById(id0);
        }
    }

    private List<Integer> getLetterIds(List<Message> letterList) {
        List<Integer> ids = new ArrayList<>();

        if (letterList != null) {
            for (Message message : letterList) {
                if (hostHolder.getUser().getId() == message.getToId() && message.getStatus() == 0) {
                    ids.add(message.getId());
                }
            }
        }

        return ids;
    }

    @RequestMapping(path = "/letter/send", method = RequestMethod.POST)
    @ResponseBody
    public String sendLetter(String toName, String content) {
        User target = userService.findUserByName(toName);
        if (target == null) {
            return CommunityUtil.getJSONString(1, "目标用户不存在!");
        }

        Message message = new Message();
        message.setFromId(hostHolder.getUser().getId());
        message.setToId(target.getId());
        if (message.getFromId() < message.getToId()) {
            message.setConversationId(message.getFromId() + "_" + message.getToId());
        } else {
            message.setConversationId(message.getToId() + "_" + message.getFromId());
        }
        message.setContent(content);
        message.setCreateTime(new Date());
        messageService.addMessage(message);

        return CommunityUtil.getJSONString(0);
    }

    @RequestMapping(path = "/notice/list", method = RequestMethod.GET)
    public String getNoticeList(Model model) {
        User user = hostHolder.getUser();

        // 查询评论类通知
        Message message = messageService.findLatestNotice(user.getId(), TOPIC_COMMENT);
        Map<String, Object> messageVO = new HashMap<>();
        if (message != null) {
            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);
        messageVO = new HashMap<>();
        if (message != null) {
            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);
        messageVO = new HashMap<>();
        if (message != null) {
            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";
    }

}

旧版：新建 MessageInterceptor 类拦截器。
新版：在 LoginTicketinterceptor 中对 posthandle 方法进行修改，代码如下：

@Component
public class LoginTicketInterceptor implements HandlerInterceptor {
    @Autowired
    private UserService userService;

    @Autowired
    private HostHolder hostHolder;

    @Autowired
    private MessageService messageService;

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        //从cookie中获取凭证
        String ticket= CookieUtil.getValue(request,"ticket");
        if(ticket!=null){
            //查询凭证
            LoginTicket loginTicket=userService.findLoginTicket(ticket);

            //检查凭证是否有效
            //一般不会检测到过期，因为时间一到，cookie自动被销毁了？？？？？
            if(loginTicket!=null && loginTicket.getStatus()==0 && loginTicket.getExpired().after(new Date())){
                //根据凭证查询用户
                User user=userService.findUserById(loginTicket.getUserId());
                //在本次请求中持有用户
                hostHolder.setUser(user);

                //构建用户认证的结果，并存入SecurityContext，以便于Security进行授权
                Authentication authentication=new UsernamePasswordAuthenticationToken(
                        user,user.getPassword(),userService.getAuthorities(user.getId()));

                SecurityContextHolder.setContext(new SecurityContextImpl(authentication));
            }
        }
        return true;
    }

    @Override
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
        User user=hostHolder.getUser();

        if(user!=null && modelAndView !=null){
            int unReadLetterCount=messageService.findLetterUnreadCount(user.getId(),null);
            int unReadNoticeCount=messageService.findNoticeUnreadCount(user.getId(),null);
            modelAndView.addObject("loginUser",user);
            modelAndView.addObject("messageUnRead",unReadLetterCount+unReadNoticeCount);
        }
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        hostHolder.clear();

       // SecurityContextHolder.clearContext();
    }
}

4. view层

前端页面代码的修改在项目汇总中可进行查看。

5. 功能测试

在这里插入图片描述

创作不易，如果有帮助到你，请给题解==点个赞和收藏==，让更多的人看到！！！
==关注博主==不迷路，内容持续更新中。

仿牛客社区项目（第五章）（总）

第三章：Kafka，构建TB级异步消息系统

一、阻塞队列

1. 阻塞队列测试方法

2. 测试结果

二、Kafka入门

1. Kafka下载

2. Kafka安装与配置

3. Kafka的启动

4. Kafka使用

三、Spring整合Kafka

1. 引入依赖

2. 配置Kafka

3. 测试

4. 测试结果

四、发送系统消息

1. 实体类

2. util包

3. dao层

4. service包

6. view层

五、显示系统通知

1. dao层

2. service层

3. controller层

4. view层

5. 功能测试

5. 功能测试

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

仿牛客社区项目（第五章）（总）

第三章：Kafka，构建TB级异步消息系统

一、阻塞队列

1. 阻塞队列测试方法

2. 测试结果

二、Kafka入门

1. Kafka下载

2. Kafka安装与配置

3. Kafka的启动

4. Kafka使用

三、Spring整合Kafka

1. 引入依赖

2. 配置Kafka

3. 测试

4. 测试结果

四、发送系统消息

1. 实体类

2. util包

3. dao层

4. service包

6. view层

五、显示系统通知

1. dao层

2. service层

3. controller层

4. view层

5. 功能测试

5. 功能测试

热门文章

最新文章

相关课程

相关电子书

相关实验场景