1、引入websocket jar包
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
2、启动类添加websocket相关注解
@EnableWebSocket//开启webSocket配置
3、配置文件设置
3.1、独立的容器，每次链接都会生成一个新的容器，但是websocket实现类不能注入spring其他的bean。
@ConditionalOnWebApplication
@Configuration
public class WebSocketConfig {
/**
* 配置WebSocketEndpointServer
* 如果使用独立的servlet容器，不是使用SpringBoot的内置容器
* 不需要注入ServerEndpointExporter, 它将由容器自己提供和管理
* @return
*/
@Bean
public ServerEndpointExporter serverEndpointExporter () {
return new ServerEndpointExporter();
}
}
3.2、集成到spring的容器，这个配置的问题就是，每次链接会生成同一个容器，新的容器会吧旧的替换掉，spring bean可以注入到实现类中去。可以根据应用场景的不同选择不同的配置。
@ConditionalOnWebApplication
@Configuration
public class WebSocketConfig extends ServerEndpointConfig.Configurator implements ApplicationContextAware{
private static volatile BeanFactory context;
/**
* 配置WebSocketEndpointServer
* 如果使用独立的servlet容器，不是使用SpringBoot的内置容器
* 不需要注入ServerEndpointExporter, 它将由容器自己提供和管理
* @return
*/
@Bean
public ServerEndpointExporter serverEndpointExporter () {
return new ServerEndpointExporter();
}
@Override
public <T> T getEndpointInstance(Class<T> clazz) throws InstantiationException
{
return context.getBean(clazz);
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
System.out.println("auto load"+this.hashCode());
WebSocketConfig.context = applicationContext;
}
}
4、endpoint配置（聊天室）
@Component
@ServerEndpoint("/endpoint")
public class SimpleEndpoint {
/**
* 日志
*/
private static final Logger _Logger = LoggerFactory.getLogger(SimpleEndpoint.class);
/**
* 线程安全基本数据
*/
private static AtomicInteger COUNT = new AtomicInteger();
/**
* 在线用户实例
*/
private static CopyOnWriteArraySet<SimpleEndpoint> container = new CopyOnWriteArraySet<SimpleEndpoint>();
/**
* 当前用户Session
*/
private Session session;
//获取连接时调用
@OnOpen
public void onOpen (Session session) {
this.session = session;
container.add(this);
addUser();
}
//连接关闭时调用
@OnClose
public void onClose () {
//从容器中删除用户
container.remove(this);
//减去用户数量
subUser();
}
//接受WebSocket发送的消息
@OnMessage
public void onMessage (String message, Session session) {
_Logger.info("Send all user: {}", message);
container.forEach(target -> target.sendMessage(message));
}
//发生错误时调用
@OnError
public void onError (Session session, Throwable error) {
_Logger.info("error: {}", error.getMessage());
}
/**
* 发送消息
* @param message
*/
public void sendMessage (String message) {
try {
this.session.getBasicRemote().sendText(message);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 自定义发送消息给全部用户
* @param message
*/
public static void sendUsers (String message) {
_Logger.info("Send all user: {}", message);
container.forEach(target -> target.sendMessage(message));
}
/**
* 减去用户数量
*/
public void subUser () {
_Logger.info("Sub user: {}", COUNT.get());
SimpleEndpoint.COUNT.addAndGet(-1);
}
/**
* 添加用户数量
*/
public void addUser () {
_Logger.info("Add user: {}", COUNT.get());
SimpleEndpoint.COUNT.addAndGet(1);
}
}
5、websocket实现（聊天室）如果websocket容器集成到spring中去，只会生成一个websocket，最好用独立的websocket容器配置
@ServerEndpoint(value ="/websocket/{sid}/{rsid}",configurator=WebSocketConfig.class)
@Component
public class WebSocketServer {
static Logger log= LoggerFactory.getLogger(WebSocketServer.class);
//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
private static int onlineCount = 0;
//concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();
//与某个客户端的连接会话，需要通过它来给客户端发送数据
private Session session;
//发送用户id
private String sid="";
//接受用户id
private String rsid="";
/**
* 连接建立成功调用的方法*/
@OnOpen
public void onOpen(Session session,@PathParam("sid") String sid,@PathParam("rsid") String rsid) {
this.session = session;
webSocketSet.add(this); //加入set中
addOnlineCount(); //在线数加1
log.info("有新窗口开始监听:"+sid+",当前在线人数为" + getOnlineCount());
this.sid=sid;
this.rsid = rsid;
try {
sendMessage("连接成功");
} catch (IOException e) {
log.error("websocket IO异常");
}
}
/**
* 连接关闭调用的方法
*/
@OnClose
public void onClose() {
webSocketSet.remove(this); //从set中删除
subOnlineCount(); //在线数减1
log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
}
/**
* 收到客户端消息后调用的方法
*
* @param message 客户端发送过来的消息*/
@OnMessage
public void onMessage(String message, Session session) {
session.getId();
//群发消息
// for (WebSocketServer item : webSocketSet) {
// try {
// item.sendMessage(message);
// } catch (IOException e) {
// e.printStackTrace();
// }
// }
/**
* 只给聊天的用户跟商家发送消息
*/
for (WebSocketServer item : webSocketSet) {
if(item.sid.equals(rsid)||item.sid.equals(sid)){
try {
item.sendMessage(sid+":"+message);
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/**
*
* @param session
* @param error
*/
@OnError
public void onError(Session session, Throwable error) {
log.error("发生错误");
error.printStackTrace();
}
/**
* 实现服务器主动推送
*/
public void sendMessage(String message) throws IOException {
this.session.getBasicRemote().sendText(message);
}
/**
* 群发自定义消息
* */
public static void sendInfo(String message,@PathParam("sid") String sid) throws IOException {
log.info("推送消息到窗口"+sid+"，推送内容:"+message);
for (WebSocketServer item : webSocketSet) {
try {
//这里可以设定只推送给这个sid的，为null则全部推送
if(sid==null) {
item.sendMessage(message);
}else if(item.sid.equals(sid)||item.sid.equals("22")){
item.sendMessage("xx店客服:"+message);
}
} catch (IOException e) {
continue;
}
}
}
public static synchronized int getOnlineCount() {
return onlineCount;
}
public static synchronized void addOnlineCount() {
WebSocketServer.onlineCount++;
}
public static synchronized void subOnlineCount() {
WebSocketServer.onlineCount--;
}
}
网友评论