server:  
  port: ${sp.publish}  
spring:  
  application:  
    name: ${sn.publish}  
  cloud:  
    stream:  
      rocketmq:  
        binder:  
          name-server: ${rocket-mq.name-server}  
        bindings:  
          output:  
            producer:  
              group: socketGroup  
              sync: true  
      bindings:  
        output:  
          destination: test-topic  
          content-type: application/json  
#开启MQ的日志  
logging:  
  level:  
    com:  
      alibaba:  
        cloud:  
          stream:  
            binder:  
              rocketmq: DEBUG

<dependency>  
    <groupId>com.alibaba.cloud</groupId>  
    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>  
</dependency>

/**  
 * <p> * 描述：Swagger配置  
 * </p>  
 * @author hamhuo  
 * @version 1.0.0  
 */@Configuration  
@EnableSwagger2WebMvc  
public class SwaggerConfig {  
    @Bean  
    Docket rpApi() {  
        return SwaggerCore.defaultDocketBuilder("消息推送", "com.zeroone.star.mq.controller", "publish");  
    }  
}

/**  
 * <p> * 描述：消息发布组件  
 * </p>  
 * @author hamhuo  
 * @version 1.0.0  
 */@Component  
public class RmqPublish {  
  
    @Resource  
    Source source;  
  
    public void publish(SampleNotifyDTO dto) {  
        source.output().send(MessageBuilder.withPayload(dto).build());  
    }  
}

  
/**  
 * <p> * 描述：示例通知数据对象  
 * </p>  
 * @author hamhuo  
 * @version 1.0.0  
 */@Data  
public class SampleNotifyDTO {  
    /**  
     * 客户端编号  
     */  
    private String clientId;  
  
    /**  
     * 通知消息内容  
     */  
    private String message;  
}

rocketmq:  
  binder:  
    name-server: ${rocket-mq.name-server}  
  bindings:  
    output:  
      producer:  
        group: socketGroup  
        sync: true  
bindings:  
  output:  
    destination: test-topic  
    content-type: application/json

bindings:  
  output:  
    destination: test-topic  
    content-type: application/json

@Component
public class AuthFilter implements GlobalFilter, Ordered

 @Component
public class AuthFilter implements GlobalFilter, Ordered{
	//常量, 记录开始访问的时间
	private static final String BEGIN_VISIT_TIME = "begin_visit_time";//开始访问时间

	//补上日志实例
	private static final Logger log = LoggerFactory.getLogger(AuthFilter.class);
	
//过滤器接口方法, 这个必须实现
	@Override  
	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
	//token和白名单稍后实现
	
	//在上下文加上开始时间
	exchange.getAttributes().put(BEGIN_VISIT_TIME, System.currentTimeMillis());

	//等待链执行完成回调
	return chain.filter(exchange)
	//链式调用then, 使用Mono.fromRunnable方法顺序执行日志输出
	.then(Mono.fromRunnable{})
		//日志输出的逻辑
		// 记录接口访问日志  
		Long beginVisitTime = exchange.getAttribute(BEGIN_VISIT_TIME);
		if (beginVisitTime != null){
			logData.put("duration", (System.currentTimeMillis() - beginVisitTime) + "ms");
			log.info("访问接口信息: {}", logData);
		}
	}
}

 @Component
public class AuthFilter implements GlobalFilter, Ordered{
	//常量, 记录开始访问的时间
	private static final String BEGIN_VISIT_TIME = "begin_visit_time";//开始访问时间

	//补上日志
	private static final Logger log = LoggerFactory.getLogger(AuthFilter.class);
	
//过滤器接口方法
	@Override  
	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
	//白名单
	//首先拿到请求路径
	//上下文拿请求
	ServerHttpRequest request = exchange.getRequest();
	//请求拿请求路径, 先拿URI, 再去掉首部
	String url = request.getURI().getPath();
	
