@Configuration
@EnableWebSocket // 啟動Websocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(myHandler(), "/myHandler/**")
            // 添加攔截器，可以獲取連接的param和 header 用作認證鑒權(quán)
            .addInterceptors(new LakerSessionHandshakeInterceptor())
            // 設(shè)置運行跨域
            .setAllowedOrigins("*");
    }
    @Bean
    public ServletServerContainerFactoryBean createWebSocketContainer() {
        ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();
        // 設(shè)置默認會話空閑超時 以毫秒為單位 非正值意味著無限超時，默認值 0 ，默認沒10s檢查一次空閑就關(guān)閉
        container.setMaxSessionIdleTimeout(10 * 1000L);
        // 設(shè)置異步發(fā)送消息的默認超時時間 以毫秒為單位 非正值意味著無限超時 ，默認值-1，還沒看到作用
        container.setAsyncSendTimeout(10 * 1000L);
        // 設(shè)置文本消息的默認最大緩沖區(qū)大小 以字符為單位，默認值 8 * 1024
        container.setMaxTextMessageBufferSize(8 * 1024);
        // 設(shè)置二進制消息的默認最大緩沖區(qū)大小 以字節(jié)為單位，默認值 8 * 1024
        container.setMaxBinaryMessageBufferSize(8 * 1024);
        return container;
    }
    @Bean
    public WebSocketHandler myHandler() {
        return new MyHandler();
    }
}

public class LakerSessionHandshakeInterceptor extends HttpSessionHandshakeInterceptor {
    // 攔截器返回false，則不會進行websocket協(xié)議的轉(zhuǎn)換
    // 可以獲取請求參數(shù)做認證鑒權(quán)
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        HttpServletRequest servletRequest = ((ServletServerHttpRequest) request).getServletRequest();
        // 將所有查詢參數(shù)復(fù)制到 WebSocketSession 屬性映射
        Enumeration<String> parameterNames = servletRequest.getParameterNames();
        while (parameterNames.hasMoreElements()) {
            String parameterName = parameterNames.nextElement();
            // 放入的值可以從WebSocketHandler里面的WebSocketSession拿出來
            attributes.put(parameterName, servletRequest.getParameter(parameterName));
        }
        if (servletRequest.getHeader(HttpHeaders.AUTHORIZATION) != null) {
            setErrorResponse(response, HttpStatus.UNAUTHORIZED);
            return false;
        }
        return true;
    }
    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
    }
    private void setErrorResponse(ServerHttpResponse response, HttpStatus status, String errorMessage) {
        response.setStatusCode(status);
        if (errorMessage != null) {
            try {
                objectMapper.writeValue(response.getBody(), new ErrorDetail(status.value(), errorMessage));
            } catch (IOException ioe) {
            }
        }
    }
}

public class MyHandler extends AbstractWebSocketHandler {
     private final Map<String, WebSocketSession> webSocketSessionMap = new ConcurrentHashMap<>();
    //成功連接時
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        super.afterConnectionEstablished(session);
        // 設(shè)置最大報文大小，如果報文過大則會報錯的,可以將限制調(diào)大。
        // 會覆蓋config中的配置。
        session.setBinaryMessageSizeLimit(8 * 1024);
        session.setTextMessageSizeLimit(8 * 1024);
        webSocketSessionMap.put(session.getId(), session);
    }
    //處理textmessage
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        super.handleTextMessage(session, message);
        // 有消息就廣播下
        for (Map.Entry<String, WebSocketSession> entry : webSocketSessionMap.entrySet()) {
            String s = entry.getKey();
            WebSocketSession webSocketSession = entry.getValue();
            if (webSocketSession.isOpen()) {
            	webSocketSession.sendMessage(new TextMessage(s + ":" + message.getPayload()));
            	log.info("send to {} msg:{}", s, message.getPayload());
            }
        }
    }
    //報錯時
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        super.handleTransportError(session, exception);
          log.warn("handleTransportError:: sessionId: {} {}", session.getId(), exception.getMessage(), exception);
        if (session.isOpen()) {
            try {
                session.close();
            } catch (Exception ex) {
                // ignore
            }
        }
    }
   //連接關(guān)閉時
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        super.afterConnectionClosed(session, status);
        WebSocketSession session =  webSocketSessionMap.remove(session.getId());
        if (session.isOpen()) {
            try {
                session.close();
            } catch (Exception ex) {
                // ignore
            }
        }
    }
    //處理binarymessage
    @Override
    protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
        super.handleBinaryMessage(session, message);
    }
    //處理pong
    @Override
    protected void handlePongMessage(WebSocketSession session, PongMessage message) throws Exception {
        super.handlePongMessage(session, message);
    }
    //是否支持報文拆包
    // 決定是否接受半包，因為websocket協(xié)議比較底層，好像Tcp協(xié)議一樣，如果發(fā)送大消息可能會拆成多個小報文。如果不希望處理不完整的報文，希望底層幫忙聚合成完整消息將此方法返回false,這樣底層會等待完整報文到達聚合后才回調(diào)。
    @Override
    public boolean supportsPartialMessages() {
        return super.supportsPartialMessages();
    }
}

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        log.info("{} Connection established!", session.getId());
        // webSocketSessionMap.put(session.getId(), session);
        // 把線程安全的session代理裝飾類放到容器里
        webSocketSessionMap.put(session.getId(), new ConcurrentWebSocketSessionDecorator(session, 10 * 1000, 64000));
    }

