springboot SseEmitter 消息推送

这是SpringMVC提供的一种技术,可以实现服务端向客户端实时推送数据.用法非常简单,只需要在Controller提供一个接口,创建并返回SseEmitter对象,发送数据可以在另一个接口调用其send方法发送数据.
SpringBoot已经集成了这个 ,所以不用再引其他依赖
废话不多说直接贴代码
服务端

package ;

import org.;
import org.;
import ;
import ;

import ;
import ;
import ;
import ;
import ;
import ;
import ;


public class SseEmitterServer {

    private static final Logger logger = ();

    /**
     * 当前连接数
     */
    private static AtomicInteger count = new AtomicInteger(0);

    /**
     * 使用map对象，便于根据userId来获取对应的SseEmitter，或者放redis里面
     */
    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    /**
     * 创建用户连接并返回 SseEmitter
     *
     * @param userId 用户ID
     * @return SseEmitter
     */
    public static SseEmitter connect(String userId) {
        // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 注册回调
        (completionCallBack(userId));
        (errorCallBack(userId));
        (timeoutCallBack(userId));
        (userId, sseEmitter);
        // 数量+1
        ();
        ("创建新的sse连接，当前用户：{}", userId);
        return sseEmitter;
    }

    /**
     * 给指定用户发送信息
     */
    public static void sendMessage(String userId, String message) {
        if ((userId)) {
            try {
                // (userId).send(message, MediaType.APPLICATION_JSON);
                (userId).send(message);
            } catch (IOException e) {
                ("用户[{}]推送异常:{}", userId, ());
                removeUser(userId);
            }
        }
    }

    /**
     * 群发消息
     */
    public static void batchSendMessage(String wsInfo, List<String> ids) {
        (userId -> sendMessage(wsInfo, userId));
    }

    /**
     * 群发所有人
     */
    public static void batchSendMessage(String wsInfo) {
        ((k, v) -> {
            try {
                (wsInfo, MediaType.APPLICATION_JSON);
            } catch (IOException e) {
                ("用户[{}]推送异常:{}", k, ());
                removeUser(k);
            }
        });
    }

    /**
     * 移除用户连接
     */
    public static void removeUser(String userId) {
        (userId);
        // 数量-1
        ();
        ("移除用户：{}", userId);
    }

    /**
     * 获取当前连接信息
     */
    public static List<String> getIds() {
        return new ArrayList<>(());
    }

    /**
     * 获取当前连接数量
     */
    public static int getUserCount() {
        return ();
    }

    private static Runnable completionCallBack(String userId) {
        return () -> {
            ("结束连接：{}", userId);
            removeUser(userId);
        };
    }

    private static Runnable timeoutCallBack(String userId) {
        return () -> {
            ("连接超时：{}", userId);
            removeUser(userId);
        };
    }

    private static Consumer<Throwable> errorCallBack(String userId) {
        return throwable -> {
            ("连接异常：{}", userId);
            removeUser(userId);
        };
    }

}

服务端接口

package ;

import ;
import .*;
import ;

@CrossOrigin
@RestController
@RequestMapping("/sse")
public class TestController {
    /**
     * 用于创建连接
     */
    @GetMapping("/connect/{userId}")
    public SseEmitter connect(@PathVariable String userId) {
        return (userId);
    }

    /**
     * 推送给所有人
     *
     * @param message
     * @return
     */
    @GetMapping("/push/{message}")
    public ResponseEntity<String> push(@PathVariable(name = "message") String message) {
        (message);
        return ("WebSocket 推送消息给所有人");
    }

    /**
     * 发送给单个人
     *
     * @param message
     * @param userid
     * @return
     */
    @GetMapping("/push_one/{messsage}/{userid}")
    public ResponseEntity<String> pushOne(@PathVariable(name = "message") String message, @PathVariable(name = "userid") String userid) {
        (userid, message);
        return ("WebSocket 推送消息给" + userid);
    }

    /**
     * 关闭连接
     */
    @GetMapping("/close/{userid}")
    public ResponseEntity<String> close(@PathVariable("userid") String userid) {
        (userid);
        return ("连接关闭");
    }

}

前端页面

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <title>SseEmitter</title>
</head>

<body>
    <button onclick="closeSse()">关闭连接</button>
    <div ></div>
</body>
<script>
    let source = null;

    // 用时间戳模拟登录用户
    const userId = new Date().getTime();

    if () {

        // 建立连接
        source = new EventSource('http://localhost:8080/sse/connect/' + userId);

        /**
         * 连接一旦建立，就会触发open事件
         * 另一种写法： = function (event) {}
         */
        ('open', function(e) {
            setMessageInnerHTML("建立连接。。。");
        }, false);

        /**
         * 客户端收到服务器发来的数据
         * 另一种写法： = function (event) {}
         */
        ('message', function(e) {
            setMessageInnerHTML();
        });


        /**
         * 如果发生通信错误（比如连接中断），就会触发error事件
         * 或者：
         * 另一种写法： = function (event) {}
         */
        ('error', function(e) {
            if ( === ) {
                setMessageInnerHTML("连接关闭");
            } else {
                (e);
            }
        }, false);

    } else {
        setMessageInnerHTML("你的浏览器不支持SSE");
    }

    // 监听窗口关闭事件，主动去关闭sse连接，如果服务端设置永不过期，浏览器关闭后手动清理服务端数据
     = function() {
        closeSse();
    };

    // 关闭Sse连接
    function closeSse() {
        ();
        const httpRequest = new XMLHttpRequest();
        ('GET', 'http://localhost:8080/sse/close/' + userId, true);
        ();
        ("close");
    }

    // 将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        ('message').innerHTML += innerHTML + '<br/>';
    }
</script>

</html>

秒客网

springboot SseEmitter 消息推送

相关文章