	// 跳过不需要验证的路径, 直接进下一级
	if (StringUtils.matches(url, ignoreWhite.getWhites())) {  
	    return chain.filter(exchange);  
	}
	
	//在上下文加上开始时间
	exchange.getAttributes().put(BEGIN_VISIT_TIME, System.currentTimeMillis());

	//等待链执行完成回调
	return chain.filter(exchange)
	//链式调用then, 使用Mono.fromRunnable方法顺序执行日志输出
	.then(Mono.fromRunnable{})
		//日志输出的逻辑
		// 记录接口访问日志  
		Long beginVisitTime = exchange.getAttribute(BEGIN_VISIT_TIME);
		if (beginVisitTime != null){
			logData.put("duration", (System.currentTimeMillis() - beginVisitTime) + "ms");
			log.info("访问接口信息: {}", logData);
		}
	}
}

 @Component
public class AuthFilter implements GlobalFilter, Ordered{
	//常量, 记录开始访问的时间
	private static final String BEGIN_VISIT_TIME = "begin_visit_time";//开始访问时间

	//补上日志
	private static final Logger log = LoggerFactory.getLogger(AuthFilter.class);
	
	//补上redis
	@Autowired  
	private RedisService redisService;
//过滤器接口方法
	@Override  
	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
	//白名单
	//首先拿到请求路径
	//上下文拿请求
	ServerHttpRequest request = exchange.getRequest();
	//请求拿请求路径, 先拿URI, 再去掉首部
	String url = request.getURI().getPath();
	
	// 跳过不需要验证的路径, 直接进下一级, 返回链的执行结果
	if (StringUtils.matches(url, ignoreWhite.getWhites())) {  
	    return chain.filter(exchange);  
	}
	
	//剩下的请求过筛子
	//从请求拿token
	String token = getToken(request);
	if (StringUtils.isEmpty(token)) {  
    return unauthorizedResponse(exchange, "令牌不能为空");  
	}
	
	//拿JWT的声明, 这里写了一个字符串工具拿声明, 令牌秘钥就在工具里
	Claims claims = JwtUtils.parseToken(token);
	if (claims == null) {  
	    return unauthorizedResponse(exchange, "令牌已过期或验证不正确！");  
	}
	
	//从声明拿用户ID
	String userkey = JwtUtils.getUserKey(claims);  
	//从缓存匹配用户ID的token, 这里getTokenKey格式化, 就加了个前缀
	boolean islogin = redisService.hasKey(getTokenKey(userkey));
	if (!islogin) {  
    return unauthorizedResponse(exchange, "登录状态已过期");  
	}
	
	//最后从声明拿用户详细信息
	String userid = JwtUtils.getUserId(claims);  
	String username = JwtUtils.getUserName(claims);  
	if (StringUtils.isEmpty(userid) || StringUtils.isEmpty(username)) {  
	    return unauthorizedResponse(exchange, "令牌验证失败");  
	}
	
	//重新设置请求, 鉴权完毕
	//设置用户信息到请求  
	addHeader(mutate, SecurityConstants.USER_KEY, userkey);  
	addHeader(mutate, SecurityConstants.DETAILS_USER_ID, userid);  
	addHeader(mutate, SecurityConstants.DETAILS_USERNAME, username);  
	// 内部请求来源参数清除（防止网关携带内部请求标识，造成系统安全风险）  
	removeHeader(mutate, SecurityConstants.FROM_SOURCE);
	//在上下文加上开始时间
	exchange.getAttributes().put(BEGIN_VISIT_TIME, System.currentTimeMillis());

