声明

本文提炼于个人练手项目，其中的实现逻辑不一定标准，实现思路没有参考权威的文档和教程，仅为个人思考得出，因此可能存在较多本人未考虑到的情况和漏洞，因此仅供参考，如果大家觉得有问题，恳请大家指出有问题的地方

如果对客户端的实现感兴趣，可以转身查看【UniApp开发小程序】私聊功能uniapp界面实现 (买家、卖家沟通商品信息)【后端基于若依管理系统开发】

聊天数据查询管理

数据库设计

【私信表】
在这里插入图片描述

Vo

package com.ruoyi.common.core.domain.vo;

import lombok.Data;

import java.util.Date;

/**
 * @Author dam
 * @create 2023/8/22 21:39
 */
@Data
public class ChatUserVo {
    private Long userId;
    private String userAvatar;
    private String userName;
    private String userNickname;
    /**
     * 最后一条消息的内容
     */
    private String lastChatContent;
    /**
     * 最后一次聊天的日期
     */
    private Date lastChatDate;
    /**
     * 未读消息数量
     */
    private Integer unReadChatNum;
}

Controller

其中两个方法较为重要，介绍如下：

listChatUserVo：当用户进入消息界面的时候，需要查询出最近聊天的用户，其中还需要展示一些信息，如ChatUserVo的属性
listChat：该方法用于查询对方最近和自己的私聊内容，当用户查询了这些私聊内容，默认用户已经看过了，将这些私聊内容设置为已读状态

package com.shm.controller;

import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import javax.servlet.http.HttpServletResponse;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ruoyi.common.core.domain.entity.Chat;
import com.ruoyi.common.core.domain.vo.ChatUserVo;
import com.shm.service.IChatService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.ruoyi.common.annotation.Log;
import com.ruoyi.common.core.controller.BaseController;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.common.utils.poi.ExcelUtil;
import com.ruoyi.common.core.page.TableDataInfo;

/**
 * 聊天数据Controller
 *
 * @author dam
 * @date 2023-08-19
 */
@RestController
@RequestMapping("/market/chat")
@Api
public class ChatController extends BaseController {
    @Autowired
    private IChatService chatService;

    /**
     * 查询聊天数据列表
     */
    @PreAuthorize("@ss.hasPermi('market:chat:list')")
    @GetMapping("/list")
    public TableDataInfo list(Chat chat) {
        startPage();
        List<Chat> list = chatService.list(new QueryWrapper<Chat>(chat));
        return getDataTable(list);
    }

    /**
     * 查询最近和自己聊天的用户
     */
    @ApiOperation("listChatUserVo")
    @PreAuthorize("@ss.hasPermi('market:chat:list')")
    @GetMapping("/listChatUserVo")
    public TableDataInfo listChatUserVo() {
        startPage();
        String username = getLoginUser().getUsername();
        List<ChatUserVo> list = chatService.listChatUserVo(username);
        return getDataTable(list);
    }

    /**
     * 查询用户和自己最近的聊天信息
     */
    @ApiOperation("listUsersChatWithMe")
    @PreAuthorize("@ss.hasPermi('market:chat:list')")
    @GetMapping("/listChat/{toUsername}")
    public TableDataInfo listChat(@PathVariable("toUsername") String toUsername) {
        String curUsername = getLoginUser().getUsername();
        startPage();
        List<Chat> list = chatService.listChat(curUsername, toUsername);
        for (Chat chat : list) {
            System.out.println("chat:"+chat.toString());
        }
        System.out.println();
        // 查出的数据，如果消息是对方发的，且是未读状态，重新设置为已读
        List<Long> unReadIdList = list.stream().filter(
                        (item1) -> {
                            if (item1.getIsRead() == 0 && item1.getFromWho().equals(toUsername)) {
                                return true;
                            } else {
                                return false;
                            }
                        }
                )
                .map(item2 -> {
                    return item2.getId();
                }).collect(Collectors.toList());
        System.out.println("将"+ unReadIdList.toString()+"设置为已读");
        if (unReadIdList.size() > 0) {
            // 批量设置私聊为已读状态
            chatService.batchRead(unReadIdList);
        }
        return getDataTable(list);
    }