// 包裝一個WebSocketSession以保證一次只有一個線程可以發(fā)送消息。
// 如果發(fā)送速度慢，后續(xù)嘗試從其他線程發(fā)送更多消息將無法獲取刷新鎖，消息將被緩沖。屆時，將檢查指定的緩沖區(qū)大小限制和發(fā)送時間限制，如果超過限制，會話將關(guān)閉。
public class ConcurrentWebSocketSessionDecorator extends WebSocketSessionDecorator {
	private static final Log logger = LogFactory.getLog(ConcurrentWebSocketSessionDecorator.class);
	private final int sendTimeLimit;
	private final int bufferSizeLimit;
	private final OverflowStrategy overflowStrategy;
	@Nullable
	private Consumer<WebSocketMessage<?>> preSendCallback;
	private final Queue<WebSocketMessage<?>> buffer = new LinkedBlockingQueue<>();
	private final AtomicInteger bufferSize = new AtomicInteger();
	private volatile long sendStartTime;
	private volatile boolean limitExceeded;
	private volatile boolean closeInProgress;
	private final Lock flushLock = new ReentrantLock();
	private final Lock closeLock = new ReentrantLock();
	public ConcurrentWebSocketSessionDecorator(WebSocketSession delegate, int sendTimeLimit, int bufferSizeLimit) {
		this(delegate, sendTimeLimit, bufferSizeLimit, OverflowStrategy.TERMINATE);
	}
    // delegate 需要代理的session
    // sendTimeLimit 表示發(fā)送**單個消息**的最大時間
    // bufferSizeLimit 表示發(fā)送消息的隊列最大字節(jié)數(shù)，不是消息的數(shù)量而是消息的總大小
    // overflowStrategy 表示超過限制時的策略有2個
                 // - 斷開連接（默認選項）
                 // - 丟棄最老的數(shù)據(jù)，直到大小滿足
	public ConcurrentWebSocketSessionDecorator(
			WebSocketSession delegate, int sendTimeLimit, int bufferSizeLimit, OverflowStrategy overflowStrategy) {
		super(delegate);
		this.sendTimeLimit = sendTimeLimit;
		this.bufferSizeLimit = bufferSizeLimit;
		this.overflowStrategy = overflowStrategy;
	}
    // 返回自當前發(fā)送開始以來的時間（毫秒），如果當前沒有發(fā)送正在進行則返回 0。
    // 即花費的耗時
	public long getTimeSinceSendStarted() {
		long start = this.sendStartTime;
		return (start > 0 ? (System.currentTimeMillis() - start) : 0);
	}
	// 設(shè)置在將消息添加到發(fā)送緩沖區(qū)后調(diào)用的回調(diào)
	public void setMessageCallback(Consumer<WebSocketMessage<?>> callback) {
		this.preSendCallback = callback;
	}
	@Override
	public void sendMessage(WebSocketMessage<?> message) throws IOException {
        //檢查超限了就不發(fā)了
		if (shouldNotSend()) {
			return;
		}
		// 消息放到buffer隊列
		this.buffer.add(message);
        // 增加bufferSize用于后面判斷是不是超限了
		this.bufferSize.addAndGet(message.getPayloadLength());
		// 發(fā)送緩沖區(qū)后調(diào)用的回調(diào)
		if (this.preSendCallback != null) {
			this.preSendCallback.accept(message);
		}
		do {
            // 嘗試獲取鎖，發(fā)送消息，只有一個線程負責發(fā)送所有消息
			if (!tryFlushMessageBuffer()) {
				// 沒獲取鎖的線程，對當前buffer和時間檢查，
                // 檢查不過就拋異常 然后框架自己會抓取異常關(guān)閉當前的連接
				checkSessionLimits();
				break;
			}
		}
		while (!this.buffer.isEmpty() && !shouldNotSend());
	}
	// 超限了 不能發(fā)送了
	private boolean shouldNotSend() {
		return (this.limitExceeded || this.closeInProgress);
	}
	// 嘗試獲取鎖 并發(fā)送所有緩存的消息
	private boolean tryFlushMessageBuffer() throws IOException {
		if (this.flushLock.tryLock()) {
			try {
                 // 循環(huán)發(fā)送消息
				while (true) {
                    // 一次拉一個消息
					WebSocketMessage<?> message = this.buffer.poll();
                    // 沒消息了 或者 超限了
					if (message == null || shouldNotSend()) {
                        // 退出 完活
						break;
					}
                    // 釋放bufferSize
					this.bufferSize.addAndGet(-message.getPayloadLength());
                    // 用于判斷單個消息是否發(fā)送超時的
					this.sendStartTime = System.currentTimeMillis();
                    // 發(fā)送消息
					getDelegate().sendMessage(message);
                    // 重置開始時間
					this.sendStartTime = 0;
				}
			}
			finally {
				this.sendStartTime = 0;
				this.flushLock.unlock();
			}
			return true;
		}
		return false;
	}
   //檢查是否超時，是否超過buffer限制
	private void checkSessionLimits() {
        // 應(yīng)該發(fā)送 且 獲取到關(guān)閉鎖
		if (!shouldNotSend() && this.closeLock.tryLock()) {
			try {
                 //檢測是否發(fā)送超時
				if (getTimeSinceSendStarted() > getSendTimeLimit()) {
					String format = "Send time %d (ms) for session '%s' exceeded the allowed limit %d";
					String reason = String.format(format, getTimeSinceSendStarted(), getId(), getSendTimeLimit());
					limitExceeded(reason);
				}
                 //檢測buffer大小，根據(jù)策略要么拋異常關(guān)閉連接，要么丟棄數(shù)據(jù)
				else if (getBufferSize() > getBufferSizeLimit()) {
					switch (this.overflowStrategy) {
                        // 關(guān)閉連接，拋出異?？蚣芫蜁P(guān)閉連接    
						case TERMINATE:
							String format = "Buffer size %d bytes for session '%s' exceeds the allowed limit %d";
							String reason = String.format(format, getBufferSize(), getId(), getBufferSizeLimit());
							limitExceeded(reason);
							break;
                        // 丟棄老數(shù)據(jù)    
						case DROP:
							int i = 0;
							while (getBufferSize() > getBufferSizeLimit()) {
								WebSocketMessage<?> message = this.buffer.poll();
								if (message == null) {
									break;
								}
								this.bufferSize.addAndGet(-message.getPayloadLength());
								i++;
							}
							if (logger.isDebugEnabled()) {
								logger.debug("Dropped " + i + " messages, buffer size: " + getBufferSize());
							}
							break;
						default:
							// Should never happen..
							throw new IllegalStateException("Unexpected OverflowStrategy: " + this.overflowStrategy);
					}
				}
			}
			finally {
				this.closeLock.unlock();
			}
		}
	}
	private void limitExceeded(String reason) {
		this.limitExceeded = true;
		throw new SessionLimitExceededException(reason, CloseStatus.SESSION_NOT_RELIABLE);
	}
	@Override
	public void close(CloseStatus status) throws IOException {
		this.closeLock.lock();
		try {
			if (this.closeInProgress) {
				return;
			}
			if (!CloseStatus.SESSION_NOT_RELIABLE.equals(status)) {
				try {
					checkSessionLimits();
				}
				catch (SessionLimitExceededException ex) {
					// Ignore
				}
				if (this.limitExceeded) {
					if (logger.isDebugEnabled()) {
						logger.debug("Changing close status " + status + " to SESSION_NOT_RELIABLE.");
					}
					status = CloseStatus.SESSION_NOT_RELIABLE;
				}
			}
			this.closeInProgress = true;
			super.close(status);
		}
		finally {
			this.closeLock.unlock();
		}
	}
	public enum OverflowStrategy {
		TERMINATE,
		DROP
	}
}