	//等待链执行完成回调
	return chain.filter(exchange)
	//链式调用then, 使用Mono.fromRunnable方法顺序执行日志输出
	.then(Mono.fromRunnable{})
		//日志输出的逻辑
		// 记录接口访问日志  
		Long beginVisitTime = exchange.getAttribute(BEGIN_VISIT_TIME);  
		if (beginVisitTime != null) {  
		    URI uri = exchange.getRequest().getURI();  
		    Map<String, Object> logData = new HashMap<>();  
		    logData.put("host", uri.getHost());  
		    logData.put("port", uri.getPort());  
		    logData.put("path", uri.getPath());  
		    logData.put("query", uri.getRawQuery());  
		    logData.put("duration", (System.currentTimeMillis() - beginVisitTime) + "ms");  
  
		    log.info("访问接口信息: {}", logData);  
		    log.info("我是美丽分割线: ###################################################");  
		}
	}
}

private void addHeader(ServerHttpRequest.Builder mutate, String name, Object value) {  
    if (value == null) {  
        return;  
    }  
    String valueStr = value.toString();  
    String valueEncode = ServletUtils.urlEncode(valueStr);  
    mutate.header(name, valueEncode);  
}  
  
private void removeHeader(ServerHttpRequest.Builder mutate, String name) {  
    mutate.headers(httpHeaders -> httpHeaders.remove(name)).build();  
}

{
  "姓名": "张三",
  "角色": "管理员",
  "到期时间": "2018年7月1日0点0分"
}

public class HeaderInterceptor implements AsyncHandlerInterceptor

public class HeaderInterceptor implements AsyncHandlerInterceptor {  
  
    @Override  
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {  }
    
    
    @Override  
public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex)  
        throws Exception {  
}
    

public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {  
	  //不是控制器方法不拦截
    if (!(handler instanceof HandlerMethod)) {  
        return true;  
    }  
	//自定义holder 用来设置信息  

SecurityContextHolder.setUserId(ServletUtils.getHeader(request, SecurityConstants.DETAILS_USER_ID));  
    SecurityContextHolder.setUserName(ServletUtils.getHeader(request, SecurityConstants.DETAILS_USERNAME));  
    SecurityContextHolder.setUserKey(ServletUtils.getHeader(request, SecurityConstants.USER_KEY));
    
    
    //根据Token创建用户
    String token = SecurityUtils.getToken();  
    if (StringUtils.isNotEmpty(token)) {  
        LoginUser loginUser = AuthUtil.getLoginUser(token);  
        if (StringUtils.isNotNull(loginUser)) {  
            AuthUtil.verifyLoginUserExpire(loginUser);  
            SecurityContextHolder.set(SecurityConstants.LOGIN_USER, loginUser);  
        }  
    } else {  
        // 首页免登场景展示  
        // 检查请求路径是否匹配特定路径  
        String requestURI = request.getRequestURI();  
        if (isExemptedPath(requestURI)) {  
            // 创建一个默认的 LoginUser 对象  
            LoginUser defaultLoginUser = createDefaultLoginUser();  
            SecurityContextHolder.set(SecurityConstants.LOGIN_USER, defaultLoginUser);  
        }  
    }  
    return true;  
}

public class SecurityContextHolder{
//TTL
private static final TransmittableThreadLocal<Map<String, Object>> THREAD_LOCAL = new TransmittableThreadLocal<>();


//设置方法, 在线程Map中存储私有信息
public static void set(String key, Object value)  
{  
    Map<String, Object> map = getLocalMap();  
    map.put(key, value == null ? StringUtils.EMPTY : value);  
}

//这里只就一个例子, 存储固定的用户名前缀和用户名的方法
public static void setUserName(String username)  
{  
    set(SecurityConstants.DETAILS_USERNAME, username);  
}

}

static {  
    // 在这里添加所有需要免登录默认展示首页的的路径  
    EXEMPTED_PATHS.add("/system/user/getInfo");  
    EXEMPTED_PATHS.add("/project/statistics");  
    EXEMPTED_PATHS.add("/project/doing");  
    EXEMPTED_PATHS.add("/project/queryMyTaskList");  
    EXEMPTED_PATHS.add("/project/select");  
    EXEMPTED_PATHS.add("/system/menu/getRouters");  
  
}

