spring boot websocket 集群环境处理

发布于:2021-10-26 02:48:25

? ? ? ? 开发中迎来一个功能重构,需要在后台系统给指定用户发送whatsApp消息。验证用户关系,用户回复后进行其他提问。原先开发这个功能负责人是基于轮询实现,看着频繁刷mongodb的日志心里不爽,决定尝试使用websocket 进行推送。


? ? ? ? 通常实现websocket 方式有几种原生注解,spring 封装,及netty实现。?具体实现可以参考下此文https://www.cnblogs.com/kiwifly/p/11729304.html


? ? ? ?相信简单实现通信对工作1-2年之后的人并不是什么难事,哈哈 不就是使用api吗。具体通信代码就不贴了,没啥意思。主要是想记录一下,开发过程遇到的问题。了解websocket的相信都知道websocket中的session无法持久化,不像http中的session能持久化到redis之类。这对集群环境显然不是很美好,例如发送whatsApp消息后台操作人员连接到A服务器,给用户发送消息。A服务器收到发送消息请求,马上调用第三方消息接口发送。由于用户回复消息不是实时的,只能是提供一个回调地址给第三方。当今项目架构想必单机的情况基本不存在了吧,若是第三方回调A服务器上,则通过websocket正常推送。若回调到B服务,那么会话就不存在了。怎么解决呢?有人会说可以用nginx把用户请求都落到一个节点上,这样就使得程序失去负载均衡的意义了。反过来想想既然消息落到某台机器,是不是可以把消息分发到其他机器呢?广播一下消息即可。dubbo redis mq 都可以做相应的事。


? ? ?鉴于我们项目使用了RabbitMq,决定使用RabbitMq方式广播消息。用过RabbitMq肯定知道有种FanoutExchange模式的交换机,问题又来了FanoutExchange只会把消息广播到不同queue里。这时候心里又嘀咕了,同一套代码跟配置都是一样的。怎么搞呢?有没有法子呀,答案肯定是有的咯。可以声明queue的时候获取本机ip及进程id,这样就能产生不同的queue的名字了。动态创建queue及绑定交换机即可贴出部分代码


?


@Bean("webSocketFanoutExchange")
FanoutExchange webSocketFanoutExchange() {
return new FanoutExchange("webSocketFanoutExchange");
}

@Bean("webSocketQueue")
public Queue webSocketQueue() {
String name = "message.websocket.";
try {
String ip = InetAddress.getLocalHost().getHostAddress();
RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
final String info = runtime.getName();
final int index = info.indexOf("@");
if (index != -1) {
name = name + ip + "_" + info.substring(0, index);
}
} catch (UnknownHostException e) {
log.warn("获取本地ip失败", e);
name = name + UUID.randomUUID().toString().replace("-", "");
}
name = name + ".queue";
return new Queue(name, false, false, true, null);
}

@Bean("bindingWebSocketExchange")
Binding binding WebSocketExchange(@Qualifier("webSocketQueue") Queue webSocketQueue,
@Qualifier("webSocketFanoutExchange") FanoutExchange webSocketFanoutExchange) {
return BindingBuilder.bind(webSocketQueue).to(webSocketFanoutExchange);
}

@Bean(" WebSocketMessageContainer")
public SimpleMessageListenerContainer WebSocketMessageContainer(ConnectionFactory connectionFactory,
@Qualifier("webSocketQueue") Queue webSocketQueue,
WebSocketMqListener webSocketMqListener) throws AmqpException {
//WebSocketMqListener 你的消费处理
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.addQueues(webSocketQueue);
container.setMessageListener(webSocketMqListener);//监听处理类
return container;
}


?

相关推荐

最新更新

猜你喜欢