    /**
     * 导出聊天数据列表
     */
    @PreAuthorize("@ss.hasPermi('market:chat:export')")
    @Log(title = "聊天数据", businessType = BusinessType.EXPORT)
    @PostMapping("/export")
    public void export(HttpServletResponse response, Chat chat) {
        List<Chat> list = chatService.list(new QueryWrapper<Chat>(chat));
        ExcelUtil<Chat> util = new ExcelUtil<Chat>(Chat.class);
        util.exportExcel(response, list, "聊天数据数据");
    }

    /**
     * 获取聊天数据详细信息
     */
    @PreAuthorize("@ss.hasPermi('market:chat:query')")
    @GetMapping(value = "/getInfo/{id}")
    public AjaxResult getInfo(@PathVariable("id") Long id) {
        return success(chatService.getById(id));
    }

    /**
     * 新增聊天数据
     */
    @PreAuthorize("@ss.hasPermi('market:chat:add')")
    @Log(title = "聊天数据", businessType = BusinessType.INSERT)
    @PostMapping
    public AjaxResult add(@RequestBody Chat chat) {
        return toAjax(chatService.save(chat));
    }

    /**
     * 修改聊天数据
     */
    @PreAuthorize("@ss.hasPermi('market:chat:edit')")
    @Log(title = "聊天数据", businessType = BusinessType.UPDATE)
    @PutMapping
    public AjaxResult edit(@RequestBody Chat chat) {
        return toAjax(chatService.updateById(chat));
    }

    /**
     * 删除聊天数据
     */
    @PreAuthorize("@ss.hasPermi('market:chat:remove')")
    @Log(title = "聊天数据", businessType = BusinessType.DELETE)
    @DeleteMapping("/{ids}")
    public AjaxResult remove(@PathVariable List<Long> ids) {
        return toAjax(chatService.removeByIds(ids));
    }
}

Service

package com.shm.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.ruoyi.common.core.domain.entity.Chat;
import com.ruoyi.common.core.domain.vo.ChatUserVo;
import com.shm.mapper.ChatMapper;
import com.shm.service.IChatService;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @author 17526
 * @description 针对表【chat(聊天数据表)】的数据库操作Service实现
 * @createDate 2023-08-19 21:12:49
 */
@Service
public class IChatServiceImpl extends ServiceImpl<ChatMapper, Chat>
        implements IChatService {

    /**
     * 查询最近和自己聊天的用户
     *
     * @return
     */
    @Override
    public List<ChatUserVo> listChatUserVo(String username) {
        return baseMapper.listChatUserVo(username);
    }

    /**
     * 查询用户和自己最近的聊天信息
     *
     * @param curUsername
     * @param toUsername
     * @return
     */
    @Override
    public List<Chat> listChat(String curUsername, String toUsername) {
        return baseMapper.listChat(curUsername, toUsername);
    }

    @Override
    public void batchRead(List<Long> unReadIdList) {
        baseMapper.batchRead(unReadIdList);
    }
}

Mapper

package com.shm.mapper;

import com.ruoyi.common.core.domain.entity.Chat;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.ruoyi.common.core.domain.vo.ChatUserVo;
import org.apache.ibatis.annotations.Param;

import java.util.List;

/**
* @author 17526
* @description 针对表【chat(聊天数据表)】的数据库操作Mapper
* @createDate 2023-08-19 21:12:49
* @Entity com.ruoyi.common.core.domain.entity.Chat
*/
public interface ChatMapper extends BaseMapper<Chat> {

    List<ChatUserVo> listChatUserVo(@Param("username") String username);

    List<Chat> listChat(@Param("curUsername") String curUsername, @Param("toUsername") String toUsername);

    void batchRead(@Param("unReadIdList") List<Long> unReadIdList);
}