// 需要免登录的路径集合  
private static final Set<String> EXEMPTED_PATHS = new HashSet<>();

// 判断请求路径是否匹配特定路径, 也就是免密白名单  
private boolean isExemptedPath(String requestURI) {  
    // 你可以根据需要调整特定路径的匹配逻辑  
    return EXEMPTED_PATHS.stream().anyMatch(requestURI::startsWith);  
}

// 创建一个默认的 LoginUser 对象  
private LoginUser createDefaultLoginUser() {  
    LoginUser defaultLoginUser = new LoginUser();  
    defaultLoginUser.setUserId(173L);  // 设置默认的用户ID  
    defaultLoginUser.setUsername(Constants.DEMO_ACCOUNT);  // 设置默认的用户名  
  
    SysUser demoSysUser = new SysUser();  
    demoSysUser.setUserId(173L);  
    demoSysUser.setUserName(Constants.DEMO_ACCOUNT);  
    demoSysUser.setDeptId(100L);  
    demoSysUser.setStatus("0");  
  
    defaultLoginUser.setUser(demoSysUser);  
    // 设置其他必要的默认属性  
    return defaultLoginUser;  
}  
//结束后清理用户信息, 防止线程复用导致内存泄露
@Override  
public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex)  
        throws Exception {  
    SecurityContextHolder.remove();  
}

GET /chat HTTP/1.1
Host: www.example.com
Upgrade: websocket
Connection: Upgrade

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade

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

/**  
 * 描述：程序启动入口  
 *  
 * @author hamhuo  
 * @version 1.0.0  
 */@SpringBootApplication  
@EnableBinding(Sink.class)  
@EnableDiscoveryClient  
public class WsApplication {  
    public static void main(String[] args) {  
        SpringApplication.run(WsApplication.class, args);  
    }  
}

@Component  
@ServerEndpoint("/chat")  
public class ChatServer {
/**  
 * 保存连接对象, 连接池, 放用户session和key
 */  
private static final ConcurrentHashMap<String, Session> SESSION_POOL = new ConcurrentHashMap<>();

@OnOpen  
public void onOpen(Session session) 

@OnMessage  
public String onMessage(String text, Session session) 

@OnClose  
public void onClose(Session session)

@OnError  
public void onError(Session session, Throwable throwable)

}

/**  
 * @program: demo  
 * @description: 服务端点配置  
 * @author: hamhuo  
 **/  
@Configuration  
@EnableWebSocket  
public class WebSocketConfig {  
    @Bean  
    public ServerEndpointExporter serverEndpoint() {  
        return new ServerEndpointExporter();  
    }  
}

@OnOpen  
public void onOpen(Session session) throws IOException {  
    //判断客户端对象是否存在  
    //排除重复的连接对象  
    if (SESSION_POOL.containsKey(session.getQueryString())) {  
        CloseReason reason = new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT, "ID冲突，连接拒绝");  
        session.getUserProperties().put("reason", reason);  
        session.close();  
        return;  
    }  
    //将客户端记录到Map中  
    SESSION_POOL.put(session.getQueryString(), session);  
    System.out.println("客户端（" + session.getQueryString() + "）：开启连接");  
}

@OnMessage  
public String onMessage(String text, Session session) throws IOException {  
    //解析消息 => ID::消息内容  
    String[] msgArr = text.split("::", 2);  
    //群发消息 ID=all表示群发  
    if ("all".equalsIgnoreCase(msgArr[0])) {  
        for (Session one : SESSION_POOL.values()) {  
            //排除自己  
            if (session == one) {  
                continue;  
            }  
            //发送消息  
            one.getBasicRemote().sendText(msgArr[1]);  
        }  
    }  
    //指定发送人, 不是all的情况  
    else {  
        Session target = SESSION_POOL.get(msgArr[0]);  
        if (target != null) {  
            target.getBasicRemote().sendText(msgArr[1]);  
        }  
    }  
    return session.getQueryString() + ":消息发送成功";  
}

