WebSocket的使用

在最近的项目中，遇见一个问题，需要在消息有新增时，能够实时更新消息的未读数量。
如下图所示：

这样的话就需要服务端在有消息新增时主动推送未读数量给客户端，我们可以采用Ajax的轮询，或者采用websocket，这里我选择采用websocket。

1、什么是websocket

通俗易懂的讲websocket就是给我们提供一个全双工相互通信，实现服务端可以主动推送信息给客户端。

2、配置websocket

本次运用是基于Springboot框架的实现。
2.1 WebsocketAutoConfig.java

@Configuration
public class WebsocketAutoConfig {
//    注入ServerEndpointExporter，这个bean会自动注册使用了@ServerEndpoint注解声明的WebsocketEndpoint
    @Bean
    public ServerEndpointExporter endpointExporter() {
        return new ServerEndpointExporter();
    }
}
1
2
3
4
5
6
7
8

2.2 WebSocketConfig.java

@Configuration //声明为配置类，并交给spring管理
@EnableWebSocket //开启WebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        //如果不是注解形式使用WebSocket,需要在该处进行声明,否则为空即可
        //WebSocketDemo实现了WebSocketHandler接口
        //registry.addHandler(new WebSocketDemo(), "/webSocket");
    }
}
1
2
3
4
5
6
7
8
9
10
11

2.3 WebSocketService.java

@Component
@Slf4j
@ServerEndpoint(value = "/ws/pushMessage/{userId}")
public class WebSocketService {

    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static AtomicInteger onlineNum = new AtomicInteger();

    //concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。
    private static ConcurrentHashMap<Long, Session> sessionPools = new ConcurrentHashMap<>();

    //建立连接成功调用
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") Long userId) throws IOException, EncodeException {
        if (sessionPools.containsKey(userId)){
            log.warn("客户端程序{}已有连接,无需建立连接", userId);
            return;
        }

        sessionPools.put(userId, session);
        addOnlineCount();
        log.info(userId + "加入webSocket！当前人数为" + onlineNum);
        int count = SpringUtils.getBean(PropertyMessageService.class).countAllMessage(userId);
        try {
           sendMessage(session, count+"");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //发送消息
    public void sendMessage(Session session, String message) throws IOException,EncodeException{
        if (session != null) {
            synchronized (session) {
                log.info("发送数据：" + message);
                session.getBasicRemote().sendText(message);
            }
        }
    }

    //给指定用户发送信息
    public void sendInfo(Long userId, String message) {
        Session session = sessionPools.get(userId);
        try {
            sendMessage(session, message);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //多个人发送给指定的几个用户
    public void SendMany(String msg, Long[] userIds) {
        try {
            for (Long userId : userIds) {
                sendMessage(sessionPools.get(userId), msg);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //信息群发
    public void sendAll(String msg) {
        try {
            for (Long key : sessionPools.keySet()) {
                sendMessage(sessionPools.get(key), msg);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    //关闭连接时调用
    @OnClose
    public void onClose(@PathParam(value = "userId") Long userId) {
        sessionPools.remove(userId);
        subOnlineCount();
        log.info(userId + "断开webSocket连接！当前人数为" + onlineNum);
    }

    //收到客户端消息后调用的方法
    // @param message客户端发送过来的消息
    @OnMessage
    public void onMessage(String message) throws IOException {
        message = "客户端发送消息：" + message + ",服务端已收到";
        log.info(message);
        for (Session session : sessionPools.values()) {
            try {
                sendMessage(session, message);
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }
        }
    }

    //错误时调用
    @OnError
    public void onError(Session session, Throwable throwable) {
        System.out.println("发生错误");
        throwable.printStackTrace();
    }


    //加入连接
    public static void addOnlineCount() {
        onlineNum.incrementAndGet();
    }

    //移除连接
    public static void subOnlineCount() {
        onlineNum.decrementAndGet();
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

3、运行websocket

3.1问题解决
当我们跟前端建立连接之后，便可以在业务实现里面实现主动推送，我们这里是在连接成功之后就给客户端实现一次推送也就是上面的 @OnOpen注解下的方法，这里我遇见一个问题，我发现通过@Resource注解或@Autowired注解的方式将相关的service注入到spring容器，是比webSocket的@OnOpen下的方法后执行的因此我选择用自己编写的方法拿到bean“ int count = SpringUtils.getBean(PropertyMessageService.class).countAllMessage(userId);”，便能成功在建立连接之后推送未读数给客服端。
3.2消息新增时实现推送

    @Resource
    private PropertyMessageService propertyMessageService;
    private static PropertyMessageService myPropertyMessageService;

    @Resource
    private WebSocketService webSocketService;
    private static WebSocketService myWebSocketService;

    @Resource
    private PropertyMessageMapper propertyMessageMapper;
    private static PropertyMessageMapper myPropertyMessageMapper;

    @PostConstruct
    public void init(){
        myPropertyMessageService = propertyMessageService;
        myWebSocketService = webSocketService;
        myPropertyMessageMapper = propertyMessageMapper;
    }

public static void savePropertyMessage(MessageDto dto) {
        myPropertyMessageService.save(createPropertyMessage(dto.getMessageTemplate(),
                dto.getDataId(),dto.getMessageType(),dto.getParams(),dto.getReadStatus(),dto.getHandleId()));

        Long id = SecurityUtils.getUserId();
        int i = myPropertyMessageMapper.countAllMessage(id);
        Integer count  = (Math.min(i, 99));
        myWebSocketService.sendInfo(dto.getHandleId(),count.toString());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

通过以上的操作成功实现主动推送消息。