【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.shm.mapper.ChatMapper">

    <resultMap id="BaseResultMap" type="com.ruoyi.common.core.domain.entity.Chat">
            <id property="id" column="id" jdbcType="BIGINT"/>
            <result property="createTime" column="create_time" jdbcType="TIMESTAMP"/>
            <result property="updateTime" column="update_time" jdbcType="TIMESTAMP"/>
            <result property="isDeleted" column="is_deleted" jdbcType="TINYINT"/>
            <result property="fromWho" column="from_who" jdbcType="BIGINT"/>
            <result property="toWho" column="to_who" jdbcType="BIGINT"/>
            <result property="content" column="content" jdbcType="VARCHAR"/>
            <result property="picUrl" column="pic_url" jdbcType="VARCHAR"/>
    </resultMap>

    <sql id="Base_Column_List">
        id,create_time,update_time,
        is_deleted,from,to,
        content,pic_url
    </sql>
    <update id="batchRead">
        update chat set is_read = 1 where id in
        <foreach collection="unReadIdList" item="chatId" separator="," open="(" close=")">
            #{chatId}
        </foreach>
    </update>
    <select id="listChatUserVo" resultType="com.ruoyi.common.core.domain.vo.ChatUserVo">
        SELECT
            (CASE WHEN c.from_who=#{username} THEN c.to_who ELSE c.from_who END) AS `userName`,
            c.content AS `lastChatContent`,
            c.create_time AS lastChatDate,
            u.user_id AS userId,
            u.avatar AS userAvatar,
            u.nick_name AS userNickname,
            ur.unReadNum as unReadChatNum
        FROM
            (SELECT
                 MAX(`id`) AS chatId,
                 CASE
                     WHEN `from_who` = #{username} THEN `to_who`
                     ELSE `from_who`
                     END AS uname
             FROM `chat`
             WHERE `from_who` = #{username} OR `to_who` = #{username}
             GROUP BY uname) AS t
                INNER JOIN `chat` c ON c.id = t.chatId
                LEFT JOIN `sys_user` u ON t.uname = u.user_name
                LEFT JOIN (SELECT from_who, SUM(CASE WHEN is_read=1 THEN 0 ELSE 1 END) AS unReadNum FROM chat WHERE is_deleted=0 AND to_who = #{username} GROUP BY from_who) ur ON ur.from_who = t.uname
        ORDER BY c.create_time DESC
    </select>
    <select id="listChat" resultType="com.ruoyi.common.core.domain.entity.Chat">
        SELECT
            *
        FROM
            chat
        WHERE
            ( from_who = #{curUsername} AND to_who = #{toUsername} )
           OR ( to_who = #{curUsername} AND from_who = #{toUsername} )
        ORDER BY
            create_time DESC
    </select>
</mapper>

【查询最近聊天的用户的用户名和那条消息的id】
因为id是自增的，所以最新的那条消息的id肯定最大，因此可以使用MAX(id)来获取最近的消息

SELECT 
        MAX(`id`) AS chatId,
				 CASE 
            WHEN `from_who` = 'admin' THEN `to_who`
            ELSE `from_who`
        END AS uname
    FROM `chat`
    WHERE `from_who` = 'admin' OR `to_who` = 'admin'
    GROUP BY uname

在这里插入图片描述
【内连接私信表获取消息的其他信息】

INNER JOIN `chat` c ON c.id = t.chatId

【左连接用户表获取用户的相关信息】

LEFT JOIN `sys_user` u ON t.uname = u.user_name

【左联接私信表获取未读对方消息的数量】
CASE WHEN is_read=1 THEN 0 ELSE 1 END 如果已读，说明未读数量为0；否则为1

LEFT JOIN (SELECT from_who, SUM(CASE WHEN is_read=1 THEN 0 ELSE 1 END) AS unReadNum FROM chat WHERE is_deleted=0 AND to_who = 'admin' GROUP BY from_who) ur ON ur.from_who = t.uname

【最后按照用户和自己最后聊天的时间来降序排序】

ORDER BY c.create_time DESC

WebSocket引入

为什么使用WebSocket

WebSocket不仅支持客户端向服务端发送消息，同时也支持服务端向客户端发送消息，这样才能完成私聊的功能。即
用户1-->服务端-->用户2

依赖

<!-- websocket -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

配置类

package com.shm.config;

import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
public class WebSocketConfig {

    /**
     * 注入一个ServerEndpointExporter,
     * 该Bean会自动注册使用@ServerEndpoint注解 声明的websocket endpoint
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

}

WebSocket服务

需要注意的是，Websocket是多例模式，无法直接使用@Autowired注解来注入rabbitTemplate，需要使用下面的方式，其中rabbitTemplate为静态变量

private static RabbitTemplate rabbitTemplate;

 @Autowired
 public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
     WebSocketServer.rabbitTemplate = rabbitTemplate;
 }

package com.shm.component;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.common.core.domain.entity.Chat;
import com.shm.component.delay.DelayQueueManager;
import com.shm.component.delay.DelayTask;
import com.shm.constant.RabbitMqConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @author websocket服务
 */
@ServerEndpoint(value = "/websocket/{username}")
@Component//将WebSocketServer注册为spring的一个bean
public class WebSocketServer {

    private static final Logger log = LoggerFactory.getLogger(WebSocketServer.class);

    /**
     * 记录当前在线连接的客户端的session
     */
    public static final Map<String, Session> usernameAndSessionMap = new ConcurrentHashMap<>();
    /**
     * 记录正在进行的聊天的发出者和接收者
     */
    public static final Map<String, Integer> fromToMap = new ConcurrentHashMap<>();
    /**
     * 用户Session保留时间，如果超过该时间，用户还没有给服务端发送消息，认为用户下线，删除其Session
     * 注意：该时间需要比客户端的心跳时间更长
     */
    private static final long expire = 6000;

    // websocket为多例模式，无法直接注入，需要换成下面的方式
//    @Autowired
//    RabbitTemplate rabbitTemplate;

    private static RabbitTemplate rabbitTemplate;

    @Autowired
    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        WebSocketServer.rabbitTemplate = rabbitTemplate;
    }

    @Autowired
    private static DelayQueueManager delayQueueManager;

    @Autowired
    public void setDelayQueueManager(DelayQueueManager delayQueueManager) {
        WebSocketServer.delayQueueManager = delayQueueManager;
    }

    /**
     * 浏览器和服务端连接建立成功之后会调用这个方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("username") String username) {
        usernameAndSessionMap.put(username, session);
        // 建立延时任务，如果到expire时间，客户端还是没有和服务器有任何交互的话，就删除该用户的session，表示该用户下线
        delayQueueManager.put(new DelayTask(username, expire));
        log.info("有新用户加入，username={}, 当前在线人数为：{}", username, usernameAndSessionMap.size());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session, @PathParam("username") String username) {
        usernameAndSessionMap.remove(username);
        log.info("有一连接关闭，移除username={}的用户session, 当前在线人数为：{}", username, usernameAndSessionMap.size());
    }

    /**
     * 发生错误的时候会调用这个方法
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误");
        error.printStackTrace();
    }

    /**
     * 服务端发送消息给客户端
     */
    public void sendMessage(String message, Session toSession) {
        try {
            log.info("服务端给客户端[{}]发送消息{}", toSession.getId(), message);
            toSession.getBasicRemote().sendText(message);
        } catch (Exception e) {
            log.error("服务端发送消息给客户端失败", e);
        }
    }


    /**
     * onMessage方法是一个消息的中转站
     * 1、首先接受浏览器端socket.send发送过来的json数据
     * 2、然后解析其数据，找到消息要发送给谁
     * 3、最后将数据发送给相应的人
     *
     * @param message 客户端发送过来的消息 数据格式：{"from":"user1","to":"admin","text":"你好呀"}
     */
    @OnMessage
    public void onMessage(String message, Session session, @PathParam("username") String username) {
//        log.info("服务端接收到 {} 的消息，消息内容是:{}", username, message);

        // 收到用户的信息，删除之前的延时任务，创建新的延时任务
        delayQueueManager.put(new DelayTask(username, expire));
        if (!usernameAndSessionMap.containsKey(username)) {
            // 可能用户挂机了一段时间，被下线了，后面又重新回来发信息了，需要重新将用户和session添加字典中
            usernameAndSessionMap.put(username, session);
        }

        // 将json字符串转化为json对象
        JSONObject obj = JSON.parseObject(message);
        String status = (String) obj.get("status");
        // 获取消息的内容
        String text = (String) obj.get("text");
        // 查看消息要发送给哪个用户
        String to = (String) obj.get("to");
        String fromToKey = username + "-" + to;
        String toFromKey = to + "-" + username;
        if (status != null) {
            if (status.equals("start")) {
                fromToMap.put(fromToKey, 1);
            } else if (status.equals("end")) {
                System.out.println("移除销毁的fromToKey:" + fromToKey);
                fromToMap.remove(fromToKey);
            } else if (status.equals("ping")) {
                // 更新用户对应的时间戳
//                usernameAndTimeStampMap.put(username, System.currentTimeMillis());
            }
        } else {
            // 封装数据发送给消息队列
            Chat chat = new Chat();
            chat.setFromWho(username);
            chat.setToWho(to);
            chat.setContent(text);
            chat.setIsRead(0);
            //        chat.setPicUrl("");

            // 根据to来获取相应的session，然后通过session将消息内容转发给相应的用户
            Session toSession = usernameAndSessionMap.get(to);
            if (toSession != null) {
                JSONObject jsonObject = new JSONObject();
                // 设置消息来源的用户名
                jsonObject.put("from", username);
                // 设置消息内容
                jsonObject.put("text", text);
                // 服务端发送消息给目标客户端
                this.sendMessage(jsonObject.toString(), toSession);
                log.info("发送消息给用户 {} ，消息内容是：{} ", toSession, jsonObject.toString());
                if (fromToMap.containsKey(toFromKey)) {
                    chat.setIsRead(1);
                }
            } else {
                log.info("发送失败，未找到用户 {} 的session", to);
            }

            rabbitTemplate.convertAndSend(RabbitMqConstant.CHAT_STORAGE_EXCHANGE, RabbitMqConstant.CHAT_STORAGE_ROUTER_KEY, chat);
        }
    }

}

RabbitMQ引入

为什么使用消息队列

在用户之间进行聊天的时候，需要将用户的聊天数据存储到数据库中，但是如果大量用户同时在线的话，可能同一时间发送的消息数量太多，如果同时将这些消息存储到数据库中，会给数据库带来较大的压力，使用RabbitMQ可以先把要存储的数据放到消息队列，然后数据库服务器压力没这么大的时候，就会从消息队列中获取数据来存储，这样可以分散数据库的压力。但是如果用户是直接从数据库获取消息的话，消息可能有一定的延迟，如果用户之间正在聊天的话，消息则不会延迟，因为聊天内容会立刻通过WebSocket发送给对方。

依赖

<!-- rabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

启动类添加注解

在启动类的上方添加@EnableRabbit注解
在这里插入图片描述

常量类

因为有多处会使用到队列命名等信息，创建一个常量类来保存相关信息

package com.shm.constant;

public class RabbitMqConstant {
    public static final String CHAT_STORAGE_QUEUE = "shm.chat-storage.queue";
    public static final String CHAT_STORAGE_EXCHANGE = "shm.chat-storage-event-exchange";
    public static final String CHAT_STORAGE_ROUTER_KEY = "shm.chat-storage.register";
}

使用配置类创建队列、交换机、绑定关系

package com.shm.config;

import com.shm.constant.RabbitMqConstant;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MyRabbitConfig {

    /**
     * 使用JSON序列化机制，进行消息转换
     * @return
     */
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 私信存储队列
     *
     * @return
     */
    @Bean
    public Queue chatStorageQueue() {
        Queue queue = new Queue(RabbitMqConstant.CHAT_STORAGE_QUEUE, true, false, false);
        return queue;
    }

    /**
     * 私信存储交换机
     * 创建交换机，由于只需要一个队列，创建direct交换机
     *
     * @return
     */
    @Bean
    public Exchange chatStorageExchange() {
        //durable：持久化
        return new DirectExchange(RabbitMqConstant.CHAT_STORAGE_EXCHANGE, true, false);
    }

    /**
     * 创建私信存储 交换机和队列的绑定关系
     *
     * @return
     */
    @Bean
    public Binding chatStorageBinding() {
        return new Binding(RabbitMqConstant.CHAT_STORAGE_QUEUE,
                Binding.DestinationType.QUEUE,
                RabbitMqConstant.CHAT_STORAGE_EXCHANGE,
                RabbitMqConstant.CHAT_STORAGE_ROUTER_KEY,
                null);
    }

}

消息监听器

创建一个消息监听类来监听队列的消息，然后调用相关的逻辑来处理信息，本文主要的处理是将私信内容存储到数据库中

package com.shm.listener;

import com.rabbitmq.client.Channel;
import com.ruoyi.common.core.domain.entity.Chat;
import com.shm.constant.RabbitMqConstant;
import com.shm.service.IChatService;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;

@Service
/**
 * 注意，类上面需要RabbitListener注解
 */
@RabbitListener(queues = RabbitMqConstant.CHAT_STORAGE_QUEUE)
public class ChatStorageListener {
    @Autowired
    private IChatService chatService;

    @RabbitHandler
    public void handleStockLockedRelease(Chat chat, Message message, Channel channel) throws IOException {
        try {
            boolean save = chatService.save(chat);
            //解锁成功，手动确认，消息才从MQ中删除
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            //只要有异常，拒绝消息，让消息重新返回队列，让别的消费者继续解锁
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        }

    }

}

发送消息到消息队列

WebSocketServer为Websocket后端服务代码，其中的onMessage方法会接受客户端发送过来的消息，当接收到消息的时候，将消息发送给消息队列

// 封装数据发送给消息队列
Chat chat = new Chat();
chat.setFromWho(username);
chat.setToWho(to);
chat.setContent(text);
chat.setPicUrl("");
rabbitTemplate.convertAndSend(RabbitMqConstant.CHAT_STORAGE_EXCHANGE,RabbitMqConstant.CHAT_STORAGE_ROUTER_KEY,chat);

延时任务

为什么使用延时任务

为了更好地感知用户的在线状态，在用户连接了WebSocket或者发送消息之后，建立一个延时任务，如果到达了所设定的延时时间，就删除用户的Session，认为用户已经下线；如果在延时期间之内，用户发送了新消息，或者发送了心跳信号，证明该用户还处于在线状态，删除前面的延时任务，并创建新的延时任务

延时任务类

package com.shm.component.delay;

import lombok.Data;
import lombok.Getter;

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * @Author dam
 * @create 2023/8/25 15:12
 */
@Getter
public class DelayTask implements Delayed {

    /**
     * 用户名
     */
    private final String userName;
    /**
     * 任务的真正执行时间
     */
    private final long executeTime;
    /**
     * 任务延时多久执行
     */
    private final long expire;

    /**
     * @param expire 任务需要延时的时间
     */
    public DelayTask(String userName, long expire) {
        this.userName = userName;
        this.executeTime = expire + System.currentTimeMillis();
        this.expire = expire;
    }

    /**
     * 根据给定的时间单位，返回与此对象关联的剩余延迟时间
     * 
     * @param unit the time unit 时间单位
     * @return 返回剩余延迟，零值或负值表示延迟已经过去
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.executeTime - System.currentTimeMillis(), unit);
    }

    @Override
    public int compareTo(Delayed o) {
        return 0;
    }
}

延时任务管理

package com.shm.component.delay;

import com.shm.component.WebSocketServer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Executors;

/**
 * @Author dam
 * @create 2023/8/25 15:12
 */
@Component
@Slf4j
public class DelayQueueManager implements CommandLineRunner {
    private final DelayQueue<DelayTask> delayQueue = new DelayQueue<>();
    private final Map<String, DelayTask> usernameAndDelayTaskMap = new ConcurrentHashMap<>();

    /**
     * 加入到延时队列中
     *
     * @param task
     */
    public void put(DelayTask task) {
        // 因为一个用户只能对应一个延时任务，所以如果已经存在了延时任务，将其进行删除
        if (usernameAndDelayTaskMap.containsKey(task.getUserName())) {
            this.remove(task.getUserName());
        }
        delayQueue.put(task);
        usernameAndDelayTaskMap.put(task.getUserName(), task);
    }

    /**
     * 取消延时任务
     *
     * @param username 要删除的任务的用户名
     * @return
     */
    public boolean remove(String username) {
        DelayTask remove = usernameAndDelayTaskMap.remove(username);
        return delayQueue.remove(remove);
    }

    @Override
    public void run(String... args) throws Exception {
        this.executeThread();
    }

    /**
     * 延时任务执行线程
     */
    private void executeThread() {
        while (true) {
            try {
                DelayTask task = delayQueue.take();
                //执行任务
                processTask(task);
            } catch (InterruptedException e) {
                break;
            }
        }
    }

    /**
     * 执行延时任务
     *
     * @param task
     */
    private void processTask(DelayTask task) {
        // 删除该用户的session，表示用户下线
        WebSocketServer.usernameAndSessionMap.remove(task.getUserName());
        log.error("执行定时任务：{}下线", task.getUserName());
    }
    
}