@OnClose  
public void onClose(Session session) {  
    //处理拒绝连接session关闭对象, 不从池子里移除  
    Object obj = session.getUserProperties().get("reason");  
    if (obj instanceof CloseReason) {  
        CloseReason reason = (CloseReason) obj;  
        if (reason.getCloseCode() == CloseReason.CloseCodes.CANNOT_ACCEPT) {  
            System.out.println("拒绝客户端（" + session.getQueryString() + "）：关闭连接");  
            return;  
        }  
    }  
    //将session对象从map中移除,正常关闭  
    SESSION_POOL.remove(session.getQueryString());  
    System.out.println("客户端（" + session.getQueryString() + "）：关闭连接");  
}

<!DOCTYPE html>  
<html lang="en">  
<head>  
    <meta charset="UTF-8">  
    <title>聊天室客户端</title>  
</head>  
<body>  
<h1>Chat Client</h1>  
<div>  
    <input id="clientId" placeholder="输入ID" value="1">  
    <input onclick="init()" value="连接服务器" type="button"><br><br>  
    <input id="receiverId" placeholder="输入接收人ID" value="all"><br><br>  
    <textarea id="message" style="margin: 0; height: 197px; width: 362px;"  
              placeholder="发送消息内容"></textarea><br>  
    <input onclick="send()" value="发送消息" type="button">  
    <input onclick="closeConnect()" value="关闭连接" type="button">  
</div>  
<div id="output"></div>  
<script type="text/javascript" language="JavaScript">  
    //屏幕回显输出方法  
    function writeToScreen(message) {  
        let pre = document.createElement("p");  
        pre.style.wordWrap = "break-word";  
        pre.innerHTML = message;  
        document.getElementById("output").appendChild(pre);  
    }  
  
    //初始化websocket  
    let echo_websocket;  
    function init() {  
        let clientId = document.getElementById("clientId").value;  
        /* 这里端口是写死的 */        let wsUri = "ws://localhost:10800/chat?" + clientId;  
        writeToScreen("连接到" + wsUri);  
        //1.创建WebSocket客户端对象  
        echo_websocket = new WebSocket(wsUri);  
        //2.开门握手完成回调  
        echo_websocket.onopen = function (evt) {  
            console.log(evt);  
            writeToScreen("连接打开成功 !");  
        };  
        //3.监听服务端的消息  
        echo_websocket.onmessage = function (evt) {  
            writeToScreen("接收服务端消息:<br> " + evt.data);  
        };  
        //4.如果连接中断  
        echo_websocket.onerror = function (evt) {  
            writeToScreen('<span style="color: red;">ERROR:'+evt.data+'</span>');  
            //关闭连接  
            closeConnect();  
        };  
        //5.注册close事件  
        echo_websocket.onclose = function(evt){  
            writeToScreen('<span style="color: green;">INFO：关闭连接</span> ');  
            if(evt.reason){  
                writeToScreen  
                (`<span style="color: red;">错误信息：${evt.reason}</span> `);  
            }  
        }  
    }  
  
    //6.向服务发送消息  
    function send() {  
        let message = document.getElementById("message").value;  
        let receiver = document.getElementById("receiverId").value;  
        echo_websocket.send(receiver + "::" + message);  
        writeToScreen("发送消息: " + message);  
    }  
  
    //7.如果不需要通讯，那么关闭连接  
    function closeConnect() {  
        echo_websocket.close();  
    }  
</script>  
</body>  
</html>

//6.向服务发送消息  
    function send() {  
        let message = document.getElementById("message").value;  
        let receiver = document.getElementById("receiverId").value;  
        echo_websocket.send(receiver + "::" + message);  
        writeToScreen("发送消息: " + message);  
    }